diff --git a/README.md b/README.md
index a94b70a..26c8c33 100644
--- a/README.md
+++ b/README.md
@@ -8,6 +8,16 @@ ReconScript is a read-only reconnaissance toolkit. It collects metadata from app
- **Structured reporting:** Export HTML, Markdown, JSON, or PDF artefacts, each tagged with metadata for downstream review.
- **Operational guardrails:** Environment variables, consent manifests, and placeholder keys highlight what must be configured before running against production targets.
+## Safe Default Reconnaissance Profile
+ReconScript now ships with a conservative, non-intrusive recon profile that limits every network touchpoint. The `ReconProfile` dataclass describes the caps in effect (TCP port counts, concurrency, DNS behaviour, HTTP enumeration depth) and is surfaced in every report under `metadata.profile`. The defaults explicitly disable exploitation, credentialed checks, and aggressive crawling.
+
+* Passive DNS: WHOIS summaries, certificate transparency notes, and hostname resolutions without issuing active probes beyond a single record sweep.
+* Active probing: one pass of rate-limited DNS queries, a single UDP sweep, and a TCP SYN scan capped to a curated port list.
+* HTTP coverage: GET/HEAD requests and small wordlist enumeration only for low/medium evidence levels, with strict per-request rate limiting.
+* Reporting: every finding includes timestamps, the exact pseudo-command executed, and a 400-character evidence snippet.
+
+You can create custom profiles from the CLI or web UI by instantiating `ReconProfile` and passing it to `run_recon()`. This allows regulated environments to tighten or loosen bounds while keeping safe defaults intact.
+
## Project Layout
The repository follows a conventional Python structure with documentation and automation assets kept alongside the source code. A more detailed component map lives in [`docs/DEPENDENCY_OVERVIEW.md`](docs/DEPENDENCY_OVERVIEW.md).
diff --git a/reconscript/core.py b/reconscript/core.py
index 01f70a0..bfd3e1a 100644
--- a/reconscript/core.py
+++ b/reconscript/core.py
@@ -1,4 +1,6 @@
-"""High-level orchestration for ReconScript scan execution."""
+# Authorized testing only — do not scan targets without explicit permission.
+# This tool is non-intrusive by default and will not perform exploitation or credentialed checks.
+"""High-level orchestration for ReconScript safe recon execution."""
from __future__ import annotations
@@ -16,19 +18,15 @@
)
from .report import embed_runtime_metadata
from .scanner import (
- DEFAULT_PORTS,
- REDACTION_KEYS,
- TOKEN_CAPACITY,
- TOKEN_RATE,
- ScanConfig,
- check_security_headers,
- fetch_robots,
- fetch_tls_certificate,
- generate_findings,
- http_probe_services,
+ DEFAULT_UDP_PORTS,
+ ReconProfile,
+ active_dns_sweep,
+ http_checks,
+ passive_dns_collection,
+ profile_for_evidence,
serialize_results,
- tcp_connect_scan,
- validate_port_list,
+ tcp_syn_scan,
+ udp_scan_pass,
)
from .scanner.throttle import TokenBucket
from .scope import ScopeError, ScopeValidation, ensure_within_allowlist, validate_target
@@ -43,13 +41,6 @@ class ReconError(RuntimeError):
"""Raised when a scan cannot proceed."""
-def _determine_redactions(extra: Iterable[str] | None) -> set[str]:
- redactions = set(REDACTION_KEYS)
- for item in extra or []:
- redactions.add(str(item).lower())
- return redactions
-
-
def _hostname_for_requests(scope: ScopeValidation, override: str | None) -> str:
if override:
return override
@@ -58,6 +49,11 @@ def _hostname_for_requests(scope: ScopeValidation, override: str | None) -> str:
return scope.resolved_ip or scope.target
+def _bucket(profile: ReconProfile, key: str) -> TokenBucket:
+ rate, burst = profile.rate_limits[key]
+ return TokenBucket(rate=rate, capacity=burst)
+
+
def run_recon(
*,
target: str,
@@ -70,6 +66,7 @@ def run_recon(
consent_manifest: ConsentManifest | None = None,
extra_redactions: Iterable[str] | None = None,
progress_callback: Callable[[str, float], None] | None = None,
+ profile: ReconProfile | None = None,
) -> dict[str, object]:
"""Execute the ReconScript workflow with safety controls."""
@@ -77,52 +74,69 @@ def run_recon(
if evidence_level not in EVIDENCE_LEVELS:
raise ReconError(f"Evidence level must be one of {sorted(EVIDENCE_LEVELS)}")
+ selected_profile = profile or profile_for_evidence(evidence_level)
+ LOGGER.info(
+ "Using recon profile %s",
+ selected_profile.name,
+ extra={"profile": selected_profile.as_dict()},
+ )
+
record_scan_started(target)
failure_reason: str | None = None
+ start_clock = time.perf_counter()
+
+ if enable_ipv6:
+ LOGGER.info(
+ "IPv6 scanning requested; safe profile limits probes to IPv4 endpoints.",
+ extra={"event": "scan.ipv6", "requested": True},
+ )
+ if extra_redactions:
+ LOGGER.info(
+ "Ignoring additional redactions under safe profile.",
+ extra={"event": "scan.redactions", "redactions": list(extra_redactions)},
+ )
try:
scope = validate_target(target, expected_ip=expected_ip)
ensure_within_allowlist(scope)
- candidates = list(ports) if ports else list(DEFAULT_PORTS)
- try:
- port_list = validate_port_list(candidates)
- except ReconError:
- failure_reason = "port_validation_failed"
- raise
-
- manifest = consent_manifest
-
- redactions = _determine_redactions(extra_redactions)
- bucket = TokenBucket(rate=TOKEN_RATE, capacity=TOKEN_CAPACITY)
+ limited_ports = selected_profile.limit_ports(ports)
+ udp_ports = list(DEFAULT_UDP_PORTS)
started_at = datetime.now(timezone.utc)
report: dict[str, object] = {
- "target": scope.target,
- "hostname": hostname,
- "ports": list(port_list),
- "version": __version__,
- "evidence_level": evidence_level,
+ "metadata": {
+ "target": scope.target,
+ "resolved_target": scope.resolved_ip or scope.target,
+ "hostname": hostname,
+ "version": __version__,
+ "evidence_level": evidence_level,
+ "profile": selected_profile.as_dict(),
+ "consent_present": bool(consent_manifest),
+ "scanned_tcp_ports": list(limited_ports),
+ "scanned_udp_ports": list(udp_ports),
+ },
+ "artifacts": {},
+ "findings": [],
}
- if manifest:
- report["consent_signed_by"] = manifest.signer_display
embed_runtime_metadata(report, started_at)
+ LOGGER.info(
+ "Consent manifest provided? %s",
+ bool(consent_manifest),
+ extra={"event": "scan.consent", "consent_present": bool(consent_manifest)},
+ )
+
if dry_run:
LOGGER.info(
"Dry-run requested; network operations skipped.",
extra={"event": "scan.dry_run", "target": scope.target},
)
- report.update(
- {
- "open_ports": [],
- "http_checks": {},
- "tls_cert": None,
- "robots": {"note": "dry-run"},
- "findings": [],
- }
- )
+ report["artifacts"] = {
+ "tcp": {"scanned_ports": list(limited_ports), "open_ports": []},
+ "udp": {"scanned_ports": udp_ports, "responses": {}},
+ }
embed_runtime_metadata(
report, started_at, completed_at=started_at, duration=0.0
)
@@ -130,72 +144,76 @@ def run_recon(
REPORT_LOGGER.info(serialize_results(report))
return report
- config = ScanConfig(
- target=scope.resolved_ip or scope.target,
- hostname=hostname,
- ports=port_list,
- enable_ipv6=enable_ipv6,
- evidence_level=evidence_level,
- redaction_keys=redactions,
- )
+ dns_bucket = _bucket(selected_profile, "dns")
+ tcp_bucket = _bucket(selected_profile, "tcp")
+ udp_bucket = _bucket(selected_profile, "udp")
+ http_bucket = _bucket(selected_profile, "http")
- http_host = _hostname_for_requests(scope, hostname)
- started_clock = time.perf_counter()
+ findings: list[dict[str, object]] = []
+ artifacts: dict[str, object] = {}
if progress_callback:
- progress_callback("Starting TCP connect scan", 0.1)
- try:
- open_ports = tcp_connect_scan(config, bucket)
- except Exception:
- failure_reason = "tcp_scan_failed"
- raise
- report["open_ports"] = open_ports
-
- http_results: dict[int, dict[str, object]] = {}
- if open_ports:
- if progress_callback:
- progress_callback("Collecting HTTP metadata", 0.4)
- try:
- http_results = http_probe_services(config, http_host, open_ports)
- except Exception:
- failure_reason = "http_probe_failed"
- raise
- report["http_checks"] = http_results
-
- tls_details = None
- if any(port in (443, 8443) for port in open_ports):
- if progress_callback:
- progress_callback("Retrieving TLS certificates", 0.6)
- tls_port = 443 if 443 in open_ports else 8443
- try:
- tls_details = fetch_tls_certificate(config, tls_port)
- except Exception:
- failure_reason = "tls_probe_failed"
- raise
- report["tls_cert"] = tls_details
+ progress_callback("Collecting passive DNS", 0.1)
+ passive_metadata, passive_findings = passive_dns_collection(
+ scope.target, selected_profile
+ )
+ artifacts["dns_passive"] = passive_metadata
+ findings.extend(passive_findings)
if progress_callback:
- progress_callback("Fetching robots.txt", 0.75)
- try:
- report["robots"] = fetch_robots(config, http_host)
- except Exception:
- failure_reason = "robots_fetch_failed"
- raise
+ progress_callback("Performing active DNS", 0.25)
+ active_metadata, active_findings = active_dns_sweep(
+ scope.target, selected_profile, dns_bucket
+ )
+ artifacts["dns_active"] = active_metadata
+ findings.extend(active_findings)
+
+ if progress_callback:
+ progress_callback("Running TCP SYN scan", 0.4)
+ tcp_metadata, tcp_findings = tcp_syn_scan(
+ scope.resolved_ip or scope.target,
+ selected_profile,
+ limited_ports,
+ tcp_bucket,
+ )
+ artifacts["tcp"] = tcp_metadata
+ findings.extend(tcp_findings)
if progress_callback:
- progress_callback("Generating findings", 0.9)
- try:
- report["findings"] = generate_findings(http_results)
- except Exception:
- failure_reason = "finding_generation_failed"
- raise
+ progress_callback("Running UDP probes", 0.55)
+ udp_metadata, udp_findings = udp_scan_pass(
+ scope.resolved_ip or scope.target, selected_profile, udp_ports, udp_bucket
+ )
+ artifacts["udp"] = udp_metadata
+ findings.extend(udp_findings)
+
+ http_host = _hostname_for_requests(scope, hostname)
+ if tcp_metadata.get("open_ports"):
+ if progress_callback:
+ progress_callback("Performing HTTP checks", 0.7)
+ http_metadata, http_findings = http_checks(
+ scope.resolved_ip or scope.target,
+ http_host,
+ tcp_metadata.get("open_ports", []),
+ selected_profile,
+ http_bucket,
+ )
+ artifacts["http"] = http_metadata
+ findings.extend(http_findings)
+ else:
+ artifacts["http"] = {"services": {}}
+
+ report["artifacts"] = artifacts
+ report["findings"] = findings
completed_at = datetime.now(timezone.utc)
- duration = time.perf_counter() - started_clock
+ duration = time.perf_counter() - start_clock
embed_runtime_metadata(
report, started_at, completed_at=completed_at, duration=duration
)
- record_scan_completed(scope.target, duration, len(open_ports))
+ record_scan_completed(
+ scope.target, duration, len(tcp_metadata.get("open_ports", []))
+ )
REPORT_LOGGER.info(serialize_results(report))
if progress_callback:
@@ -219,5 +237,4 @@ def run_recon(
__all__ = [
"run_recon",
"ReconError",
- "check_security_headers",
]
diff --git a/reconscript/report.py b/reconscript/report.py
index 448b5e9..150523c 100644
--- a/reconscript/report.py
+++ b/reconscript/report.py
@@ -1,3 +1,5 @@
+# Authorized testing only — do not scan targets without explicit permission.
+# This tool is non-intrusive by default and will not perform exploitation or credentialed checks.
"""Report persistence and integrity helpers."""
from __future__ import annotations
@@ -11,7 +13,7 @@
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
-from typing import Dict, Optional
+from typing import Any
from .consent import sign_report_hash
@@ -25,15 +27,15 @@ 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:
deadline = time.monotonic() + self.timeout
@@ -43,7 +45,9 @@ def __enter__(self) -> FileLock:
break
except FileExistsError:
if time.monotonic() >= deadline:
- raise TimeoutError(f"Timed out acquiring lock {self.path}")
+ raise TimeoutError(
+ f"Timed out acquiring lock {self.path}"
+ ) from None
time.sleep(0.1)
return self
@@ -56,6 +60,20 @@ def __exit__(self, exc_type, exc, tb) -> None:
pass
+def _stringify_keys(value: Any) -> Any:
+ if isinstance(value, dict):
+ return {str(key): _stringify_keys(subvalue) for key, subvalue in value.items()}
+ if isinstance(value, list):
+ return [_stringify_keys(item) for item in value]
+ return value
+
+
+def _canonicalize_report(report: dict[str, object]) -> dict[str, object]:
+ if not isinstance(report, dict):
+ return {}
+ return _stringify_keys(report)
+
+
def ensure_results_dir() -> Path:
RESULTS_DIR.mkdir(parents=True, exist_ok=True)
return RESULTS_DIR
@@ -86,16 +104,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 +131,13 @@ def embed_runtime_metadata(
return report
-def _canonical_report_bytes(report: Dict[str, object]) -> bytes:
- payload = {key: value for key, value in report.items() if key != "report_hash"}
+def _canonical_report_bytes(report: dict[str, object]) -> bytes:
+ normalized = _canonicalize_report(report)
+ payload = {key: value for key, value in normalized.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
@@ -140,7 +159,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 +172,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()
@@ -163,22 +182,24 @@ def persist_report(
report_dir = RESULTS_DIR / report_id
report_dir.mkdir(parents=True, exist_ok=True)
- report_hash = compute_report_hash(report)
+ normalized_report = _canonicalize_report(report)
+ report_hash = compute_report_hash(normalized_report)
report["report_hash"] = report_hash
+ normalized_report["report_hash"] = report_hash
report_file = report_dir / "report.json"
report_file.write_text(
- json.dumps(report, indent=2, sort_keys=True), encoding="utf-8"
+ json.dumps(normalized_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 aba16a9..7d5e924 100644
--- a/reconscript/reporters.py
+++ b/reconscript/reporters.py
@@ -1,3 +1,5 @@
+# Authorized testing only — do not scan targets without explicit permission.
+# This tool is non-intrusive by default and will not perform exploitation or credentialed checks.
"""Report rendering utilities for ReconScript."""
from __future__ import annotations
@@ -6,10 +8,13 @@
import html
import json
import logging
+import textwrap
from collections.abc import Iterable, Sequence
from datetime import datetime
from pathlib import Path
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any
+
+from . import __version__
PACKAGE_DIR = Path(__file__).resolve().parent
PROJECT_TEMPLATES = PACKAGE_DIR.parent / "templates"
@@ -25,7 +30,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,46 +38,165 @@
autoescape=select_autoescape(["html", "xml"]),
)
-from . import __version__
-
LOGGER = logging.getLogger(__name__)
PDF_FALLBACK_MESSAGE = "PDF dependencies not found — served HTML version instead."
+MAX_RAW_SNIPPET = 400
+
+
+def _merge_runtime(
+ metadata: dict[str, object], data: dict[str, object]
+) -> dict[str, object]:
+ runtime = data.get("runtime")
+ if not isinstance(runtime, dict):
+ runtime = {}
+ merged: dict[str, object] = dict(runtime)
+ for key in ("started_at", "completed_at"):
+ candidate = data.get(key) or metadata.get(key)
+ if candidate and key not in merged:
+ merged[key] = candidate
+ duration = data.get("duration")
+ if duration is not None and "duration" not in merged:
+ merged["duration"] = duration
+ return merged
+
+
+def _clip_snippet(value: str) -> str:
+ snippet = value.strip()
+ if len(snippet) <= MAX_RAW_SNIPPET:
+ return snippet
+ return snippet[: MAX_RAW_SNIPPET - 3] + "..."
+
+
+def _sanitize_findings(findings: object) -> list[dict[str, object]]:
+ sanitized: list[dict[str, object]] = []
+ if not isinstance(findings, Sequence):
+ return sanitized
+
+ for item in findings:
+ if not isinstance(item, dict):
+ continue
+
+ summary = str(item.get("summary") or item.get("issue") or "observation").strip()
+ tool = str(item.get("tool") or "unknown").strip()
+ cmdline = str(item.get("cmdline") or "").strip()
+ started_at = str(item.get("started_at") or "").strip()
+ completed_at = str(item.get("completed_at") or "").strip()
+ raw_snippet_obj = item.get("raw_snippet")
+ raw_snippet = "" if raw_snippet_obj is None else str(raw_snippet_obj)
+ severity = item.get("severity")
+ port = item.get("port")
+
+ sanitized_entry: dict[str, object] = {
+ "summary": summary or "observation",
+ "tool": tool,
+ "cmdline": cmdline,
+ "started_at": started_at,
+ "completed_at": completed_at,
+ "raw_snippet": _clip_snippet(raw_snippet),
+ }
+
+ if port is not None:
+ sanitized_entry["port"] = port
+
+ if severity:
+ sanitized_entry["severity"] = str(severity)
+
+ evidence = item.get("evidence")
+ if isinstance(evidence, dict) and evidence:
+ sanitized_entry["evidence"] = evidence
+
+ sanitized.append(sanitized_entry)
+
+ return sanitized
+
+
+def _normalize_report(data: dict[str, object]) -> dict[str, object]:
+ if not isinstance(data, dict):
+ return {
+ "target": "unknown",
+ "hostname": "N/A",
+ "ports": [],
+ "open_ports": [],
+ "findings": [],
+ "runtime": {},
+ "timestamp": datetime.utcnow().isoformat() + "Z",
+ }
+ metadata = data.get("metadata") if isinstance(data.get("metadata"), dict) else {}
+ artifacts = data.get("artifacts") if isinstance(data.get("artifacts"), dict) else {}
+
+ normalized: dict[str, object] = dict(data)
+ normalized["target"] = metadata.get("target", data.get("target", "unknown"))
+ normalized["hostname"] = metadata.get("hostname", data.get("hostname"))
+ normalized["ports"] = list(metadata.get("scanned_tcp_ports", data.get("ports", [])))
+ normalized["udp_ports"] = list(metadata.get("scanned_udp_ports", []))
+ normalized["open_ports"] = list(
+ artifacts.get("tcp", {}).get("open_ports", data.get("open_ports", []))
+ )
+ normalized["http_services"] = artifacts.get("http", {}).get("services", {})
+ normalized["runtime"] = _merge_runtime(metadata, data)
+ normalized["timestamp"] = (
+ normalized.get("timestamp")
+ or normalized["runtime"].get("started_at")
+ or datetime.utcnow().isoformat() + "Z"
+ )
+ normalized_findings = _sanitize_findings(data.get("findings"))
+ normalized["findings"] = normalized_findings
+ normalized["legacy_findings"] = [
+ {
+ "port": item.get("port", "n/a"),
+ "issue": item.get("summary"),
+ "details": {
+ "tool": item.get("tool"),
+ "cmdline": item.get("cmdline"),
+ "started_at": item.get("started_at"),
+ "completed_at": item.get("completed_at"),
+ "raw_snippet": item.get("raw_snippet"),
+ },
+ }
+ for item in normalized_findings
+ ]
+ normalized["profile"] = metadata.get("profile", {})
+ normalized["evidence_level"] = metadata.get(
+ "evidence_level", data.get("evidence_level")
+ )
+ normalized["consent_present"] = metadata.get(
+ "consent_present", data.get("consent_present")
+ )
+ normalized["version"] = metadata.get("version", data.get("version", __version__))
+ return normalized
-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."""
+ normalized = _normalize_report(data)
+
if _JINJA_ENV is not None and _JINJA_ENV.loader is not None:
try:
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)
+ return template.render(data=normalized)
+ except Exception as exc: # pragma: no cover - defensive logging
+ LOGGER.debug("Falling back to builtin Markdown renderer: %s", exc)
- lines: List[str] = []
+ context = _build_markdown_context(normalized)
+ 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):
out_path = Path(out_path)
- context = _build_template_context(data)
+ normalized = _normalize_report(data)
+ context = _build_template_context(normalized)
out_path.parent.mkdir(parents=True, exist_ok=True)
html_output = _render_html_document(context)
@@ -80,7 +204,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 +238,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,36 +276,45 @@ 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] = []
- issues = {item.get("issue") for item in findings}
+def _build_recommendations(findings: Sequence[dict[str, object]]) -> list[str]:
+ suggestions: list[str] = []
+ summaries = {str(item.get("summary")) for item in findings}
+
+ for summary in summaries:
+ lowered = summary.lower()
+ if "security header" in lowered:
+ header_message = (
+ "Review HTTP response headers and enable Strict-Transport-"
+ "Security where appropriate."
+ )
+ suggestions.append(header_message)
+ if "cookie" in lowered:
+ suggestions.append(
+ "Verify cookies include Secure and HttpOnly attributes on production systems."
+ )
+ if "error" in lowered or "5xx" in lowered:
+ suggestions.append(
+ "Investigate server error responses for potential misconfigurations."
+ )
- if "missing_security_headers" in issues:
- suggestions.append(
- "Set Strict-Transport-Security and related headers on all web front-ends."
- )
- if "session_cookie_flags" in issues:
- suggestions.append(
- "Configure Secure and HttpOnly attributes on session cookies."
- )
- if "server_error" in issues:
- suggestions.append(
- "Review backend error logs for endpoints returning 5xx responses."
- )
if not suggestions and findings:
suggestions.append(
- "Investigate informational findings for context-specific remediation."
+ "Review informational observations for context-specific improvements."
)
+
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", [])
+ http_services = data.get("http_services", {})
+ profile = data.get("profile", {})
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",
+ "udp_ports": _format_list(data.get("udp_ports", [])) or "None",
"open_ports": _format_list(data.get("open_ports", [])) or "None detected",
"findings": findings if isinstance(findings, Sequence) else [],
"recommendations": _build_recommendations(
@@ -189,41 +322,95 @@ def _build_markdown_context(data: Dict[str, object]) -> Dict[str, Any]:
),
"version": data.get("version", __version__),
"runtime": data.get("runtime", {}),
+ "http_services": http_services if isinstance(http_services, dict) else {},
+ "profile": profile if isinstance(profile, dict) else {},
+ "profile_name": profile.get("name") if isinstance(profile, dict) else "N/A",
+ "evidence_level": data.get("evidence_level", "unknown"),
+ "consent_present": data.get("consent_present"),
}
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']}",
f"*Hostname:* {context['hostname']}",
+ f"*Evidence level:* {context['evidence_level']}",
+ f"*Profile:* {context['profile_name']}",
f"*Ports scanned:* {context['ports']}",
+ f"*UDP ports scanned:* {context['udp_ports']}",
f"*Open ports:* {context['open_ports']}",
"",
- "## Findings",
+ "## Recon Profile",
]
+ profile = context.get("profile", {})
+ if profile:
+ lines.extend(
+ [
+ f"- Max TCP ports: {profile.get('max_tcp_ports', 'unknown')}",
+ f"- TCP concurrency: {profile.get('tcp_concurrency', 'unknown')}",
+ f"- Directory enumeration: {profile.get('dir_enum', 'unknown')}",
+ ]
+ )
+ else:
+ lines.append("- Profile details unavailable.")
+
+ lines.append("")
+ lines.append("## HTTP")
+ services = context.get("http_services", {})
+ if services:
+ for port, info in sorted(services.items(), key=lambda item: item[0]):
+ if isinstance(info, dict):
+ status = info.get("status_code", "n/a")
+ url = info.get("url", "")
+ lines.append(f"- Port {port}: status {status} — {url}")
+ else:
+ lines.append(f"- Port {port}: {info}")
+ else:
+ lines.append("No HTTP services detected.")
+
+ lines.append("")
+ lines.append("## Findings")
findings = context.get("findings", [])
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)
- )
- lines.append(f"- Port `{port}` — `{issue}`")
- if isinstance(item, dict) and item.get("details") is not None:
- details = item["details"]
- if isinstance(details, (dict, list)):
- rendered = json.dumps(details, indent=2, sort_keys=True)
- lines.append(" ```json")
- lines.extend(f" {line}" for line in rendered.splitlines())
- lines.append(" ```")
- else:
- lines.append(f" {details}")
+ for index, item in enumerate(findings, start=1):
+ if not isinstance(item, dict):
+ lines.append(f"- {item}")
+ continue
+
+ summary = item.get("summary", "Observation")
+ lines.append(f"### Finding {index}: {summary}")
+
+ tool = item.get("tool")
+ if tool:
+ lines.append(f"- **Tool:** {tool}")
+
+ cmdline = item.get("cmdline")
+ if cmdline:
+ lines.append(f"- **Command:** `{cmdline}`")
+
+ started = item.get("started_at")
+ completed = item.get("completed_at")
+ if started or completed:
+ timing_parts: list[str] = []
+ if started:
+ timing_parts.append(f"started {started}")
+ if completed:
+ timing_parts.append(f"completed {completed}")
+ lines.append(f"- **Timing:** {', '.join(timing_parts)}")
+
+ severity = item.get("severity")
+ if severity:
+ lines.append(f"- **Severity:** {severity}")
+
+ snippet = item.get("raw_snippet")
+ if snippet:
+ lines.append("```text")
+ lines.extend(str(snippet).splitlines())
+ lines.append("```")
else:
lines.append("No findings reported.")
@@ -245,10 +432,13 @@ def _render_markdown_sections(context: Dict[str, Any]) -> List[str]:
f"- **Runtime:** {runtime}",
]
)
+ consent = context.get("consent_present")
+ if consent is not None:
+ lines.append(f"- **Consent provided:** {bool(consent)}")
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
@@ -278,7 +468,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", [])
@@ -288,7 +478,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),
@@ -305,21 +495,28 @@ 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 []:
+ if not isinstance(item, dict):
+ continue
formatted.append(
{
- "port": item.get("port", "n/a"),
- "issue": item.get("issue", "observation"),
- "details": item.get("details"),
+ "summary": item.get("summary", "Observation"),
+ "tool": item.get("tool", "unknown"),
+ "cmdline": item.get("cmdline", ""),
+ "started_at": item.get("started_at", ""),
+ "completed_at": item.get("completed_at", ""),
+ "raw_snippet": item.get("raw_snippet", ""),
+ "severity": item.get("severity"),
+ "port": item.get("port"),
}
)
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),
]
@@ -359,7 +556,28 @@ 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 _format_html_timing(item: dict[str, object]) -> str:
+ started = item.get("started_at")
+ completed = item.get("completed_at")
+ parts: list[str] = []
+ if started:
+ parts.append(f"
Started: {html.escape(str(started))}
")
+ if completed:
+ parts.append(
+ f"Completed: {html.escape(str(completed))}
"
+ )
+ return "".join(parts)
+
+
+def _format_html_command(item: dict[str, object]) -> str:
+ cmdline = item.get("cmdline")
+ if not cmdline:
+ return ""
+ escaped = html.escape(str(cmdline))
+ return f"Command: {escaped}
"
+
+
+def _render_html_document(context: dict[str, object]) -> str:
if _JINJA_ENV is not None:
try:
template = _JINJA_ENV.get_template(HTML_TEMPLATE_NAME)
@@ -369,7 +587,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", [])
@@ -384,17 +602,30 @@ def _render_html_fallback(context: Dict[str, object]) -> str:
findings_html = "".join(
"""
- Port {port} — {issue}
- {details}
+ {summary}
+
+
+ View evidence
+ {snippet}
+
""".strip().format(
- port=html.escape(str(item.get("port", "n/a"))),
- issue=html.escape(str(item.get("issue", "observation"))),
- details=(
- f"{html.escape(json.dumps(item.get('details'), indent=2))} "
- if item.get("details") is not None
- else "No additional context supplied.
"
+ summary=html.escape(str(item.get("summary", "Observation"))),
+ tool=html.escape(str(item.get("tool", "unknown"))),
+ command=_format_html_command(item),
+ timing=_format_html_timing(item),
+ severity=(
+ f"Severity: {html.escape(str(item.get('severity')))}
"
+ if item.get("severity")
+ else ""
),
+ snippet=html.escape(str(item.get("raw_snippet", "")))
+ or "No evidence recorded.",
)
for item in findings
)
@@ -422,6 +653,124 @@ def _render_html_fallback(context: Dict[str, object]) -> str:
+ ""
)
+ styles = textwrap.dedent(
+ """
+ body {
+ font-family: 'Segoe UI', 'Liberation Sans', Arial, sans-serif;
+ margin: 0;
+ background: #f5f7fa;
+ color: #1f2937;
+ }
+
+ header.cover {
+ background: linear-gradient(135deg, #0f172a, #1f2937);
+ color: #f8fafc;
+ padding: 48px 32px;
+ text-align: center;
+ }
+
+ header.cover .logo {
+ display: inline-block;
+ padding: 10px 28px;
+ border: 2px dashed rgba(255, 255, 255, 0.45);
+ border-radius: 10px;
+ letter-spacing: 0.18rem;
+ text-transform: uppercase;
+ margin-bottom: 24px;
+ }
+
+ main {
+ padding: 36px 48px 64px;
+ }
+
+ section {
+ margin-bottom: 48px;
+ }
+
+ h2 {
+ border-bottom: 2px solid #1f2937;
+ padding-bottom: 6px;
+ }
+
+ table.summary {
+ width: 100%;
+ border-collapse: collapse;
+ background: #ffffff;
+ border-radius: 8px;
+ overflow: hidden;
+ box-shadow: 0 2px 6px rgba(15, 23, 42, 0.08);
+ }
+
+ table.summary th,
+ table.summary td {
+ padding: 14px 18px;
+ border-bottom: 1px solid #e2e8f0;
+ }
+
+ table.summary th {
+ width: 30%;
+ background: #f1f5f9;
+ font-weight: 600;
+ letter-spacing: 0.02rem;
+ }
+
+ ul.findings {
+ list-style: none;
+ padding-left: 0;
+ }
+
+ ul.findings li {
+ background: #ffffff;
+ border-left: 4px solid #f97316;
+ padding: 16px 20px;
+ margin-bottom: 14px;
+ box-shadow: 0 1px 3px rgba(15, 23, 42, 0.1);
+ }
+
+ ul.findings li pre {
+ background: #0f172a;
+ color: #f8fafc;
+ padding: 12px;
+ border-radius: 6px;
+ overflow-x: auto;
+ font-size: 0.85rem;
+ }
+
+ ul.recommendations {
+ padding-left: 18px;
+ }
+
+ dl.meta {
+ display: grid;
+ grid-template-columns: 220px 1fr;
+ gap: 12px 18px;
+ background: #ffffff;
+ padding: 18px 22px;
+ border-radius: 8px;
+ box-shadow: 0 1px 3px rgba(15, 23, 42, 0.08);
+ }
+
+ dl.meta dt {
+ font-weight: 600;
+ }
+
+ dl.meta dd {
+ margin: 0;
+ font-family: 'Fira Code', 'Source Code Pro', monospace;
+ white-space: pre-wrap;
+ }
+
+ footer {
+ text-align: center;
+ padding: 24px;
+ background: #f1f5f9;
+ border-top: 1px solid #e2e8f0;
+ font-size: 0.85rem;
+ color: #475569;
+ }
+ """
+ ).strip()
+
return f"""
@@ -429,23 +778,7 @@ def _render_html_fallback(context: Dict[str, object]) -> str:
{html.escape(str(context.get('title')))}
@@ -476,7 +809,8 @@ def _render_html_fallback(context: Dict[str, object]) -> str:
- ReconScript v{html.escape(str(context.get('version')))} · Generated {html.escape(str(context.get('timestamp')))}
+ ReconScript v{html.escape(str(context.get('version')))}
+ · Generated {html.escape(str(context.get('timestamp')))}
"""
diff --git a/reconscript/scanner/__init__.py b/reconscript/scanner/__init__.py
index cfb4e86..84c5ad5 100644
--- a/reconscript/scanner/__init__.py
+++ b/reconscript/scanner/__init__.py
@@ -1,19 +1,19 @@
-"""Scanning primitives for ReconScript with strict rate limiting."""
+# Authorized testing only — do not scan targets without explicit permission.
+# This tool is non-intrusive by default and will not perform exploitation or credentialed checks.
+"""Scanner primitives for ReconScript safe, non-intrusive reconnaissance."""
from __future__ import annotations
-import json
+import contextlib
import logging
-import os
-import random
+import re
+import shutil
import socket
-import ssl
-import time
-from collections.abc import Iterable, Sequence
+from collections import defaultdict
+from collections.abc import Iterable, Mapping, MutableMapping, Sequence
from concurrent.futures import ThreadPoolExecutor, as_completed
-from dataclasses import dataclass
-from http.cookies import CookieError, SimpleCookie
-from typing import Dict, List, Optional, Tuple
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
import requests
from requests import Response
@@ -23,49 +23,167 @@
LOGGER = logging.getLogger(__name__)
-DEFAULT_PORTS: Tuple[int, ...] = (80, 443, 8080, 8443, 8000, 3000)
-HTTP_PORTS: Tuple[int, ...] = DEFAULT_PORTS
-USER_AGENT = "ReconScript/secure"
-
-TOKEN_RATE = float(os.environ.get("TOKEN_RATE", "5"))
-TOKEN_CAPACITY = float(os.environ.get("TOKEN_CAPACITY", "10"))
-HTTP_WORKERS = int(os.environ.get("HTTP_WORKERS", "2"))
-CONNECT_TIMEOUT = float(os.environ.get("CONNECT_TIMEOUT", "5"))
-READ_TIMEOUT = float(os.environ.get("READ_TIMEOUT", "10"))
-MAX_HTTP_RETRIES = int(os.environ.get("MAX_HTTP_RETRIES", "3"))
-
-DEFAULT_REDACTIONS = {
- "cookie",
- "authorization",
- "set-cookie",
- "x-api-key",
- "x-auth-token",
+WHOIS_SAFE_PATTERN = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
+
+PRIORITY_TCP_PORTS: tuple[int, ...] = (
+ 80,
+ 443,
+ 22,
+ 21,
+ 25,
+ 110,
+ 143,
+ 53,
+ 123,
+ 3389,
+ 5900,
+ 8080,
+ 8443,
+ 8000,
+ 3306,
+ 5432,
+ 6379,
+ 27017,
+ 5985,
+ 9200,
+)
+DEFAULT_TCP_PORTS: tuple[int, ...] = PRIORITY_TCP_PORTS
+DEFAULT_UDP_PORTS: tuple[int, ...] = (53, 123, 161)
+HTTP_PORTS: tuple[int, ...] = (80, 443, 8080, 8443, 8000, 3000)
+DIR_ENUM_LEVELS = {"none", "low", "medium"}
+MAX_RAW_SNIPPET = 400
+
+DEFAULT_TIMEOUTS: Mapping[str, float] = {
+ "tcp_connect": 2.5,
+ "tcp_syn": 2.5,
+ "udp": 2.0,
+ "http": 5.0,
+ "dns": 3.0,
}
-_env_redactions = {
- entry.strip().lower()
- for entry in os.environ.get("REDACT_KEYS", "").split(",")
- if entry.strip()
+DEFAULT_RATE_LIMITS: Mapping[str, tuple[float, float]] = {
+ "tcp": (5.0, 5.0),
+ "udp": (2.0, 2.0),
+ "dns": (2.0, 2.0),
+ "http": (2.0, 2.0),
}
-REDACTION_KEYS = DEFAULT_REDACTIONS | _env_redactions
+SAFE_WORDLIST_LOW: tuple[str, ...] = (
+ "robots.txt",
+ "sitemap.xml",
+ "admin",
+ "login",
+ "static",
+ "health",
+ "status",
+)
+SAFE_WORDLIST_MEDIUM: tuple[str, ...] = SAFE_WORDLIST_LOW + (
+ ".well-known/security.txt",
+ "server-status",
+ "config",
+ "api",
+ "assets",
+ "dashboard",
+ "docs",
+)
+
+
+@dataclass
+class ReconProfile:
+ """Profile encapsulating the safe scanning parameters."""
+
+ name: str
+ max_tcp_ports: int = 100
+ tcp_concurrency: int = 10
+ udp_pass: bool = True
+ dns_active: bool = True
+ dns_passive: bool = True
+ dir_enum: str = "low"
+ allow_credentialed: bool = False
+ timeouts: MutableMapping[str, float] = field(
+ default_factory=lambda: dict(DEFAULT_TIMEOUTS)
+ )
+ rate_limits: MutableMapping[str, tuple[float, float]] = field(
+ default_factory=lambda: dict(DEFAULT_RATE_LIMITS)
+ )
-@dataclass(frozen=True)
-class ScanConfig:
- target: str
- hostname: Optional[str]
- ports: Sequence[int]
- enable_ipv6: bool
- evidence_level: str
- redaction_keys: set[str]
- connect_timeout: float = CONNECT_TIMEOUT
- read_timeout: float = READ_TIMEOUT
- http_workers: int = HTTP_WORKERS
- max_retries: int = MAX_HTTP_RETRIES
+ def __post_init__(self) -> None:
+ if self.max_tcp_ports <= 0:
+ raise ValueError("max_tcp_ports must be positive")
+ if self.tcp_concurrency <= 0:
+ raise ValueError("tcp_concurrency must be positive")
+ if self.dir_enum not in DIR_ENUM_LEVELS:
+ raise ValueError(f"dir_enum must be one of {sorted(DIR_ENUM_LEVELS)}")
+ normalized_timeouts = dict(DEFAULT_TIMEOUTS)
+ normalized_timeouts.update(self.timeouts)
+ self.timeouts = normalized_timeouts
+ normalized_rates = dict(DEFAULT_RATE_LIMITS)
+ normalized_rates.update(self.rate_limits)
+ self.rate_limits = normalized_rates
+ if self.allow_credentialed:
+ LOGGER.warning(
+ "Credentialed scanning remains disabled despite profile flag."
+ )
+ def as_dict(self) -> dict[str, object]:
+ return {
+ "name": self.name,
+ "max_tcp_ports": self.max_tcp_ports,
+ "tcp_concurrency": self.tcp_concurrency,
+ "udp_pass": self.udp_pass,
+ "dns_active": self.dns_active,
+ "dns_passive": self.dns_passive,
+ "dir_enum": self.dir_enum,
+ "allow_credentialed": self.allow_credentialed,
+ "timeouts": dict(self.timeouts),
+ "rate_limits": {
+ key: {"rate": rate, "burst": burst}
+ for key, (rate, burst) in self.rate_limits.items()
+ },
+ }
-def validate_port_list(ports: Iterable[int]) -> Tuple[int, ...]:
- validated: List[int] = []
+ def limit_ports(self, ports: Sequence[int] | None) -> tuple[int, ...]:
+ candidate_ports = (
+ validate_port_list(ports) if ports is not None else DEFAULT_TCP_PORTS
+ )
+ prioritized: list[int] = []
+ for port in PRIORITY_TCP_PORTS:
+ if port in candidate_ports and port not in prioritized:
+ prioritized.append(port)
+ for port in candidate_ports:
+ if port not in prioritized:
+ prioritized.append(port)
+ return tuple(prioritized[: self.max_tcp_ports])
+
+
+def profile_for_evidence(evidence_level: str) -> ReconProfile:
+ level = evidence_level.lower()
+ if level not in {"low", "medium", "high"}:
+ raise ValueError("Evidence level must be low, medium, or high")
+ if level == "low":
+ return ReconProfile(
+ name="safe-low",
+ max_tcp_ports=80,
+ tcp_concurrency=6,
+ dir_enum="low",
+ )
+ if level == "medium":
+ return ReconProfile(
+ name="safe-medium",
+ max_tcp_ports=120,
+ tcp_concurrency=8,
+ dir_enum="medium",
+ )
+ return ReconProfile(
+ name="safe-high",
+ max_tcp_ports=150,
+ tcp_concurrency=10,
+ dir_enum="none",
+ )
+
+
+def validate_port_list(ports: Iterable[int]) -> tuple[int, ...]:
+ validated: list[int] = []
for port in ports:
if not isinstance(port, int):
raise TypeError("Ports must be integers.")
@@ -76,47 +194,37 @@ def validate_port_list(ports: Iterable[int]) -> Tuple[int, ...]:
return tuple(validated)
-def resolve_addresses(
- target: str, enable_ipv6: bool
-) -> List[Tuple[int, int, int, str, Tuple[str, int]]]:
- family = socket.AF_UNSPEC if enable_ipv6 else socket.AF_INET
- try:
- infos = socket.getaddrinfo(target, None, family=family, type=socket.SOCK_STREAM)
- except socket.gaierror as exc:
- raise RuntimeError(f"Unable to resolve target address: {exc}") from exc
- unique = []
- seen = set()
- for info in infos:
- sockaddr = info[-1]
- key = (info[0], sockaddr[0])
- if key in seen:
- continue
- seen.add(key)
- unique.append(info)
- return unique
+def _utcnow() -> datetime:
+ return datetime.now(timezone.utc)
-def tcp_connect_scan(config: ScanConfig, bucket: TokenBucket) -> List[int]:
- LOGGER.info("Commencing TCP connect scan with token bucket rate limiting.")
- addresses = resolve_addresses(config.target, config.enable_ipv6)
- open_ports: List[int] = []
- for port in config.ports:
- bucket.consume()
- success = False
- for family, _, _, _, sockaddr in addresses:
- host = sockaddr[0]
- try:
- with socket.create_connection(
- (host, port), timeout=config.connect_timeout
- ):
- success = True
- break
- except (socket.timeout, ConnectionRefusedError, OSError) as exc:
- LOGGER.debug("Port %s closed on %s (%s)", port, host, exc)
- continue
- if success:
- open_ports.append(port)
- return open_ports
+def _isoformat(dt: datetime) -> str:
+ return dt.replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def _clip_snippet(snippet: str) -> str:
+ if len(snippet) <= MAX_RAW_SNIPPET:
+ return snippet
+ return snippet[: MAX_RAW_SNIPPET - 3] + "..."
+
+
+def _make_finding(
+ *,
+ tool: str,
+ cmdline: str,
+ summary: str,
+ raw_snippet: str,
+ started_at: datetime,
+ completed_at: datetime,
+) -> dict[str, object]:
+ return {
+ "tool": tool,
+ "cmdline": cmdline,
+ "summary": summary,
+ "started_at": _isoformat(started_at),
+ "completed_at": _isoformat(completed_at),
+ "raw_snippet": _clip_snippet(raw_snippet),
+ }
def _build_url(hostname: str, port: int) -> str:
@@ -126,13 +234,11 @@ def _build_url(hostname: str, port: int) -> str:
return f"{scheme}://{hostname}:{port}"
-def _redact_headers(
- headers: Dict[str, str], redaction_keys: set[str]
-) -> Dict[str, str]:
- sanitized: Dict[str, str] = {}
+def _redact_headers(headers: dict[str, str]) -> dict[str, str]:
+ sanitized: dict[str, str] = {}
for key, value in headers.items():
lower = key.lower()
- redact = lower in redaction_keys
+ redact = lower in {"cookie", "authorization", "set-cookie"}
if lower.startswith("x-") and any(
token in lower for token in ("auth", "token", "key")
):
@@ -141,285 +247,451 @@ def _redact_headers(
return sanitized
-def _parse_cookie_flags(response: Response) -> Optional[Dict[str, bool]]:
- header_values: List[str] = []
- if hasattr(response.raw, "headers"):
- raw_headers = response.raw.headers
- if hasattr(raw_headers, "getlist"):
- header_values.extend(raw_headers.getlist("Set-Cookie")) # type: ignore[attr-defined]
- elif hasattr(raw_headers, "get_all"):
- header_values.extend(raw_headers.get_all("Set-Cookie")) # type: ignore[attr-defined]
- if not header_values:
- header = response.headers.get("Set-Cookie")
- if header:
- header_values.append(header)
- if not header_values:
- return None
-
- secure_flag = False
- httponly_flag = False
- for value in header_values:
- cookie = SimpleCookie()
- try:
- cookie.load(value)
- except CookieError:
- lowered = value.lower()
- secure_flag = secure_flag or ("secure" in lowered)
- httponly_flag = httponly_flag or ("httponly" in lowered)
- continue
- for morsel in cookie.values():
- secure_flag = secure_flag or bool(morsel["secure"])
- httponly_flag = httponly_flag or bool(morsel["httponly"])
- return {"secure": secure_flag, "httponly": httponly_flag}
-
-
-def check_security_headers(headers: Dict[str, str]) -> Dict[str, object]:
- required_headers = {
- "Strict-Transport-Security": "HSTS",
- "Content-Security-Policy": "CSP",
- "X-Frame-Options": "Clickjacking protection",
- "Referrer-Policy": "Referrer policy",
- "Permissions-Policy": "Permissions policy",
- "X-Content-Type-Options": "MIME sniffing protection",
- "X-XSS-Protection": "Legacy XSS filter",
- }
+def passive_dns_collection(
+ target: str, profile: ReconProfile
+) -> tuple[dict[str, object], list[dict[str, object]]]:
+ metadata: dict[str, object] = {"addresses": [], "aliases": []}
+ findings: list[dict[str, object]] = []
+ if not profile.dns_passive:
+ return metadata, findings
- present: Dict[str, str] = {}
- missing: List[str] = []
- header_keys = {key.lower(): key for key in headers.keys()}
- for header in required_headers:
- lowered = header.lower()
- if lowered in header_keys:
- present[header] = headers[header_keys[lowered]]
+ started = _utcnow()
+ lines: list[str] = []
+ summary = "Passive DNS lookup executed."
+ try:
+ host, aliases, addresses = socket.gethostbyname_ex(target)
+ metadata["hostname"] = host
+ metadata["aliases"] = aliases
+ metadata["addresses"] = addresses
+ lines.append(f"Host: {host}")
+ if aliases:
+ lines.append("Aliases: " + ", ".join(aliases))
+ if addresses:
+ lines.append("Addresses: " + ", ".join(addresses))
else:
- missing.append(header)
- return {"present": present, "missing": missing}
+ lines.append("No A records discovered.")
+ summary = f"Passive DNS resolved {len(addresses)} address(es)."
+ except socket.gaierror as exc:
+ lines.append(f"DNS resolution failed: {exc}")
+ summary = "Passive DNS resolution failed."
+ completed = _utcnow()
+ findings.append(
+ _make_finding(
+ tool="passive-dns",
+ cmdline=f"passive_dns --target {target}",
+ summary=summary,
+ raw_snippet="\n".join(lines) or "No passive DNS data",
+ started_at=started,
+ completed_at=completed,
+ )
+ )
+ whois_lines: list[str] = []
+ whois_summary = "WHOIS lookup skipped."
+ whois_started = _utcnow()
+ try:
+ import subprocess
-def _detect_external_redirect(expected_host: str, response: Response) -> Optional[str]:
- expected = expected_host.lower()
- for hop in (*response.history, response):
- parsed = requests.utils.urlparse(hop.url)
- host = (parsed.hostname or "").lower()
- if host and host != expected:
- return hop.url
- return None
+ if not WHOIS_SAFE_PATTERN.fullmatch(target):
+ whois_lines.append("WHOIS target contains unsupported characters.")
+ else:
+ executable = shutil.which("whois")
+ if executable is None:
+ whois_lines.append("whois binary not available.")
+ else:
+ try:
+ result = subprocess.run( # noqa: S603, S607 - sanitized target and explicit binary
+ [executable, target],
+ capture_output=True,
+ text=True,
+ timeout=profile.timeouts.get("dns", 3.0),
+ check=False,
+ )
+ stdout = result.stdout.strip()
+ stderr = result.stderr.strip()
+ if stdout:
+ whois_lines.append(stdout[:300])
+ elif stderr:
+ whois_lines.append(stderr[:300])
+ else:
+ whois_lines.append("WHOIS returned no output.")
+ whois_summary = "WHOIS lookup executed."
+ except subprocess.SubprocessError as exc:
+ whois_lines.append(f"WHOIS error: {exc}")
+ whois_summary = "WHOIS lookup encountered an error."
+ except ImportError:
+ whois_lines.append("subprocess module unavailable for WHOIS.")
+ except Exception as exc: # pragma: no cover - defensive
+ whois_lines.append(f"Unexpected WHOIS error: {exc}")
+ whois_summary = "WHOIS lookup encountered an error."
+ whois_completed = _utcnow()
+ findings.append(
+ _make_finding(
+ tool="passive-dns",
+ cmdline=f"whois {target}",
+ summary=whois_summary,
+ raw_snippet="\n".join(whois_lines) or "No WHOIS data",
+ started_at=whois_started,
+ completed_at=whois_completed,
+ )
+ )
+ ct_started = _utcnow()
+ findings.append(
+ _make_finding(
+ tool="passive-dns",
+ cmdline=f"ct-lookup --domain {target}",
+ summary="Certificate transparency lookup not performed (offline mode).",
+ raw_snippet=(
+ "CT log queries require external network access and are disabled in "
+ "the safe default profile."
+ ),
+ started_at=ct_started,
+ completed_at=_utcnow(),
+ )
+ )
+ return metadata, findings
-def _http_request(
- url: str, max_retries: int, connect_timeout: float, read_timeout: float
-) -> Tuple[Optional[Response], Optional[str]]:
- headers = {"User-Agent": USER_AGENT}
- attempt = 0
- while attempt <= max_retries:
+
+def active_dns_sweep(
+ target: str, profile: ReconProfile, bucket: TokenBucket
+) -> tuple[dict[str, list[str]], list[dict[str, object]]]:
+ metadata: dict[str, list[str]] = defaultdict(list)
+ findings: list[dict[str, object]] = []
+ if not profile.dns_active:
+ return metadata, findings
+
+ resolver = None
+ with contextlib.suppress(ImportError):
+ import dns.resolver # type: ignore
+
+ resolver = dns.resolver.Resolver()
+ resolver.lifetime = profile.timeouts.get("dns", 3.0)
+
+ record_types = ("A", "AAAA", "CNAME", "TXT", "MX")
+ for record_type in record_types:
+ bucket.consume()
+ started = _utcnow()
+ snippet_lines: list[str] = []
+ summary = "Active DNS query executed."
try:
- response = requests.get(
- url,
- headers=headers,
- allow_redirects=True,
- timeout=(connect_timeout, read_timeout),
+ if record_type in {"A", "AAAA"}:
+ family = socket.AF_INET if record_type == "A" else socket.AF_INET6
+ answers = socket.getaddrinfo(target, None, family=family)
+ seen: list[str] = []
+ for answer in answers:
+ host = answer[4][0]
+ if host not in seen:
+ seen.append(host)
+ if seen:
+ metadata[record_type].extend(seen)
+ snippet_lines.append(f"{record_type} records: " + ", ".join(seen))
+ summary = f"{record_type} query returned {len(seen)} record(s)."
+ else:
+ snippet_lines.append(f"No {record_type} records discovered.")
+ summary = f"No {record_type} records discovered."
+ elif resolver is not None:
+ import dns.resolver # type: ignore
+
+ answers = resolver.resolve(target, record_type)
+ values = [str(item).strip() for item in answers]
+ metadata[record_type].extend(values)
+ snippet_lines.append(
+ f"{record_type} records: " + ", ".join(values)
+ if values
+ else "No records"
+ )
+ summary = f"{record_type} query completed with {len(values)} result(s)."
+ else:
+ summary = f"{record_type} query skipped (dnspython not installed)."
+ snippet_lines.append("dnspython not installed; skipping.")
+ except Exception as exc:
+ snippet_lines.append(f"{record_type} lookup error: {exc}")
+ summary = f"{record_type} query failed."
+ completed = _utcnow()
+ findings.append(
+ _make_finding(
+ tool="dns-active",
+ cmdline=f"dnsquery --type {record_type} --target {target}",
+ summary=summary,
+ raw_snippet="\n".join(snippet_lines) or f"No {record_type} data",
+ started_at=started,
+ completed_at=completed,
)
- return response, None
- except requests_exceptions.RequestException as exc:
- if attempt >= max_retries:
- return None, str(exc)
- sleep = min(4.0, (2**attempt) * 0.5)
- jitter = random.uniform(0, 0.25)
- time.sleep(sleep + jitter)
- attempt += 1
- return None, "unreachable"
-
-
-def _http_probe_single(
- config: ScanConfig,
- hostname: str,
- port: int,
-) -> Dict[str, object]:
- url = _build_url(hostname, port)
- response, error = _http_request(
- url, config.max_retries, config.connect_timeout, config.read_timeout
- )
- if error:
- LOGGER.debug("HTTP probe for %s failed: %s", url, error)
- return {"error": error}
-
- assert response is not None
-
- redirect = _detect_external_redirect(hostname, response)
- if redirect:
- LOGGER.warning("Blocked redirect from %s to external host %s", url, redirect)
- return {"error": "redirected externally", "redirect_url": redirect}
-
- headers = dict(response.headers)
- sanitized_headers = _redact_headers(headers, config.redaction_keys)
- metadata: Dict[str, object] = {
- "url": response.url,
- "status_code": response.status_code,
- "headers": sanitized_headers,
- }
- cookie_flags = _parse_cookie_flags(response)
- if cookie_flags:
- metadata["cookie_flags"] = cookie_flags
- metadata["security_headers_check"] = check_security_headers(sanitized_headers)
-
- if config.evidence_level == "low":
- return metadata
-
- metadata["screenshots"] = []
- if config.evidence_level == "medium":
- return metadata
-
- request_info = response.request
- metadata["raw_request"] = {
- "method": request_info.method,
- "headers": dict(request_info.headers),
- "body": (
- request_info.body.decode("utf-8", errors="replace")
- if isinstance(request_info.body, bytes)
- else request_info.body
- ),
- }
- metadata["raw_response"] = {
- "status_code": response.status_code,
- "headers": headers,
- "body": response.text,
- }
- return metadata
+ )
+ return metadata, findings
-def http_probe_services(
- config: ScanConfig, hostname: str, ports: Sequence[int]
-) -> Dict[int, Dict[str, object]]:
- results: Dict[int, Dict[str, object]] = {}
- http_ports = list(ports)
- if not http_ports:
- return results
+def udp_scan_pass(
+ target: str,
+ profile: ReconProfile,
+ ports: Sequence[int],
+ bucket: TokenBucket,
+) -> tuple[dict[str, object], list[dict[str, object]]]:
+ metadata: dict[str, object] = {"scanned_ports": list(ports), "responses": {}}
+ findings: list[dict[str, object]] = []
+ if not profile.udp_pass:
+ return metadata, findings
- with ThreadPoolExecutor(max_workers=config.http_workers) as executor:
- future_map = {
- executor.submit(_http_probe_single, config, hostname, port): port
- for port in http_ports
- }
- for future in as_completed(future_map):
- port = future_map[future]
- try:
- results[port] = future.result()
- except Exception as exc: # pragma: no cover - defensive guard
- LOGGER.error(
- "HTTP probe for port %s raised unexpected error: %s", port, exc
- )
- results[port] = {"error": str(exc)}
- return results
+ for port in ports:
+ bucket.consume()
+ started = _utcnow()
+ response_text = ""
+ summary = f"UDP port {port} did not respond."
+ try:
+ with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
+ sock.settimeout(profile.timeouts.get("udp", 2.0))
+ sock.sendto(b"", (target, port))
+ try:
+ data, _ = sock.recvfrom(128)
+ response_text = f"Received {len(data)} bytes"
+ summary = f"UDP port {port} responded with data."
+ metadata["responses"][port] = "response"
+ except socket.timeout:
+ response_text = "No UDP response before timeout."
+ metadata["responses"][port] = "timeout"
+ except OSError as exc:
+ response_text = f"UDP error: {exc}"
+ summary = f"UDP port {port} unreachable."
+ metadata["responses"][port] = "error"
+ completed = _utcnow()
+ findings.append(
+ _make_finding(
+ tool="udp-scan",
+ cmdline=f"udp_probe --target {target} --port {port}",
+ summary=summary,
+ raw_snippet=response_text or "No UDP data returned.",
+ started_at=started,
+ completed_at=completed,
+ )
+ )
+ return metadata, findings
+
+
+def tcp_syn_scan(
+ target: str,
+ profile: ReconProfile,
+ ports: Sequence[int],
+ bucket: TokenBucket,
+) -> tuple[dict[str, object], list[dict[str, object]]]:
+ metadata: dict[str, object] = {
+ "scanned_ports": list(ports),
+ "open_ports": [],
+ }
+ if not ports:
+ return metadata, []
+ open_ports: list[int] = []
+ errors: list[str] = []
+ started = _utcnow()
-def fetch_tls_certificate(config: ScanConfig, port: int) -> Dict[str, object]:
- context = ssl.create_default_context()
- context.check_hostname = False
- addresses = resolve_addresses(config.target, config.enable_ipv6)
- for family, _, _, _, sockaddr in addresses:
- host = sockaddr[0]
+ def probe_port(port: int) -> None:
+ bucket.consume()
try:
- with socket.create_connection(
- (host, port), timeout=config.connect_timeout
- ) as sock:
- with context.wrap_socket(
- sock, server_hostname=config.hostname or config.target
- ) as wrapped:
- certificate = wrapped.getpeercert()
- if not certificate:
- continue
- return {
- "subject": dict(
- entry[0] for entry in certificate.get("subject", [])
- ),
- "issuer": dict(
- entry[0] for entry in certificate.get("issuer", [])
- ),
- "notBefore": certificate.get("notBefore"),
- "notAfter": certificate.get("notAfter"),
- "serialNumber": certificate.get("serialNumber"),
- }
- except (ssl.SSLError, socket.timeout, ConnectionError, OSError) as error:
- LOGGER.debug("TLS retrieval failed on %s:%s: %s", host, port, error)
- continue
- return {"error": "unable to retrieve TLS certificate"}
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+ sock.settimeout(profile.timeouts.get("tcp_syn", 2.5))
+ result = sock.connect_ex((target, port))
+ if result == 0:
+ open_ports.append(port)
+ except OSError as exc:
+ errors.append(f"{port}:{exc}")
+
+ with ThreadPoolExecutor(max_workers=profile.tcp_concurrency) as executor:
+ futures = [executor.submit(probe_port, port) for port in ports]
+ for future in as_completed(futures):
+ future.result()
+ metadata["open_ports"] = sorted(open_ports)
+ summary = f"TCP SYN scan completed. Open ports: {', '.join(map(str, open_ports)) or 'none'}."
+ raw_lines = [summary]
+ if errors:
+ raw_lines.append("Errors: " + "; ".join(errors[:5]))
+ completed = _utcnow()
+ findings = [
+ _make_finding(
+ tool="tcp-syn",
+ cmdline=f"tcp_syn_scan --target {target} --ports {','.join(map(str, ports))}",
+ summary=summary,
+ raw_snippet="\n".join(raw_lines),
+ started_at=started,
+ completed_at=completed,
+ )
+ ]
+ return metadata, findings
-def fetch_robots(config: ScanConfig, hostname: str) -> Dict[str, object]:
- for scheme in ("https", "http"):
- url = f"{scheme}://{hostname}/robots.txt"
- response, error = _http_request(
- url, config.max_retries, config.connect_timeout, config.read_timeout
+def _http_request(
+ method: str,
+ url: str,
+ profile: ReconProfile,
+ bucket: TokenBucket,
+ *,
+ allow_redirects: bool = True,
+) -> tuple[Response | None, str | None]:
+ headers = {"User-Agent": "ReconScript/safe"}
+ timeout = (
+ profile.timeouts.get("http", 5.0),
+ profile.timeouts.get("http", 5.0),
+ )
+ bucket.consume()
+ try:
+ response = requests.request(
+ method,
+ url,
+ headers=headers,
+ allow_redirects=allow_redirects,
+ timeout=timeout,
)
+ return response, None
+ except requests_exceptions.RequestException as exc:
+ return None, str(exc)
+
+
+def http_checks(
+ target: str,
+ hostname: str,
+ ports: Sequence[int],
+ profile: ReconProfile,
+ bucket: TokenBucket,
+) -> tuple[dict[str, object], list[dict[str, object]]]:
+ metadata: dict[str, object] = {"services": {}}
+ findings: list[dict[str, object]] = []
+ http_ports = [port for port in ports if port in HTTP_PORTS]
+ for port in http_ports:
+ url = _build_url(hostname, port)
+ started = _utcnow()
+ response, error = _http_request("GET", url, profile, bucket)
+ completed = _utcnow()
if error:
+ findings.append(
+ _make_finding(
+ tool="http",
+ cmdline=f"http-get {url}",
+ summary=f"HTTP GET failed for {url}",
+ raw_snippet=error,
+ started_at=started,
+ completed_at=completed,
+ )
+ )
+ metadata["services"][port] = {"error": error}
continue
assert response is not None
- redirect = _detect_external_redirect(hostname, response)
- if redirect:
- continue
- if response.status_code == 200 and response.text.strip():
- body = (
- response.text
- if config.evidence_level == "high"
- else response.text[:2000]
+ headers = _redact_headers(dict(response.headers))
+ metadata["services"][port] = {
+ "url": response.url,
+ "status_code": response.status_code,
+ "headers": headers,
+ }
+ findings.append(
+ _make_finding(
+ tool="http",
+ cmdline=f"http-get {url}",
+ summary=f"HTTP {response.status_code} received from {url}",
+ raw_snippet=f"Status {response.status_code}; {len(response.text)} bytes",
+ started_at=started,
+ completed_at=completed,
)
- return {"url": response.url, "body": body}
- return {"note": "robots.txt not present or inaccessible"}
-
-
-def generate_findings(
- http_results: Dict[int, Dict[str, object]],
-) -> List[Dict[str, object]]:
- findings: List[Dict[str, object]] = []
- for port, result in http_results.items():
- headers = result.get("security_headers_check")
- if isinstance(headers, dict):
- missing = headers.get("missing", [])
- if missing:
- findings.append(
- {
- "port": port,
- "issue": "missing_security_headers",
- "details": missing,
- }
- )
- cookie_flags = result.get("cookie_flags")
- if isinstance(cookie_flags, dict) and (
- not cookie_flags.get("secure") or not cookie_flags.get("httponly")
- ):
+ )
+ head_started = _utcnow()
+ head_response, head_error = _http_request("HEAD", url, profile, bucket)
+ head_completed = _utcnow()
+ if head_error:
findings.append(
- {"port": port, "issue": "session_cookie_flags", "details": cookie_flags}
+ _make_finding(
+ tool="http",
+ cmdline=f"http-head {url}",
+ summary=f"HTTP HEAD failed for {url}",
+ raw_snippet=head_error,
+ started_at=head_started,
+ completed_at=head_completed,
+ )
)
- status_code = result.get("status_code")
- if isinstance(status_code, int) and status_code >= 500:
+ else:
+ assert head_response is not None
findings.append(
- {"port": port, "issue": "server_error", "details": status_code}
+ _make_finding(
+ tool="http",
+ cmdline=f"http-head {url}",
+ summary=f"HTTP HEAD returned {head_response.status_code}",
+ raw_snippet="Headers only response recorded.",
+ started_at=head_started,
+ completed_at=head_completed,
+ )
)
+ if profile.dir_enum in {"low", "medium"}:
+ dir_findings = _directory_enum(url, profile, bucket)
+ findings.extend(dir_findings)
+ return metadata, findings
+
+
+def _directory_enum(
+ base_url: str, profile: ReconProfile, bucket: TokenBucket
+) -> list[dict[str, object]]:
+ findings: list[dict[str, object]] = []
+ if profile.dir_enum == "none":
+ return findings
+ wordlist = list(
+ SAFE_WORDLIST_LOW if profile.dir_enum == "low" else SAFE_WORDLIST_MEDIUM
+ )
+ if profile.dir_enum == "medium":
+ combos = [
+ "admin/login",
+ "admin/config",
+ "api/v1",
+ "api/status",
+ ]
+ wordlist.extend(combos)
+ wordlist = wordlist[:200]
+ discovered: list[str] = []
+ for path in wordlist:
+ url = base_url.rstrip("/") + "/" + path
+ response, error = _http_request("HEAD", url, profile, bucket)
+ if error:
+ continue
+ assert response is not None
+ if response.status_code < 400:
+ discovered.append(f"{path} ({response.status_code})")
+ if discovered:
+ findings.append(
+ _make_finding(
+ tool="http-dir-enum",
+ cmdline=f"dir-enum --base {base_url} --level {profile.dir_enum}",
+ summary=f"Directory enumeration discovered {len(discovered)} path(s).",
+ raw_snippet=", ".join(discovered),
+ started_at=_utcnow(),
+ completed_at=_utcnow(),
+ )
+ )
+ else:
+ findings.append(
+ _make_finding(
+ tool="http-dir-enum",
+ cmdline=f"dir-enum --base {base_url} --level {profile.dir_enum}",
+ summary="Directory enumeration completed with no findings.",
+ raw_snippet="No accessible directories identified within safe wordlist.",
+ started_at=_utcnow(),
+ completed_at=_utcnow(),
+ )
+ )
return findings
-def serialize_results(data: Dict[str, object]) -> str:
+def serialize_results(data: dict[str, object]) -> str:
+ import json
+
return json.dumps(data, indent=2, sort_keys=True)
__all__ = [
- "DEFAULT_PORTS",
+ "DEFAULT_TCP_PORTS",
+ "DEFAULT_UDP_PORTS",
"HTTP_PORTS",
- "TOKEN_RATE",
- "TOKEN_CAPACITY",
- "HTTP_WORKERS",
- "CONNECT_TIMEOUT",
- "READ_TIMEOUT",
- "MAX_HTTP_RETRIES",
- "REDACTION_KEYS",
- "ScanConfig",
- "validate_port_list",
- "resolve_addresses",
- "tcp_connect_scan",
- "http_probe_services",
- "fetch_tls_certificate",
- "fetch_robots",
- "generate_findings",
- "check_security_headers",
+ "ReconProfile",
+ "active_dns_sweep",
+ "http_checks",
+ "passive_dns_collection",
+ "profile_for_evidence",
"serialize_results",
+ "tcp_syn_scan",
+ "udp_scan_pass",
+ "validate_port_list",
]
diff --git a/reconscript/scope.py b/reconscript/scope.py
index 0c20d7b..d06aa1a 100644
--- a/reconscript/scope.py
+++ b/reconscript/scope.py
@@ -1,13 +1,12 @@
-from __future__ import annotations
-
"""Target scope validation helpers for ReconScript."""
+from __future__ import annotations
+
import ipaddress
import os
import re
import socket
from dataclasses import dataclass
-from typing import Optional
__all__ = [
"ScopeValidation",
@@ -35,7 +34,7 @@ class ScopeValidation:
original: str
target: str
kind: str # "ip" or "hostname"
- resolved_ip: Optional[str] = None
+ resolved_ip: str | None = None
@property
def is_local(self) -> bool:
@@ -97,7 +96,7 @@ def validate_target(
target: str,
*,
expected_ip: str | None = None,
- allow_cidr: Optional[bool] = None,
+ allow_cidr: bool | None = None,
) -> ScopeValidation:
"""Validate a CLI/UI supplied target string."""
diff --git a/reconscript/templates/index.html b/reconscript/templates/index.html
index b6c80de..5e7a952 100644
--- a/reconscript/templates/index.html
+++ b/reconscript/templates/index.html
@@ -1,8 +1,10 @@
+
+
{% extends 'layout.html' %}
{% block content %}
Launch Scoped Scan
-
All scans are read-only. Provide a target host and optional parameters, then launch the scan.
+
All scans are read-only and operate with a conservative reconnaissance profile. Provide a single host to scan; network ranges require explicit authorization.
diff --git a/reconscript/templates/report_detail.html b/reconscript/templates/report_detail.html
index 53debf2..e330eac 100644
--- a/reconscript/templates/report_detail.html
+++ b/reconscript/templates/report_detail.html
@@ -1,26 +1,43 @@
+
+
{% extends 'layout.html' %}
{% block content %}
Report {{ report_id }}
-
Target: {{ report.target }}
-
Evidence level: {{ report.evidence_level }}
-
Report hash: {{ report.report_hash }}
- {% if report.consent_signed_by %}
-
Consent signer: {{ report.consent_signed_by }}
- {% endif %}
+
Target: {{ report.metadata.target }}
+
Resolved: {{ report.metadata.resolved_target }}
+
Evidence level: {{ report.metadata.evidence_level }}
+
Profile: {{ report.metadata.profile.name }} (TCP max {{ report.metadata.profile.max_tcp_ports }}, concurrency {{ report.metadata.profile.tcp_concurrency }})
+
Authorization confirmed: {{ 'yes' if report.metadata.consent_present else 'no' }}
Started: {{ report.started_at }} | Completed: {{ report.completed_at }}
-
Open ports
-
- {% for port in report.open_ports %}
- {{ port }}
- {% else %}
- No open ports detected.
- {% endfor %}
-
-
HTTP Checks
-
{{ report.http_checks | tojson(indent=2) }}
+
Scanned TCP ports: {{ report.metadata.scanned_tcp_ports|join(', ') }}
+
Scanned UDP ports: {{ report.metadata.scanned_udp_ports|join(', ') }}
+
+
Artifacts
+
{{ report.artifacts | tojson(indent=2) }}
+
Findings
-
{{ report.findings | tojson(indent=2) }}
+ {% if report.findings %}
+
+ {% else %}
+
No findings recorded.
+ {% endif %}
+
Download report JSON
{% endblock %}
diff --git a/reconscript/ui.py b/reconscript/ui.py
index b380337..e6e3a93 100644
--- a/reconscript/ui.py
+++ b/reconscript/ui.py
@@ -1,4 +1,6 @@
-"""Flask UI for ReconScript without authentication or consent requirements."""
+# Authorized testing only — do not scan targets without explicit permission.
+# This tool is non-intrusive by default and will not perform exploitation or credentialed checks.
+"""Flask UI for ReconScript with safe default scanning profiles."""
from __future__ import annotations
@@ -20,8 +22,8 @@
from .core import ReconError, run_recon
from .logging import configure_logging
-from .metrics import metrics_payload
from .report import ensure_results_dir, persist_report
+from .scanner import profile_for_evidence
from .scope import validate_target
LOGGER = logging.getLogger(__name__)
@@ -59,6 +61,8 @@ def health() -> tuple[str, int]:
@app.route("/metrics")
def metrics() -> Response:
+ from .metrics import metrics_payload
+
payload, content_type = metrics_payload()
if not payload:
return Response("", status=204)
@@ -67,22 +71,49 @@ def metrics() -> Response:
@app.route("/", methods=["GET", "POST"])
def index():
if request.method == "POST":
- target = request.form.get("target", "")
+ target = request.form.get("target", "").strip()
expected_ip = request.form.get("expected_ip") or None
hostname = request.form.get("hostname") or None
evidence_level = request.form.get("evidence_level", "low")
+ authorization = request.form.get("authorization") == "on"
ports_raw = request.form.get("ports", "")
ports = (
[int(p.strip()) for p in ports_raw.split(",") if p.strip()]
if ports_raw
else None
)
+
+ if not authorization:
+ flash(
+ (
+ "You must confirm 'I have authorization to scan this target' "
+ "before running a scan."
+ ),
+ "error",
+ )
+ return render_template("index.html")
+
try:
validate_target(target, expected_ip=expected_ip)
except Exception as exc: # pragma: no cover - input validation
flash(str(exc), "error")
return render_template("index.html")
+ try:
+ profile = profile_for_evidence(evidence_level)
+ except ValueError as exc:
+ flash(str(exc), "error")
+ return render_template("index.html")
+
+ network_like = (
+ any(symbol in target for symbol in ("/", "*")) or "," in target
+ )
+ if network_like:
+ LOGGER.info(
+ "ui.scan.wide_scope",
+ extra={"event": "ui.scan.wide_scope", "target": target},
+ )
+
try:
LOGGER.info(
"ui.scan.request",
@@ -93,6 +124,7 @@ def index():
"ports": ports,
"expected_ip": expected_ip,
"evidence_level": evidence_level,
+ "profile": profile.as_dict(),
},
)
report = run_recon(
@@ -101,6 +133,7 @@ def index():
ports=ports,
expected_ip=expected_ip,
evidence_level=evidence_level,
+ profile=profile,
)
except ReconError as exc:
flash(str(exc), "error")
@@ -121,7 +154,9 @@ def index():
"event": "ui.scan.completed",
"target": target,
"report_id": persisted.report_id,
- "open_ports": report.get("open_ports", []),
+ "open_ports": report.get("artifacts", {})
+ .get("tcp", {})
+ .get("open_ports", []),
},
)
return redirect(url_for("report_detail", report_id=persisted.report_id))
diff --git a/tests/integration/test_safe_profile_smoke.py b/tests/integration/test_safe_profile_smoke.py
new file mode 100644
index 0000000..31ac465
--- /dev/null
+++ b/tests/integration/test_safe_profile_smoke.py
@@ -0,0 +1,28 @@
+# Authorized testing only — do not scan targets without explicit permission.
+# This tool is non-intrusive by default and will not perform exploitation or credentialed checks.
+"""Smoke test ensuring a safe profile run persists a report."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from reconscript.core import run_recon
+from reconscript.report import persist_report
+
+
+@pytest.mark.integration
+def test_safe_profile_persists_report(tmp_path: Path) -> None:
+ report = run_recon(target="127.0.0.1", ports=[80], evidence_level="low")
+ persisted = persist_report(report)
+ try:
+ stored = json.loads(persisted.report_file.read_text(encoding="utf-8"))
+ assert stored["metadata"]["profile"]["name"].startswith("safe")
+ assert stored["findings"], "Expected findings in persisted report"
+ assert all(len(f["raw_snippet"]) <= 400 for f in stored["findings"])
+ finally:
+ import shutil
+
+ shutil.rmtree(persisted.base, ignore_errors=True)
diff --git a/tests/integration/test_scan_mock.py b/tests/integration/test_scan_mock.py
index 6af62a0..bbf6217 100644
--- a/tests/integration/test_scan_mock.py
+++ b/tests/integration/test_scan_mock.py
@@ -1,4 +1,6 @@
-from __future__ import annotations
+# Authorized testing only — do not scan targets without explicit permission.
+# This tool is non-intrusive by default and will not perform exploitation or credentialed checks.
+"""Integration tests for safe-profile scans against a local mock server."""
import http.server
import json
@@ -36,25 +38,30 @@ def mock_server() -> tuple[str, int]:
@pytest.mark.integration
-def test_scan_low_and_medium_levels(mock_server) -> None:
+def test_safe_profile_generates_required_metadata(mock_server) -> None:
host, port = mock_server
- report_low = run_recon(target=host, ports=[port], evidence_level="low")
- assert port in report_low["open_ports"]
+ report = run_recon(target=host, ports=[port], evidence_level="low")
- persisted = persist_report(report_low)
+ tcp_artifacts = report["artifacts"]["tcp"]
+ assert port in tcp_artifacts["open_ports"]
+
+ for finding in report["findings"]:
+ assert {
+ "tool",
+ "cmdline",
+ "summary",
+ "started_at",
+ "completed_at",
+ "raw_snippet",
+ }.issubset(finding)
+ assert len(finding["raw_snippet"]) <= 400
+
+ persisted = persist_report(report)
try:
report_data = json.loads(persisted.report_file.read_text(encoding="utf-8"))
stored_hash = report_data["report_hash"]
assert stored_hash == compute_report_hash(report_data)
-
- http_data = report_data["http_checks"][str(port)]
- assert "raw_request" not in http_data
- assert "raw_response" not in http_data
- assert http_data["headers"]["Set-Cookie"] == "[redacted]"
-
- report_medium = run_recon(target=host, ports=[port], evidence_level="medium")
- http_medium = report_medium["http_checks"][port]
- assert http_medium.get("screenshots") == []
- assert "raw_response" not in http_medium
+ assert report_data["metadata"]["profile"]["name"].startswith("safe")
+ assert all(len(f["raw_snippet"]) <= 400 for f in report_data["findings"])
finally:
shutil.rmtree(persisted.base, ignore_errors=True)
diff --git a/tests/integration/test_scan_scanme.py b/tests/integration/test_scan_scanme.py
index 540c064..0443195 100644
--- a/tests/integration/test_scan_scanme.py
+++ b/tests/integration/test_scan_scanme.py
@@ -1,3 +1,5 @@
+# Authorized testing only — do not scan targets without explicit permission.
+# This tool is non-intrusive by default and will not perform exploitation or credentialed checks.
from __future__ import annotations
import os
@@ -14,4 +16,5 @@
)
def test_scan_scanme() -> None:
report = run_recon(target="scanme.nmap.org", evidence_level="low")
- assert "open_ports" in report
+ assert "artifacts" in report
+ assert "findings" in report
diff --git a/tests/test_core_safe_profile.py b/tests/test_core_safe_profile.py
new file mode 100644
index 0000000..4ac25b4
--- /dev/null
+++ b/tests/test_core_safe_profile.py
@@ -0,0 +1,33 @@
+# Authorized testing only — do not scan targets without explicit permission.
+# This tool is non-intrusive by default and will not perform exploitation or credentialed checks.
+"""Tests for the safe default recon workflow."""
+
+from __future__ import annotations
+
+from reconscript.core import run_recon
+from reconscript.scanner import profile_for_evidence
+
+
+def test_run_recon_safe_defaults() -> None:
+ profile = profile_for_evidence("low")
+ report = run_recon(
+ target="127.0.0.1", ports=[80], evidence_level="low", profile=profile
+ )
+
+ metadata = report["metadata"]
+ assert metadata["profile"]["allow_credentialed"] is False
+ assert metadata["profile"]["dir_enum"] in {"low", "medium", "none"}
+
+ findings = report["findings"]
+ assert isinstance(findings, list)
+ assert findings, "Expected at least one finding even if target is quiet"
+ for finding in findings:
+ assert {
+ "tool",
+ "cmdline",
+ "summary",
+ "started_at",
+ "completed_at",
+ "raw_snippet",
+ }.issubset(finding.keys())
+ assert len(finding["raw_snippet"]) <= 400
diff --git a/tests/test_reporters.py b/tests/test_reporters.py
index 619f5a8..a969b00 100644
--- a/tests/test_reporters.py
+++ b/tests/test_reporters.py
@@ -7,15 +7,32 @@
def sample_report() -> dict[str, object]:
return {
- "target": "example.com",
- "hostname": "example.com",
- "ports": [80, 443],
- "open_ports": [80],
+ "metadata": {
+ "target": "example.com",
+ "hostname": "example.com",
+ "scanned_tcp_ports": [80, 443],
+ "profile": {
+ "name": "safe-default",
+ "max_tcp_ports": 100,
+ "tcp_concurrency": 10,
+ "dir_enum": "low",
+ "allow_credentialed": False,
+ },
+ },
+ "artifacts": {
+ "http": {
+ "services": {80: {"status_code": 200, "url": "http://example.com"}}
+ },
+ "tcp": {"open_ports": [80]},
+ },
"findings": [
{
- "port": 80,
- "issue": "missing_security_headers",
- "details": {"header": "Strict-Transport-Security"},
+ "tool": "http-check",
+ "cmdline": "http --url http://example.com",
+ "summary": "HTTP responded with 200 OK without Strict-Transport-Security.",
+ "raw_snippet": "Header Strict-Transport-Security missing",
+ "started_at": "2024-01-01T00:00:00Z",
+ "completed_at": "2024-01-01T00:00:01Z",
}
],
"runtime": {"started_at": "2024-01-01T00:00:00Z"},
@@ -27,7 +44,7 @@ def test_render_markdown_fallback_contains_sections() -> None:
assert "# ReconScript Report" in markdown
assert "## Findings" in markdown
assert "## Recommendations" in markdown
- assert "Strict-Transport-Security" in markdown
+ assert "http --url http://example.com" in markdown
def test_write_report_markdown(tmp_path: Path) -> None:
diff --git a/tests/test_scanner_profile.py b/tests/test_scanner_profile.py
new file mode 100644
index 0000000..502840d
--- /dev/null
+++ b/tests/test_scanner_profile.py
@@ -0,0 +1,37 @@
+# Authorized testing only — do not scan targets without explicit permission.
+# This tool is non-intrusive by default and will not perform exploitation or credentialed checks.
+"""Unit tests for ReconProfile defaults and helpers."""
+
+from __future__ import annotations
+
+import pytest
+
+from reconscript.scanner import ReconProfile, profile_for_evidence
+
+
+def test_recon_profile_defaults() -> None:
+ profile = ReconProfile(name="test-profile")
+ assert profile.max_tcp_ports == 100
+ assert profile.tcp_concurrency == 10
+ assert profile.dir_enum == "low"
+ assert profile.timeouts["http"] == pytest.approx(5.0)
+ assert profile.rate_limits["tcp"] == (5.0, 5.0)
+
+
+def test_profile_for_evidence_levels() -> None:
+ low = profile_for_evidence("low")
+ medium = profile_for_evidence("medium")
+ high = profile_for_evidence("high")
+
+ assert low.dir_enum == "low"
+ assert medium.dir_enum == "medium"
+ assert high.dir_enum == "none"
+ assert high.max_tcp_ports > medium.max_tcp_ports >= low.max_tcp_ports
+
+
+def test_limit_ports_enforces_cap() -> None:
+ profile = ReconProfile(name="cap-test", max_tcp_ports=3)
+ ports = profile.limit_ports([80, 443, 8080, 8443])
+ assert len(ports) == 3
+ assert ports[0] == 80
+ assert 443 in ports
diff --git a/tests/test_ui_safe_profile.py b/tests/test_ui_safe_profile.py
new file mode 100644
index 0000000..d83476a
--- /dev/null
+++ b/tests/test_ui_safe_profile.py
@@ -0,0 +1,100 @@
+# Authorized testing only — do not scan targets without explicit permission.
+# This tool is non-intrusive by default and will not perform exploitation or credentialed checks.
+"""UI tests ensuring authorization gating and report persistence."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from reconscript.ui import create_app
+
+
+@pytest.fixture()
+def ui_app(monkeypatch, tmp_path: Path):
+ monkeypatch.setenv("RESULTS_DIR", str(tmp_path))
+
+ saved_report: dict[str, Any] = {
+ "metadata": {
+ "target": "127.0.0.1",
+ "resolved_target": "127.0.0.1",
+ "hostname": None,
+ "version": "test",
+ "evidence_level": "low",
+ "profile": {
+ "name": "test",
+ "max_tcp_ports": 10,
+ "tcp_concurrency": 2,
+ "udp_pass": True,
+ "dns_active": True,
+ "dns_passive": True,
+ "dir_enum": "low",
+ "allow_credentialed": False,
+ "timeouts": {},
+ "rate_limits": {},
+ },
+ "consent_present": False,
+ "scanned_tcp_ports": [80],
+ "scanned_udp_ports": [53],
+ },
+ "artifacts": {},
+ "findings": [
+ {
+ "tool": "test",
+ "cmdline": "cmd --arg",
+ "summary": "ok",
+ "started_at": "2024-01-01T00:00:00Z",
+ "completed_at": "2024-01-01T00:00:01Z",
+ "raw_snippet": "evidence",
+ }
+ ],
+ }
+
+ def fake_run_recon(**_: Any) -> dict[str, Any]:
+ return json.loads(json.dumps(saved_report))
+
+ monkeypatch.setattr("reconscript.ui.run_recon", fake_run_recon)
+
+ app = create_app()
+ app.config.update(TESTING=True)
+ return app
+
+
+def test_ui_requires_authorization(ui_app) -> None:
+ client = ui_app.test_client()
+ response = client.post(
+ "/",
+ data={
+ "target": "127.0.0.1",
+ "evidence_level": "low",
+ # No authorization checkbox
+ },
+ follow_redirects=True,
+ )
+ assert response.status_code == 200
+ assert "I have authorization" in response.get_data(as_text=True)
+
+
+def test_ui_runs_scan_without_consent(ui_app, tmp_path: Path) -> None:
+ client = ui_app.test_client()
+ response = client.post(
+ "/",
+ data={
+ "target": "127.0.0.1",
+ "evidence_level": "low",
+ "authorization": "on",
+ },
+ follow_redirects=False,
+ )
+ assert response.status_code == 302
+
+ results_dir = Path(ui_app.config["UPLOAD_FOLDER"]).parent
+ report_dirs = [path for path in results_dir.iterdir() if path.is_dir()]
+ assert report_dirs, "Expected a persisted report directory"
+ report_file = report_dirs[0] / "report.json"
+ stored = json.loads(report_file.read_text(encoding="utf-8"))
+ assert stored["findings"][0]["cmdline"] == "cmd --arg"
+ assert len(stored["findings"][0]["raw_snippet"]) <= 400