Skip to content

Finalize safe recon reporting #35

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
danielpmadden merged 1 commit into from
Oct 26, 2025
Merged

Finalize safe recon reporting #35

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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).

Expand Down
221 changes: 119 additions & 102 deletions reconscript/core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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

Expand All @@ -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
Expand All @@ -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
Expand All @@ -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,
Expand All @@ -70,132 +66,154 @@ 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."""

evidence_level = evidence_level.lower()
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
)
record_scan_completed(scope.target, 0.0, 0)
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:
Expand All @@ -219,5 +237,4 @@ def run_recon(
__all__ = [
"run_recon",
"ReconError",
"check_security_headers",
]
Loading
Loading