diff --git a/CHANGELOG.md b/CHANGELOG.md index 9de1a07..f554e3c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,11 @@ All notable changes to pcapper will be documented in this file. This project follows [Semantic Versioning](https://semver.org/). +## Unreleased + +### Security +- **Recovered secrets and carved artifacts are now written owner-only (`0600`, `0700` for directories).** pcapper deliberately does not redact what it recovers (`reporting._redact_secret` is a documented no-op), so reports, exports, the run log, carved artifacts, extracted files and decrypted streams routinely contain cleartext credentials, LDAP binds, SNMP community strings, session tokens and malware samples. None of those write paths restricted permissions, so under a default umask they landed at `0644` (case and output directories at `0755`) — on a shared analysis host every other local account could read recovered passwords out of another analyst's case directory, and world-readable evidence is harder to defend on chain-of-custody grounds. Tightened at every write path: reports and JSON export (`utils.safe_write_text`), CSV and `_hosts` CSV, the SQLite export (tightened immediately after `sqlite3.connect()`, so the database is never briefly readable while being populated), the `--log-file` run log, case directories, export output directories, carved artifacts, extracted file artifacts, and decrypted TLS/SSH streams. New `utils.restrict_permissions()` / `utils.restrict_dir_permissions()` helpers are best-effort by design: POSIX mode bits are largely a no-op on Windows, and a failure to tighten permissions never aborts an analysis run. + ## 2.1.0 — 2026-07-11 ### Changed diff --git a/pcapper/carving.py b/pcapper/carving.py index 77b2bbc..cbcb433 100644 --- a/pcapper/carving.py +++ b/pcapper/carving.py @@ -8,7 +8,12 @@ from pathlib import Path from .pcap_cache import get_reader -from .utils import detect_file_type_bytes, extract_packet_endpoints +from .utils import ( + detect_file_type_bytes, + extract_packet_endpoints, + restrict_dir_permissions, + restrict_permissions, +) try: from scapy.layers.inet import IP, TCP # type: ignore @@ -270,7 +275,7 @@ def analyze_carving( detections: list[dict[str, object]] = [] if output_dir: - output_dir.mkdir(parents=True, exist_ok=True) + restrict_dir_permissions(output_dir) for stream_key in stream_stats.keys(): src, sport, dst, dport = stream_key @@ -297,6 +302,7 @@ def analyze_carving( out_path = output_dir / filename try: out_path.write_bytes(blob) + restrict_permissions(out_path) extracted.append(out_path) except Exception as exc: errors.append(f"Carve write error: {exc}") diff --git a/pcapper/cli.py b/pcapper/cli.py index f1437e2..96b4f4a 100644 --- a/pcapper/cli.py +++ b/pcapper/cli.py @@ -294,7 +294,13 @@ ) from .tls import analyze_tls from .udp import analyze_udp -from .utils import hexdump, parse_time_arg, safe_write_text +from .utils import ( + hexdump, + parse_time_arg, + restrict_dir_permissions, + restrict_permissions, + safe_write_text, +) from .vlan import analyze_vlans from .vnc import analyze_vnc, merge_vnc_summaries from .vpn import analyze_vpn, merge_vpn_summaries @@ -751,8 +757,11 @@ def _log_event(log_config: LogConfig | None, event: str, **fields: Any) -> None: ) try: if log_config.path: - log_config.path.parent.mkdir(parents=True, exist_ok=True) + restrict_dir_permissions(log_config.path.parent) + existed = log_config.path.exists() with log_config.path.open("a", encoding="utf-8") as handle: + if not existed: + restrict_permissions(log_config.path) handle.write(f"{line}\n") elif log_config.stream: log_config.stream.write(f"{line}\n") @@ -2243,7 +2252,7 @@ def _analyze_paths( suricata_scans = 0 if case_dir: - case_dir.mkdir(parents=True, exist_ok=True) + restrict_dir_permissions(case_dir) def _render_packet(path: Path, index: int, packets: list[object] | None) -> None: if index <= 0: @@ -2312,11 +2321,11 @@ def _resolve_export_path(base: Path, pcap_path: Path, suffix: str) -> Path: if base.suffix: if multi_export: out_dir = base.parent / base.stem - out_dir.mkdir(parents=True, exist_ok=True) + restrict_dir_permissions(out_dir) return out_dir / f"{pcap_path.stem}{base.suffix}" return base out_dir = base - out_dir.mkdir(parents=True, exist_ok=True) + restrict_dir_permissions(out_dir) return out_dir / f"{pcap_path.stem}.{suffix}" def _resolve_misc_path(path: Path) -> Path: @@ -5068,7 +5077,7 @@ def _write_case_metadata(end_time: datetime) -> None: if not case_dir: return try: - case_dir.mkdir(parents=True, exist_ok=True) + restrict_dir_permissions(case_dir) except Exception: pass case_id = getattr(args, "case_id", None) or getattr(args, "case_name", None) diff --git a/pcapper/decryption.py b/pcapper/decryption.py index 8b0a29d..a5e7c71 100644 --- a/pcapper/decryption.py +++ b/pcapper/decryption.py @@ -6,6 +6,8 @@ from dataclasses import dataclass, field from pathlib import Path +from .utils import restrict_dir_permissions, restrict_permissions + @dataclass(frozen=True) class DecryptConfig: @@ -144,7 +146,7 @@ def decrypt_tls( if not streams: notes.append("No TLS streams detected for decryption.") - output_dir.mkdir(parents=True, exist_ok=True) + restrict_dir_permissions(output_dir) count = 0 for stream_id in streams: if limit and count >= limit: @@ -159,6 +161,7 @@ def decrypt_tls( errors.append(stderr.strip()) if stdout: filename.write_text(stdout, encoding="utf-8", errors="ignore") + restrict_permissions(filename) outputs.append(filename) count += 1 @@ -200,7 +203,7 @@ def decrypt_ssh( if not streams: notes.append("No SSH streams detected for decryption.") - output_dir.mkdir(parents=True, exist_ok=True) + restrict_dir_permissions(output_dir) count = 0 for stream_id in streams: if limit and count >= limit: @@ -213,6 +216,7 @@ def decrypt_ssh( errors.append(stderr.strip()) if stdout: filename.write_text(stdout, encoding="utf-8", errors="ignore") + restrict_permissions(filename) outputs.append(filename) count += 1 diff --git a/pcapper/exporting.py b/pcapper/exporting.py index 41376a0..1c6b30e 100644 --- a/pcapper/exporting.py +++ b/pcapper/exporting.py @@ -7,7 +7,12 @@ from pathlib import Path from typing import Any, Iterable -from .utils import safe_write_text, to_serializable +from .utils import ( + restrict_dir_permissions, + restrict_permissions, + safe_write_text, + to_serializable, +) @dataclass @@ -17,7 +22,7 @@ class ExportBundle: def _ensure_parent(path: Path) -> None: - path.parent.mkdir(parents=True, exist_ok=True) + restrict_dir_permissions(path.parent) def _detections_of(summary: Any) -> list[dict[str, Any]]: @@ -185,6 +190,7 @@ def export_csv(bundle: ExportBundle, output_path: Path) -> None: else: fieldnames = sorted({key for row in rows for key in row.keys()}) with output_path.open("w", newline="", encoding="utf-8") as handle: + restrict_permissions(output_path) writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() for row in rows: @@ -197,6 +203,7 @@ def export_csv(bundle: ExportBundle, output_path: Path) -> None: _ensure_parent(host_path) host_fields = sorted({key for row in host_rows for key in row.keys()}) with host_path.open("w", newline="", encoding="utf-8") as handle: + restrict_permissions(host_path) writer = csv.DictWriter(handle, fieldnames=host_fields) writer.writeheader() for row in host_rows: @@ -210,6 +217,7 @@ def export_sqlite(bundle: ExportBundle, output_path: Path) -> None: raise ValueError(f"SQLite export path is a directory: {output_path}") output_path.unlink() conn = sqlite3.connect(str(output_path)) + restrict_permissions(output_path) cur = conn.cursor() cur.execute("CREATE TABLE detections (module TEXT, data TEXT)") cur.execute("CREATE TABLE artifacts (module TEXT, data TEXT)") diff --git a/pcapper/files.py b/pcapper/files.py index c1e9169..5f2bf10 100644 --- a/pcapper/files.py +++ b/pcapper/files.py @@ -41,7 +41,15 @@ from .cip import CIP_SERVICE_NAMES from .nfs import analyze_nfs from .pcap_cache import get_reader -from .utils import detect_file_type_bytes, extract_packet_endpoints, memoize_analysis, packet_length, safe_float +from .utils import ( + detect_file_type_bytes, + extract_packet_endpoints, + memoize_analysis, + packet_length, + restrict_dir_permissions, + restrict_permissions, + safe_float, +) try: from scapy.layers.inet import IP, TCP, UDP @@ -4324,7 +4332,7 @@ def _in_scope(src_ip: str, dst_ip: str) -> bool: if extract_name: out_root = output_dir or Path.cwd() / "files" - out_root.mkdir(parents=True, exist_ok=True) + restrict_dir_permissions(out_root) search = extract_name.lower() for art in artifacts: if art.payload and search in art.filename.lower(): @@ -4333,6 +4341,7 @@ def _in_scope(src_ip: str, dst_ip: str) -> bool: continue try: out_p.write_bytes(art.payload) + restrict_permissions(out_p) extracted_paths.append(out_p) except Exception: continue diff --git a/pcapper/utils.py b/pcapper/utils.py index 57e9f74..7a1f60c 100644 --- a/pcapper/utils.py +++ b/pcapper/utils.py @@ -103,6 +103,38 @@ def safe_read_text( return "" +def restrict_permissions(path: Path) -> None: + """Restrict a written artifact to owner-only access (0600, 0700 for a dir). + + pcapper deliberately does not redact recovered secrets (see + reporting._redact_secret), so its outputs routinely contain cleartext + credentials, session tokens and malware samples. Under a default umask + those land world-readable, exposing them to every other local account on a + shared analysis host. + + Best effort by design: POSIX mode bits are largely a no-op on Windows, and + failing to tighten permissions must never abort an analysis run. + """ + try: + path.chmod(0o700 if path.is_dir() else 0o600) + except Exception: + pass + + +def restrict_dir_permissions(path: Path) -> None: + """``mkdir(parents=True, exist_ok=True)`` plus an owner-only chmod. + + Only a directory this call actually creates is tightened; one the analyst + already had keeps the mode they gave it, so pointing an output flag at an + existing shared directory does not silently re-permission it. mkdir errors + propagate exactly as a bare ``mkdir`` would — only the chmod is best effort. + """ + created = not path.exists() + path.mkdir(parents=True, exist_ok=True) + if created: + restrict_permissions(path) + + def safe_write_text( path: Path, text: str, @@ -113,6 +145,7 @@ def safe_write_text( ) -> None: try: path.write_text(text, encoding=encoding) + restrict_permissions(path) except Exception as exc: record_error(errors_list, context, exc) if errors_list is None: