95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
"""Full deterministic pipeline on canned fixtures — zero network, no live scan.
|
|
|
|
This is the offline demo and the proof that recon->normalize->ground->report works
|
|
without any model or external call.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from recon_triage.orchestrator import build_report_from_fixtures
|
|
from recon_triage.report import markdown
|
|
from recon_triage.schema import ReconReport
|
|
from recon_triage.scope import Scope
|
|
from recon_triage.triage.llm import run_triage
|
|
from recon_triage.triage.ranking import deterministic_triage
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def _build(scope=None, search_fn=None) -> ReconReport:
|
|
return build_report_from_fixtures(
|
|
FIXTURES, run_id="test", scope=scope, search_fn=search_fn, enable_nuclei=True
|
|
)
|
|
|
|
|
|
def test_offline_run_schema_valid(fixture_search_fn):
|
|
report = _build(search_fn=fixture_search_fn)
|
|
# Re-validate to prove schema-validity.
|
|
ReconReport.model_validate_json(report.model_dump_json())
|
|
assert report.hosts
|
|
assert report.nuclei_findings
|
|
|
|
|
|
def test_offline_run_merges_hosts(fixture_search_fn):
|
|
report = _build(search_fn=fixture_search_fn)
|
|
by_name = {h.hostname: h for h in report.hosts}
|
|
api = by_name["api.example.com"]
|
|
# ports merged from naabu + nmap; service+http merged
|
|
ports = {p.number for p in api.ports}
|
|
assert {22, 443}.issubset(ports)
|
|
svc443 = next(p.service for p in api.ports if p.number == 443)
|
|
assert svc443.product == "Apache httpd"
|
|
assert svc443.version == "2.4.49"
|
|
assert svc443.http is not None
|
|
assert svc443.http.status == 200
|
|
|
|
|
|
def test_offline_run_grounds_exploitdb(fixture_search_fn):
|
|
report = _build(search_fn=fixture_search_fn)
|
|
api = next(h for h in report.hosts if h.hostname == "api.example.com")
|
|
svc443 = next(p.service for p in api.ports if p.number == 443)
|
|
assert {m.edb_id for m in svc443.exploitdb_candidates} == {"50383", "50406"}
|
|
assert all(not m.verified for m in svc443.exploitdb_candidates)
|
|
|
|
|
|
def test_offline_run_scope_drops_out_of_scope(fixture_search_fn):
|
|
scope = Scope.load(FIXTURES / "scope.yaml")
|
|
report = _build(scope=scope, search_fn=fixture_search_fn)
|
|
names = {h.hostname for h in report.hosts}
|
|
assert "external.notmine.com" not in names
|
|
assert "api.example.com" in names
|
|
|
|
|
|
def test_offline_run_markdown_renders(fixture_search_fn):
|
|
report = _build(search_fn=fixture_search_fn)
|
|
report.triage = deterministic_triage(report)
|
|
md = markdown.render(report)
|
|
assert "# recon-triage report" in md
|
|
assert "Authorized use only" in md
|
|
assert "EDB-50383" in md
|
|
assert "Triage priorities" in md
|
|
|
|
|
|
def test_deterministic_triage_without_llm(fixture_search_fn, monkeypatch):
|
|
monkeypatch.delenv("LLM_BASE_URL", raising=False)
|
|
report = _build(search_fn=fixture_search_fn)
|
|
triage = run_triage(report)
|
|
assert triage.generated_by == "deterministic"
|
|
assert triage.prioritized_findings
|
|
# High-severity nuclei finding should rank at the top.
|
|
assert triage.prioritized_findings[0].severity == "high"
|
|
|
|
|
|
def test_triage_refs_are_real(fixture_search_fn):
|
|
report = _build(search_fn=fixture_search_fn)
|
|
triage = deterministic_triage(report)
|
|
from recon_triage.triage.ranking import collect_identifiers
|
|
|
|
valid = collect_identifiers(report)
|
|
for item in triage.prioritized_findings:
|
|
assert item.evidence_refs
|
|
for ref in item.evidence_refs:
|
|
assert ref in valid or any(v in ref for v in valid)
|