diff --git a/install_dependencies.py b/install_dependencies.py index cd7f66e..c9d2728 100644 --- a/install_dependencies.py +++ b/install_dependencies.py @@ -1,15 +1,17 @@ #!/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" @@ -17,7 +19,8 @@ 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 diff --git a/reconscript/consent.py b/reconscript/consent.py index ea2d2cb..a4e3d49 100644 --- a/reconscript/consent.py +++ b/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) diff --git a/reconscript/report.py b/reconscript/report.py index c107d32..1144008 100644 --- a/reconscript/report.py +++ b/reconscript/report.py @@ -4,6 +4,7 @@ import hashlib import json +import logging import os import time import uuid @@ -11,7 +12,6 @@ from dataclasses import dataclass from datetime import datetime from pathlib import Path -from typing import Dict, Optional from .consent import sign_report_hash @@ -19,31 +19,33 @@ 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" diff --git a/reconscript/reporters.py b/reconscript/reporters.py index 169d518..9971dd4 100644 --- a/reconscript/reporters.py +++ b/reconscript/reporters.py @@ -3,13 +3,15 @@ from __future__ import annotations # Modified by codex: 2024-05-08 - import html import json import logging +from collections.abc import Iterable, Sequence from datetime import datetime from pathlib import Path -from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple +from typing import Any + +from . import __version__ PACKAGE_DIR = Path(__file__).resolve().parent PROJECT_TEMPLATES = PACKAGE_DIR.parent / "templates" @@ -25,7 +27,7 @@ Environment = None # type: ignore[assignment] FileSystemLoader = None # type: ignore[assignment] select_autoescape = None # type: ignore[assignment] - _JINJA_ENV: Optional[Environment] = None + _JINJA_ENV: Environment | None = None else: # pragma: no cover - exercised when Jinja2 installed loader_paths = [str(path) for path in TEMPLATE_DIRS] or [str(PACKAGE_TEMPLATES)] _JINJA_ENV = Environment( @@ -33,19 +35,17 @@ autoescape=select_autoescape(["html", "xml"]), ) -from . import __version__ - LOGGER = logging.getLogger(__name__) PDF_FALLBACK_MESSAGE = "PDF dependencies not found — served HTML version instead." -def render_json(data: Dict[str, object]) -> str: +def render_json(data: dict[str, object]) -> str: """Return a canonical JSON representation of the scan results.""" return json.dumps(data, indent=2, sort_keys=True) -def render_markdown(data: Dict[str, object]) -> str: +def render_markdown(data: dict[str, object]) -> str: """Render the report data in Markdown format.""" if _JINJA_ENV is not None and _JINJA_ENV.loader is not None: @@ -53,20 +53,16 @@ def render_markdown(data: Dict[str, object]) -> str: template = _JINJA_ENV.get_template("report.md.j2") return template.render(data=data) except Exception: - pass - - target = data.get("target", "unknown") - timestamp = data.get("timestamp", datetime.utcnow().isoformat() + "Z") - hostname = data.get("hostname") or "N/A" - ports = _format_list(data.get("ports", [])) - open_ports = _format_list(data.get("open_ports", [])) - findings = data.get("findings", []) - recommendations = _build_recommendations(findings) + LOGGER.exception( + "Markdown template rendering failed; using fallback renderer." + ) - lines: List[str] = [] + context = _build_markdown_context(data) + lines = _render_markdown_sections(context) + return "\n".join(lines) + "\n" -def render_html(data: Dict[str, object], out_path: Path) -> Path: +def render_html(data: dict[str, object], out_path: Path) -> Path: """Render the report data as HTML and persist it to ``out_path``.""" if not isinstance(out_path, Path): @@ -80,7 +76,7 @@ def render_html(data: Dict[str, object], out_path: Path) -> Path: return out_path.resolve() -def generate_pdf(html_path: Path, pdf_path: Path) -> Tuple[Path, bool]: +def generate_pdf(html_path: Path, pdf_path: Path) -> tuple[Path, bool]: """Generate a PDF from ``html_path`` and return the written file.""" if not isinstance(html_path, Path): @@ -114,8 +110,8 @@ def generate_pdf(html_path: Path, pdf_path: Path) -> Tuple[Path, bool]: def write_report( - data: Dict[str, object], outfile: Path, format: str -) -> Tuple[Path, str]: + data: dict[str, object], outfile: Path, format: str +) -> tuple[Path, str]: """Write ``data`` to ``outfile`` using ``format`` and return the resulting path.""" if not isinstance(outfile, Path): @@ -152,8 +148,8 @@ def _format_list(values: Iterable[object]) -> str: return ", ".join(str(item) for item in items) -def _build_recommendations(findings: Sequence[Dict[str, object]]) -> List[str]: - suggestions: List[str] = [] +def _build_recommendations(findings: Sequence[dict[str, object]]) -> list[str]: + suggestions: list[str] = [] issues = {item.get("issue") for item in findings} if "missing_security_headers" in issues: @@ -175,24 +171,25 @@ def _build_recommendations(findings: Sequence[Dict[str, object]]) -> List[str]: return suggestions -def _build_markdown_context(data: Dict[str, object]) -> Dict[str, Any]: +def _build_markdown_context(data: dict[str, object]) -> dict[str, Any]: findings = data.get("findings", []) + safe_findings = findings if isinstance(findings, Sequence) else [] context = { "target": data.get("target", "unknown"), "timestamp": data.get("timestamp", datetime.utcnow().isoformat() + "Z"), "hostname": data.get("hostname") or "N/A", "ports": _format_list(data.get("ports", [])) or "None", "open_ports": _format_list(data.get("open_ports", [])) or "None detected", - "findings": findings if isinstance(findings, Sequence) else [], - "recommendations": _build_recommendations(findings if isinstance(findings, Sequence) else []), + "findings": safe_findings, + "recommendations": _build_recommendations(safe_findings), "version": data.get("version", __version__), "runtime": data.get("runtime", {}), } return context -def _render_markdown_sections(context: Dict[str, Any]) -> List[str]: - lines: List[str] = [ +def _render_markdown_sections(context: dict[str, Any]) -> list[str]: + lines: list[str] = [ f"# ReconScript Report — {context['target']}", "", f"*Generated:* {context['timestamp']}", @@ -207,7 +204,11 @@ def _render_markdown_sections(context: Dict[str, Any]) -> List[str]: if findings: for item in findings: port = item.get("port", "n/a") if isinstance(item, dict) else "n/a" - issue = item.get("issue", "observation") if isinstance(item, dict) else str(item) + issue = ( + item.get("issue", "observation") + if isinstance(item, dict) + else str(item) + ) lines.append(f"- Port `{port}` — `{issue}`") if isinstance(item, dict) and item.get("details") is not None: details = item["details"] @@ -242,7 +243,7 @@ def _render_markdown_sections(context: Dict[str, Any]) -> List[str]: return lines -def _build_template_context(data: Dict[str, object]) -> Dict[str, object]: +def _build_template_context(data: dict[str, object]) -> dict[str, object]: timestamp_raw = str(data.get("timestamp", datetime.utcnow().isoformat() + "Z")) findings = data.get("findings", []) findings_count = len(findings) if isinstance(findings, Sequence) else 0 @@ -272,7 +273,7 @@ def _build_template_context(data: Dict[str, object]) -> Dict[str, object]: } -def _summary_rows(data: Dict[str, object]) -> List[Tuple[str, str]]: +def _summary_rows(data: dict[str, object]) -> list[tuple[str, str]]: ports = _format_list(data.get("ports", [])) or "None" open_ports = _format_list(data.get("open_ports", [])) or "None detected" findings = data.get("findings", []) @@ -282,7 +283,7 @@ def _summary_rows(data: Dict[str, object]) -> List[Tuple[str, str]]: runtime.get("duration") if isinstance(runtime, dict) else data.get("duration") ) - rows: List[Tuple[str, str]] = [ + rows: list[tuple[str, str]] = [ ("Target", str(data.get("target", "unknown"))), ("Hostname", str(data.get("hostname") or "N/A")), ("Ports Scanned", ports), @@ -299,8 +300,8 @@ def _summary_rows(data: Dict[str, object]) -> List[Tuple[str, str]]: return rows -def _format_findings(findings: Sequence[Dict[str, object]]) -> List[Dict[str, object]]: - formatted: List[Dict[str, object]] = [] +def _format_findings(findings: Sequence[dict[str, object]]) -> list[dict[str, object]]: + formatted: list[dict[str, object]] = [] for item in findings or []: formatted.append( { @@ -312,8 +313,8 @@ def _format_findings(findings: Sequence[Dict[str, object]]) -> List[Dict[str, ob return formatted -def _metadata_entries(data: Dict[str, object], timestamp: str) -> List[Tuple[str, str]]: - entries: List[Tuple[str, str]] = [ +def _metadata_entries(data: dict[str, object], timestamp: str) -> list[tuple[str, str]]: + entries: list[tuple[str, str]] = [ ("Tool Version", str(data.get("version", __version__))), ("Report Generated", timestamp), ] @@ -353,7 +354,7 @@ def _format_human_date(timestamp: str) -> str: return parsed.strftime("%d %B %Y %H:%M UTC") -def _render_html_document(context: Dict[str, object]) -> str: +def _render_html_document(context: dict[str, object]) -> str: if _JINJA_ENV is not None: try: template = _JINJA_ENV.get_template(HTML_TEMPLATE_NAME) @@ -363,7 +364,7 @@ def _render_html_document(context: Dict[str, object]) -> str: return _render_html_fallback(context) -def _render_html_fallback(context: Dict[str, object]) -> str: +def _render_html_fallback(context: dict[str, object]) -> str: summary_rows = context.get("summary_rows", []) findings = context.get("findings", []) recommendations = context.get("recommendations", []) @@ -423,23 +424,91 @@ def _render_html_fallback(context: Dict[str, object]) -> str: