chore: resolve lint issues in installer and reporting modules

install_dependencies.py

@@ -1,23 +1,26 @@
 #!/usr/bin/env python3
-from __future__ import annotations
 
 """Helper script to ensure ReconScript runtime dependencies are installed."""
 
+from __future__ import annotations
+
 import argparse
 import hashlib
 import importlib
+import logging
 import subprocess
 import sys
+from collections.abc import Iterable
 from pathlib import Path
-from typing import Dict, Iterable
 
 ROOT = Path(__file__).resolve().parent
 REQUIREMENTS_FILE = ROOT / "requirements.txt"
 MARKER_IN_VENV = ROOT / ".venv" / ".requirements-hash"
 MARKER_FALLBACK = ROOT / ".requirements-hash"
 
 # Mapping of requirement name to import name so we can detect missing modules quickly.
-REQUIREMENT_IMPORTS: Dict[str, str] = {
+LOGGER = logging.getLogger(__name__)
+REQUIREMENT_IMPORTS: dict[str, str] = {
     "requests": "requests",
     "urllib3": "urllib3",
     "jinja2": "jinja2",
@@ -46,6 +49,7 @@ def create_console():  # type: ignore[return-value]
 
         return Console(highlight=False)
     except Exception:
+        LOGGER.exception("Rich console unavailable; falling back to plain console.")
 
         class _PlainConsole:
             def print(self, *values: object, sep: str = " ", end: str = "\n") -> None:
@@ -73,11 +77,12 @@ def _marker_path() -> Path:
     return MARKER_FALLBACK
 
 
-def _missing_modules(modules: Dict[str, str]) -> Iterable[str]:
+def _missing_modules(modules: dict[str, str]) -> Iterable[str]:
     for requirement, module_name in modules.items():
         try:
             importlib.import_module(module_name)
         except Exception:
+            LOGGER.exception("Failed to import module '%s'", module_name)
             yield requirement
 
 
@@ -123,7 +128,7 @@ def install_dependencies(
             str(requirements),
         ]
         output.print("Resolving Python requirements (this may take a moment)…")
-        result = subprocess.run(command, capture_output=True, text=True)
+        result = subprocess.run(command, capture_output=True, text=True)  # noqa: S603
         if result.returncode != 0:
             combined_output = "\n".join(
                 part for part in (result.stdout, result.stderr) if part
@@ -172,7 +177,7 @@ def main(argv: Iterable[str] | None = None) -> None:
         )
     except Exception as exc:  # pragma: no cover - invoked from CLI
         console.print(f"[red]Failed to install dependencies: {exc}[/red]")
-        raise SystemExit(1)
+        raise SystemExit(1) from exc
 
 
 if __name__ == "__main__":  # pragma: no cover - CLI entrypoint

reconscript/consent.py

@@ -4,26 +4,20 @@
 
 import base64
 import json
+import logging
 import os
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import Any
 
 from nacl import exceptions as nacl_exceptions
 from nacl.signing import SigningKey, VerifyKey
 
 SCHEMA_PATH = Path(__file__).resolve().parent / "schemas" / "scope_manifest.v1.json"
-<<<<<<< HEAD
 DEV_KEYS_DIR = Path(__file__).resolve().parents[1] / "keys"
-=======
-DEFAULT_PUBLIC_KEY = Path(
-    os.environ.get("CONSENT_PUBLIC_KEY_PATH", "keys/dev_ed25519.pub")
-)
-DEFAULT_PRIVATE_KEY = Path(
-    os.environ.get("REPORT_SIGNING_KEY_PATH", "keys/dev_ed25519.priv")
-)
->>>>>>> 74be2f0 (style: fix reporters.py syntax and apply Black formatting)
+
+LOGGER = logging.getLogger(__name__)
 
 
 class ConsentError(ValueError):
@@ -70,7 +64,7 @@ def _parse_iso8601(value: str) -> datetime:
     return parsed.astimezone(timezone.utc)
 
 
-def _canonical_json(data: Dict[str, Any]) -> bytes:
+def _canonical_json(data: dict[str, Any]) -> bytes:
     return json.dumps(data, separators=(",", ":"), sort_keys=True).encode("utf-8")
 
 
@@ -81,16 +75,18 @@ def _allow_dev_secrets() -> bool:
 def _guard_key_path(path: Path, *, env_var: str) -> Path:
     resolved = path.expanduser().resolve()
     if not resolved.exists():
-        raise ConsentError(f"{env_var} must point to an existing file (got {resolved}).")
-    try:
-        if DEV_KEYS_DIR in resolved.parents and not _allow_dev_secrets():
-            raise ConsentError(
-                f"{env_var} references developer sample keys. Provide production keys or set ALLOW_DEV_SECRETS=true for local testing."
-            )
-    except ConsentError:
-        raise
-    except Exception:
-        pass
+        raise ConsentError(
+            f"{env_var} must point to an existing file (got {resolved})."
+        )
+    if DEV_KEYS_DIR in resolved.parents and not _allow_dev_secrets():
+        LOGGER.warning(
+            "Rejected developer key reference for %s due to ALLOW_DEV_SECRETS=false.",
+            env_var,
+        )
+        raise ConsentError(
+            f"{env_var} references developer sample keys. Provide production keys "
+            "or set ALLOW_DEV_SECRETS=true for local testing."
+        )
     return resolved
 
 
@@ -112,12 +108,14 @@ def _load_private_key(path: Path) -> SigningKey:
     return SigningKey(raw)
 
 
-def _resolve_key_path(provided: Optional[Path], env_var: str) -> Path:
+def _resolve_key_path(provided: Path | None, env_var: str) -> Path:
     if provided is not None:
         return _guard_key_path(provided, env_var=env_var)
     env_value = os.environ.get(env_var)
     if not env_value:
-        raise ConsentError(f"{env_var} must be set to the path of the authorised signing key.")
+        raise ConsentError(
+            f"{env_var} must be set to the path of the authorised signing key."
+        )
     return _guard_key_path(Path(env_value), env_var=env_var)
 
 
@@ -169,15 +167,10 @@ def load_manifest(path: Path | str) -> ConsentManifest:
     return manifest
 
 
-<<<<<<< HEAD
-def validate_manifest(manifest: ConsentManifest, *, public_key_path: Optional[Path] = None) -> ConsentValidationResult:
-    key_path = _resolve_key_path(public_key_path, "CONSENT_PUBLIC_KEY_PATH")
-=======
 def validate_manifest(
-    manifest: ConsentManifest, *, public_key_path: Optional[Path] = None
+    manifest: ConsentManifest, *, public_key_path: Path | None = None
 ) -> ConsentValidationResult:
-    key_path = public_key_path or DEFAULT_PUBLIC_KEY
->>>>>>> 74be2f0 (style: fix reporters.py syntax and apply Black formatting)
+    key_path = _resolve_key_path(public_key_path, "CONSENT_PUBLIC_KEY_PATH")
     verify_key = _load_public_key(key_path)
 
     body = {
@@ -214,15 +207,10 @@ def validate_manifest(
     return ConsentValidationResult(manifest=manifest, verify_key=verify_key)
 
 
-<<<<<<< HEAD
-def sign_report_hash(report_hash: str, *, private_key_path: Optional[Path] = None) -> bytes:
-    key_path = _resolve_key_path(private_key_path, "REPORT_SIGNING_KEY_PATH")
-=======
 def sign_report_hash(
-    report_hash: str, *, private_key_path: Optional[Path] = None
+    report_hash: str, *, private_key_path: Path | None = None
 ) -> bytes:
-    key_path = private_key_path or DEFAULT_PRIVATE_KEY
->>>>>>> 74be2f0 (style: fix reporters.py syntax and apply Black formatting)
+    key_path = _resolve_key_path(private_key_path, "REPORT_SIGNING_KEY_PATH")
     signing_key = _load_private_key(key_path)
     message = report_hash.encode("utf-8")
     signed = signing_key.sign(message)

reconscript/report.py

@@ -4,46 +4,48 @@
 
 import hashlib
 import json
+import logging
 import os
 import time
 import uuid
 from contextlib import AbstractContextManager
 from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
-from typing import Dict, Optional
 
 from .consent import sign_report_hash
 
 RESULTS_DIR = Path(os.environ.get("RESULTS_DIR", "results")).expanduser()
 INDEX_FILE = RESULTS_DIR / "index.json"
 LOCK_PATH = RESULTS_DIR / ".index.lock"
 
+LOGGER = logging.getLogger(__name__)
+
 
 @dataclass(frozen=True)
 class ReportPaths:
     report_id: str
     base: Path
     report_file: Path
-    manifest_path: Optional[Path] = None
-    signature_path: Optional[Path] = None
+    manifest_path: Path | None = None
+    signature_path: Path | None = None
 
 
 class FileLock(AbstractContextManager):
     def __init__(self, path: Path, timeout: float = 10.0) -> None:
         self.path = path
         self.timeout = timeout
-        self._fd: Optional[int] = None
+        self._fd: int | None = None
 
-    def __enter__(self) -> "FileLock":
+    def __enter__(self) -> FileLock:
         deadline = time.monotonic() + self.timeout
         while True:
             try:
                 self._fd = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_RDWR)
                 break
-            except FileExistsError:
+            except FileExistsError as exc:
                 if time.monotonic() >= deadline:
-                    raise TimeoutError(f"Timed out acquiring lock {self.path}")
+                    raise TimeoutError(f"Timed out acquiring lock {self.path}") from exc
                 time.sleep(0.1)
         return self
 
@@ -53,7 +55,7 @@ def __exit__(self, exc_type, exc, tb) -> None:
         try:
             self.path.unlink(missing_ok=True)
         except OSError:
-            pass
+            LOGGER.exception("Failed to remove lock file %s", self.path)
 
 
 def ensure_results_dir() -> Path:
@@ -86,16 +88,16 @@ def default_output_path(
 
 
 def embed_runtime_metadata(
-    report: Dict[str, object],
+    report: dict[str, object],
     started_at: datetime,
-    completed_at: Optional[datetime] = None,
-    duration: Optional[float] = None,
-) -> Dict[str, object]:
+    completed_at: datetime | None = None,
+    duration: float | None = None,
+) -> dict[str, object]:
     start_iso = started_at.replace(microsecond=0).isoformat() + "Z"
     report["timestamp"] = report.get("timestamp", start_iso)
     report["started_at"] = report.get("started_at", start_iso)
 
-    runtime: Dict[str, object] = {
+    runtime: dict[str, object] = {
         "started_at": report["started_at"],
     }
 
@@ -113,12 +115,12 @@ def embed_runtime_metadata(
     return report
 
 
-def _canonical_report_bytes(report: Dict[str, object]) -> bytes:
+def _canonical_report_bytes(report: dict[str, object]) -> bytes:
     payload = {key: value for key, value in report.items() if key != "report_hash"}
     return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
 
 
-def compute_report_hash(report: Dict[str, object]) -> str:
+def compute_report_hash(report: dict[str, object]) -> str:
     digest = hashlib.sha256(_canonical_report_bytes(report)).hexdigest()
     return digest
 
@@ -130,6 +132,7 @@ def _load_index() -> list[dict[str, object]]:
         content = INDEX_FILE.read_text(encoding="utf-8")
         return json.loads(content)
     except json.JSONDecodeError:
+        LOGGER.exception("Failed to parse index file %s", INDEX_FILE)
         return []
 
 
@@ -140,7 +143,7 @@ def _write_index(entries: list[dict[str, object]]) -> None:
     )
 
 
-def _index_entry(report_id: str, report: Dict[str, object]) -> dict[str, object]:
+def _index_entry(report_id: str, report: dict[str, object]) -> dict[str, object]:
     return {
         "report_id": report_id,
         "target": report.get("target"),
@@ -153,9 +156,9 @@ def _index_entry(report_id: str, report: Dict[str, object]) -> dict[str, object]
 
 
 def persist_report(
-    report: Dict[str, object],
+    report: dict[str, object],
     *,
-    consent_source: Optional[Path] = None,
+    consent_source: Path | None = None,
     sign: bool = False,
 ) -> ReportPaths:
     ensure_results_dir()
@@ -171,14 +174,14 @@ def persist_report(
         json.dumps(report, indent=2, sort_keys=True), encoding="utf-8"
     )
 
-    manifest_path: Optional[Path] = None
+    manifest_path: Path | None = None
     if consent_source:
         manifest_dir = report_dir / "consent"
         manifest_dir.mkdir(parents=True, exist_ok=True)
         manifest_path = manifest_dir / "manifest.json"
         manifest_path.write_bytes(Path(consent_source).read_bytes())
 
-    signature_path: Optional[Path] = None
+    signature_path: Path | None = None
     if sign:
         signature_bytes = sign_report_hash(report_hash)
         signature_path = report_dir / "report.sig"