Write recovered secrets and carved artifacts owner-only (0600/0700)

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

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}")

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)

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
 

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)")

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

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: