test: offline suite + committed fixtures for every stage
This commit is contained in:
@@ -0,0 +1,40 @@
|
||||
"""Shared test fixtures and the offline searchsploit search function.
|
||||
|
||||
The injectable ``fixture_search_fn`` reads canned ``searchsploit --json`` output
|
||||
from ``tests/fixtures/searchsploit/index.json`` so the whole grounding path is
|
||||
exercised with zero network and no searchsploit binary.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from recon_triage.schema import ExploitDBMatch
|
||||
from recon_triage.tools import searchsploit
|
||||
|
||||
FIXTURES = Path(__file__).parent / "fixtures"
|
||||
|
||||
|
||||
def make_fixture_search_fn():
|
||||
index = json.loads((FIXTURES / "searchsploit" / "index.json").read_text())
|
||||
|
||||
def _search(term: str) -> list[ExploitDBMatch]:
|
||||
doc = index.get(term)
|
||||
if doc is None:
|
||||
return []
|
||||
return searchsploit.normalize(json.dumps(doc), query=term)
|
||||
|
||||
return _search
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fixtures_dir() -> Path:
|
||||
return FIXTURES
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fixture_search_fn():
|
||||
return make_fixture_search_fn()
|
||||
+4
@@ -0,0 +1,4 @@
|
||||
{"host":"api.example.com","a":["93.184.216.34"],"cname":[],"status_code":"NOERROR"}
|
||||
{"host":"www.example.com","a":["93.184.216.35"],"cname":[],"status_code":"NOERROR"}
|
||||
{"host":"admin.example.com","a":["93.184.216.36"],"cname":["internal.example.com"],"status_code":"NOERROR"}
|
||||
{"host":"external.notmine.com","a":["198.51.100.7"],"cname":[],"status_code":"NOERROR"}
|
||||
+4
@@ -0,0 +1,4 @@
|
||||
{"input":"api.example.com","host":"93.184.216.34","port":"443","url":"https://api.example.com","scheme":"https","status_code":200,"title":"API Gateway","webserver":"Apache/2.4.49","tech":["Apache HTTP Server:2.4.49"],"tls":{"subject_cn":"api.example.com","issuer_org":["Let's Encrypt"],"not_after":"2026-09-01T00:00:00Z"}}
|
||||
{"input":"www.example.com","host":"93.184.216.35","port":"80","url":"http://www.example.com","scheme":"http","status_code":301,"title":"Moved","webserver":"nginx/1.18.0","tech":["Nginx:1.18.0"]}
|
||||
{"input":"www.example.com","host":"93.184.216.35","port":"443","url":"https://www.example.com","scheme":"https","status_code":200,"title":"Example Corp","webserver":"nginx/1.18.0","tech":["Nginx:1.18.0"],"tls":{"subject_cn":"www.example.com","issuer_cn":"R3","not_after":"2026-08-01T00:00:00Z"}}
|
||||
{"input":"admin.example.com","host":"93.184.216.36","port":"8080","url":"http://admin.example.com:8080","scheme":"http","status_code":200,"title":"Tomcat Manager","webserver":"Apache-Coyote/1.1","tech":["Apache Tomcat:9.0.30"]}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{"host":"api.example.com","ip":"93.184.216.34","port":443,"protocol":"tcp"}
|
||||
{"host":"api.example.com","ip":"93.184.216.34","port":22,"protocol":"tcp"}
|
||||
{"host":"www.example.com","ip":"93.184.216.35","port":80,"protocol":"tcp"}
|
||||
{"host":"www.example.com","ip":"93.184.216.35","port":443,"protocol":"tcp"}
|
||||
{"host":"admin.example.com","ip":"93.184.216.36","port":8080,"protocol":"tcp"}
|
||||
{"host":"external.notmine.com","ip":"198.51.100.7","port":80,"protocol":"tcp"}
|
||||
Vendored
+59
@@ -0,0 +1,59 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!DOCTYPE nmaprun>
|
||||
<nmaprun scanner="nmap" args="nmap -sV -sT -oX - -Pn -p 22,80,443,8080 93.184.216.34 93.184.216.35 93.184.216.36" version="7.94">
|
||||
<host>
|
||||
<address addr="93.184.216.34" addrtype="ipv4"/>
|
||||
<hostnames>
|
||||
<hostname name="api.example.com" type="user"/>
|
||||
</hostnames>
|
||||
<ports>
|
||||
<port protocol="tcp" portid="22">
|
||||
<state state="open" reason="syn-ack"/>
|
||||
<service name="ssh" product="OpenSSH" version="8.2p1 Ubuntu 4ubuntu0.5" method="probed" conf="10">
|
||||
<cpe>cpe:/a:openbsd:openssh:8.2p1</cpe>
|
||||
</service>
|
||||
</port>
|
||||
<port protocol="tcp" portid="443">
|
||||
<state state="open" reason="syn-ack"/>
|
||||
<service name="http" product="Apache httpd" version="2.4.49" method="probed" conf="10" tunnel="ssl">
|
||||
<cpe>cpe:/a:apache:http_server:2.4.49</cpe>
|
||||
</service>
|
||||
</port>
|
||||
</ports>
|
||||
</host>
|
||||
<host>
|
||||
<address addr="93.184.216.35" addrtype="ipv4"/>
|
||||
<hostnames>
|
||||
<hostname name="www.example.com" type="user"/>
|
||||
</hostnames>
|
||||
<ports>
|
||||
<port protocol="tcp" portid="80">
|
||||
<state state="open" reason="syn-ack"/>
|
||||
<service name="http" product="nginx" version="1.18.0" method="probed" conf="10">
|
||||
<cpe>cpe:/a:igor_sysoev:nginx:1.18.0</cpe>
|
||||
</service>
|
||||
</port>
|
||||
<port protocol="tcp" portid="443">
|
||||
<state state="open" reason="syn-ack"/>
|
||||
<service name="http" product="nginx" version="1.18.0" method="probed" conf="10" tunnel="ssl"/>
|
||||
</port>
|
||||
</ports>
|
||||
</host>
|
||||
<host>
|
||||
<address addr="93.184.216.36" addrtype="ipv4"/>
|
||||
<hostnames>
|
||||
<hostname name="admin.example.com" type="user"/>
|
||||
</hostnames>
|
||||
<ports>
|
||||
<port protocol="tcp" portid="8080">
|
||||
<state state="open" reason="syn-ack"/>
|
||||
<service name="http-proxy" product="Apache Tomcat" version="9.0.30" method="probed" conf="10">
|
||||
<cpe>cpe:/a:apache:tomcat:9.0.30</cpe>
|
||||
</service>
|
||||
</port>
|
||||
</ports>
|
||||
</host>
|
||||
<runstats>
|
||||
<finished time="1700000000" timestr="Tue Nov 14 00:00:00 2023" elapsed="12.34" exit="success"/>
|
||||
</runstats>
|
||||
</nmaprun>
|
||||
+3
@@ -0,0 +1,3 @@
|
||||
{"template-id":"apache-detect","info":{"name":"Apache Detection","severity":"info","reference":["https://httpd.apache.org/"]},"host":"api.example.com","matched-at":"https://api.example.com"}
|
||||
{"template-id":"tomcat-manager-exposed","info":{"name":"Apache Tomcat Manager Exposed","severity":"high","reference":["https://tomcat.apache.org/"]},"host":"admin.example.com","matched-at":"http://admin.example.com:8080/manager/html"}
|
||||
{"template-id":"tls-version","info":{"name":"TLS Version Detection","severity":"info","reference":[]},"host":"www.example.com","matched-at":"www.example.com:443"}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
in_scope_domains:
|
||||
- example.com
|
||||
in_scope_cidrs:
|
||||
- 93.184.216.0/24
|
||||
out_of_scope:
|
||||
- internal-only.example.com
|
||||
@@ -0,0 +1,33 @@
|
||||
{
|
||||
"Apache httpd 2.4.49": {
|
||||
"RESULTS_EXPLOIT": [
|
||||
{
|
||||
"Title": "Apache HTTP Server 2.4.49 - Path Traversal & Remote Code Execution (RCE)",
|
||||
"EDB-ID": "50383",
|
||||
"Date_Published": "2021-10-07",
|
||||
"Path": "/usr/share/exploitdb/exploits/multiple/webapps/50383.sh",
|
||||
"URL": "https://www.exploit-db.com/exploits/50383",
|
||||
"Type": "webapps",
|
||||
"Platform": "multiple"
|
||||
},
|
||||
{
|
||||
"Title": "Apache HTTP Server 2.4.50 - Path Traversal & Remote Code Execution (RCE)",
|
||||
"EDB-ID": "50406",
|
||||
"Date_Published": "2021-10-08",
|
||||
"Path": "/usr/share/exploitdb/exploits/multiple/webapps/50406.py",
|
||||
"URL": "https://www.exploit-db.com/exploits/50406",
|
||||
"Type": "webapps",
|
||||
"Platform": "multiple"
|
||||
}
|
||||
]
|
||||
},
|
||||
"Apache Tomcat 9.0.30": {
|
||||
"RESULTS_EXPLOIT": []
|
||||
},
|
||||
"OpenSSH 8.2p1 Ubuntu 4ubuntu0.5": {
|
||||
"RESULTS_EXPLOIT": []
|
||||
},
|
||||
"nginx 1.18.0": {
|
||||
"RESULTS_EXPLOIT": []
|
||||
}
|
||||
}
|
||||
+4
@@ -0,0 +1,4 @@
|
||||
{"host":"api.example.com","input":"example.com","source":["crtsh"]}
|
||||
{"host":"www.example.com","input":"example.com","source":["dnsdumpster"]}
|
||||
{"host":"admin.example.com","input":"example.com","source":["crtsh"]}
|
||||
{"host":"external.notmine.com","input":"example.com","source":["crtsh"]}
|
||||
@@ -0,0 +1,68 @@
|
||||
"""Exploit-DB matcher: parses real searchsploit JSON, never fabricates, fails soft."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from recon_triage.grounding.exploitdb import ground_report
|
||||
from recon_triage.schema import Host, Port, ReconReport, Service
|
||||
from recon_triage.tools import searchsploit
|
||||
|
||||
FIXTURES = Path(__file__).parent / "fixtures"
|
||||
|
||||
|
||||
def _index() -> dict:
|
||||
return json.loads((FIXTURES / "searchsploit" / "index.json").read_text())
|
||||
|
||||
|
||||
def test_normalize_parses_results():
|
||||
doc = _index()["Apache httpd 2.4.49"]
|
||||
matches = searchsploit.normalize(json.dumps(doc), query="Apache httpd 2.4.49")
|
||||
assert {m.edb_id for m in matches} == {"50383", "50406"}
|
||||
assert all(m.verified is False for m in matches)
|
||||
assert all(m.source == "exploit-db" for m in matches)
|
||||
assert matches[0].url.endswith("/50383")
|
||||
|
||||
|
||||
def test_normalize_empty_results():
|
||||
doc = _index()["nginx 1.18.0"]
|
||||
assert searchsploit.normalize(json.dumps(doc)) == []
|
||||
|
||||
|
||||
def test_normalize_handles_garbage():
|
||||
assert searchsploit.normalize("") == []
|
||||
assert searchsploit.normalize("not json") == []
|
||||
|
||||
|
||||
def test_ground_report_attaches_only_returned(fixture_search_fn):
|
||||
report = ReconReport(
|
||||
run_id="t",
|
||||
hosts=[
|
||||
Host(
|
||||
hostname="api.example.com",
|
||||
ports=[
|
||||
Port(number=443, service=Service(product="Apache httpd", version="2.4.49")),
|
||||
Port(number=22, service=Service(product="OpenSSH", version="8.2p1 Ubuntu 4ubuntu0.5")),
|
||||
],
|
||||
)
|
||||
],
|
||||
)
|
||||
attached = ground_report(report, fixture_search_fn)
|
||||
assert attached == 2 # only the Apache service yields 2 candidates
|
||||
apache = report.hosts[0].ports[0].service
|
||||
ssh = report.hosts[0].ports[1].service
|
||||
assert {m.edb_id for m in apache.exploitdb_candidates} == {"50383", "50406"}
|
||||
assert ssh.exploitdb_candidates == [] # nothing returned -> empty, not fabricated
|
||||
|
||||
|
||||
def test_ground_report_failsoft_on_search_error():
|
||||
def boom(term: str):
|
||||
raise RuntimeError("searchsploit blew up")
|
||||
|
||||
report = ReconReport(
|
||||
run_id="t",
|
||||
hosts=[Host(hostname="h", ports=[Port(number=443, service=Service(product="X", version="1"))])],
|
||||
)
|
||||
attached = ground_report(report, boom) # must not raise
|
||||
assert attached == 0
|
||||
@@ -0,0 +1,75 @@
|
||||
"""LLM triage stage — offline tests of validation, acceptance gate, and skip."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from recon_triage.orchestrator import build_report_from_fixtures
|
||||
from recon_triage.triage import llm
|
||||
from recon_triage.triage.ranking import collect_identifiers
|
||||
|
||||
FIXTURES = Path(__file__).parent / "fixtures"
|
||||
|
||||
|
||||
def _report(fixture_search_fn):
|
||||
return build_report_from_fixtures(
|
||||
FIXTURES, run_id="t", search_fn=fixture_search_fn, enable_nuclei=True
|
||||
)
|
||||
|
||||
|
||||
def test_skips_when_unconfigured(monkeypatch, fixture_search_fn):
|
||||
monkeypatch.delenv("LLM_BASE_URL", raising=False)
|
||||
assert not llm.is_configured()
|
||||
triage = llm.run_triage(_report(fixture_search_fn))
|
||||
assert triage.generated_by == "deterministic"
|
||||
|
||||
|
||||
def test_extract_json_handles_fences():
|
||||
assert llm._extract_json('```json\n{"a": 1}\n```') == {"a": 1}
|
||||
assert llm._extract_json('prose {"a": 2} more') == {"a": 2}
|
||||
assert llm._extract_json("not json") is None
|
||||
|
||||
|
||||
def test_acceptance_gate_strips_invented_refs(fixture_search_fn):
|
||||
from recon_triage.schema import TriageItem, TriageReport
|
||||
|
||||
report = _report(fixture_search_fn)
|
||||
valid = collect_identifiers(report)
|
||||
real_id = next(iter(valid))
|
||||
triage = TriageReport(
|
||||
prioritized_findings=[
|
||||
TriageItem(summary="real", rationale="r", evidence_refs=[real_id], suggested_next_step=f"look at {real_id}"),
|
||||
TriageItem(summary="invented", rationale="r", evidence_refs=["EDB-99999999"], suggested_next_step="run made-up-exploit"),
|
||||
TriageItem(summary="mixed", rationale="r", evidence_refs=[real_id, "CVE-FAKE"], suggested_next_step="do CVE-FAKE"),
|
||||
]
|
||||
)
|
||||
cleaned = llm._enforce_acceptance(triage, valid)
|
||||
summaries = {i.summary for i in cleaned.prioritized_findings}
|
||||
# invented item (no valid ref at all) is dropped entirely
|
||||
assert "invented" not in summaries
|
||||
assert "real" in summaries
|
||||
# mixed item kept but bad ref stripped and bad next-step cleared
|
||||
mixed = next(i for i in cleaned.prioritized_findings if i.summary == "mixed")
|
||||
assert "CVE-FAKE" not in mixed.evidence_refs
|
||||
assert mixed.suggested_next_step is None
|
||||
|
||||
|
||||
def test_llm_path_with_mocked_transport(monkeypatch, fixture_search_fn):
|
||||
"""Exercise the real run_triage LLM branch with a stubbed _call_llm (no network)."""
|
||||
report = _report(fixture_search_fn)
|
||||
valid = collect_identifiers(report)
|
||||
real_id = next(r for r in valid if ":" in r) # a host:port id
|
||||
|
||||
def fake_call(report_json, **kwargs):
|
||||
return (
|
||||
'{"prioritized_findings": [{"summary": "test", "rationale": "r", '
|
||||
f'"severity": "high", "evidence_refs": ["{real_id}"], '
|
||||
f'"suggested_next_step": "review {real_id}", "confidence": 0.9}}]}}'
|
||||
)
|
||||
|
||||
monkeypatch.setenv("LLM_BASE_URL", "http://fake:11434/v1")
|
||||
monkeypatch.setattr(llm, "_call_llm", fake_call)
|
||||
triage = llm.run_triage(report)
|
||||
assert triage.generated_by == "llm"
|
||||
assert triage.prioritized_findings[0].summary == "test"
|
||||
assert triage.prioritized_findings[0].evidence_refs == [real_id]
|
||||
@@ -0,0 +1,71 @@
|
||||
"""Each tool normalizer is exercised against its committed fixture."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from recon_triage.schema import Severity
|
||||
from recon_triage.tools import dnsx, httpx, naabu, nmap, nuclei, subfinder
|
||||
|
||||
FIXTURES = Path(__file__).parent / "fixtures"
|
||||
|
||||
|
||||
def _read(name: str) -> str:
|
||||
return (FIXTURES / name).read_text()
|
||||
|
||||
|
||||
def test_subfinder_normalize():
|
||||
hosts = subfinder.normalize(_read("subfinder.jsonl"))
|
||||
assert "api.example.com" in hosts
|
||||
assert "external.notmine.com" in hosts # scope filtering happens later, not here
|
||||
assert len(hosts) == len(set(hosts)) # de-duped
|
||||
|
||||
|
||||
def test_dnsx_normalize():
|
||||
recs = dnsx.normalize(_read("dnsx.jsonl"))
|
||||
by_host = {r.host: r for r in recs}
|
||||
assert by_host["api.example.com"].ips == ["93.184.216.34"]
|
||||
assert by_host["admin.example.com"].cname == ["internal.example.com"]
|
||||
|
||||
|
||||
def test_naabu_normalize():
|
||||
hits = naabu.normalize(_read("naabu.jsonl"))
|
||||
ports = {(h.host, h.port) for h in hits}
|
||||
assert ("api.example.com", 443) in ports
|
||||
assert ("api.example.com", 22) in ports
|
||||
assert all(isinstance(h.port, int) for h in hits)
|
||||
|
||||
|
||||
def test_nmap_normalize_captures_product_version():
|
||||
hosts = nmap.normalize(_read("nmap.xml"))
|
||||
by_ip = {h.ip: h for h in hosts}
|
||||
api = by_ip["93.184.216.34"]
|
||||
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.cpe == "cpe:/a:apache:http_server:2.4.49"
|
||||
# multiple hosts and multiple ports handled (list coercion)
|
||||
assert len(hosts) == 3
|
||||
|
||||
|
||||
def test_nmap_handles_empty():
|
||||
assert nmap.normalize("") == []
|
||||
assert nmap.normalize("not xml at all") == []
|
||||
|
||||
|
||||
def test_httpx_normalize():
|
||||
probes = httpx.normalize(_read("httpx.jsonl"))
|
||||
api = next(p for p in probes if p.host == "api.example.com")
|
||||
assert api.info.status == 200
|
||||
assert api.info.title == "API Gateway"
|
||||
assert api.info.webserver == "Apache/2.4.49"
|
||||
assert api.info.tls is not None
|
||||
assert api.port == 443
|
||||
|
||||
|
||||
def test_nuclei_normalize_severity():
|
||||
findings = nuclei.normalize(_read("nuclei.jsonl"))
|
||||
by_id = {f.template_id: f for f in findings}
|
||||
assert by_id["tomcat-manager-exposed"].severity == Severity.HIGH.value
|
||||
assert by_id["apache-detect"].severity == Severity.INFO.value
|
||||
assert by_id["tomcat-manager-exposed"].reference # references preserved
|
||||
@@ -0,0 +1,94 @@
|
||||
"""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)
|
||||
@@ -0,0 +1,70 @@
|
||||
"""Schema contract tests: round-trip, JSON-schema export, query-term logic."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
from recon_triage.schema import (
|
||||
ExploitDBMatch,
|
||||
Host,
|
||||
Port,
|
||||
ReconReport,
|
||||
Service,
|
||||
export_json_schema,
|
||||
)
|
||||
|
||||
|
||||
def test_recon_report_minimal_valid():
|
||||
r = ReconReport(run_id="t1")
|
||||
assert r.run_id == "t1"
|
||||
assert r.hosts == []
|
||||
assert "Authorized use only" in r.notice
|
||||
|
||||
|
||||
def test_round_trip_json():
|
||||
r = ReconReport(
|
||||
run_id="t1",
|
||||
hosts=[
|
||||
Host(
|
||||
hostname="api.example.com",
|
||||
ips=["1.2.3.4"],
|
||||
ports=[
|
||||
Port(
|
||||
number=443,
|
||||
service=Service(
|
||||
name="http",
|
||||
product="Apache httpd",
|
||||
version="2.4.49",
|
||||
exploitdb_candidates=[
|
||||
ExploitDBMatch(edb_id="50383", title="x", path="p")
|
||||
],
|
||||
),
|
||||
)
|
||||
],
|
||||
)
|
||||
],
|
||||
)
|
||||
dumped = r.model_dump_json()
|
||||
restored = ReconReport.model_validate_json(dumped)
|
||||
assert restored.hosts[0].ports[0].service.product == "Apache httpd"
|
||||
assert restored.hosts[0].ports[0].service.exploitdb_candidates[0].verified is False
|
||||
|
||||
|
||||
def test_exploitdb_match_always_unverified():
|
||||
m = ExploitDBMatch(edb_id="1", title="t", path="p")
|
||||
assert m.verified is False
|
||||
assert m.source == "exploit-db"
|
||||
|
||||
|
||||
def test_query_term():
|
||||
assert Service(product="Apache httpd", version="2.4.49").query_term() == "Apache httpd 2.4.49"
|
||||
assert Service(product="nginx").query_term() == "nginx"
|
||||
assert Service(name="http").query_term() is None
|
||||
|
||||
|
||||
def test_export_json_schema(tmp_path):
|
||||
p = export_json_schema(tmp_path)
|
||||
assert p.exists()
|
||||
doc = json.loads(p.read_text())
|
||||
assert "properties" in doc and "hosts" in doc["properties"]
|
||||
assert (tmp_path / "triage_report.schema.json").exists()
|
||||
@@ -0,0 +1,57 @@
|
||||
"""Scope gating: in-scope kept, out-of-scope dropped, exclusions honoured."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from recon_triage.scope import Scope
|
||||
|
||||
FIXTURES = Path(__file__).parent / "fixtures"
|
||||
|
||||
|
||||
def _scope() -> Scope:
|
||||
return Scope.load(FIXTURES / "scope.yaml")
|
||||
|
||||
|
||||
def test_in_scope_domain_and_subdomains():
|
||||
sc = _scope()
|
||||
assert sc.is_in_scope("example.com")
|
||||
assert sc.is_in_scope("api.example.com")
|
||||
assert sc.is_in_scope("deep.api.example.com")
|
||||
|
||||
|
||||
def test_out_of_scope_domain_dropped():
|
||||
sc = _scope()
|
||||
assert not sc.is_in_scope("external.notmine.com")
|
||||
assert not sc.is_in_scope("notexample.com")
|
||||
|
||||
|
||||
def test_exclusion_precedence():
|
||||
sc = _scope()
|
||||
# internal-only.example.com is under example.com but explicitly excluded.
|
||||
assert not sc.is_in_scope("internal-only.example.com")
|
||||
|
||||
|
||||
def test_cidr_matching():
|
||||
sc = _scope()
|
||||
assert sc.is_in_scope("93.184.216.34")
|
||||
assert not sc.is_in_scope("198.51.100.7")
|
||||
|
||||
|
||||
def test_filter_drops_out_of_scope():
|
||||
sc = _scope()
|
||||
given = ["api.example.com", "external.notmine.com", "93.184.216.34", "8.8.8.8"]
|
||||
kept = sc.filter(given)
|
||||
assert kept == ["api.example.com", "93.184.216.34"]
|
||||
|
||||
|
||||
def test_substring_domain_not_matched():
|
||||
# "fakeexample.com" must not match "example.com"
|
||||
sc = _scope()
|
||||
assert not sc.is_in_scope("fakeexample.com")
|
||||
|
||||
|
||||
def test_invalid_cidr_ignored():
|
||||
sc = Scope.from_dict({"in_scope_cidrs": ["not-a-cidr", "10.0.0.0/8"]})
|
||||
assert sc.cidrs == ["10.0.0.0/8"]
|
||||
assert sc.is_in_scope("10.1.2.3")
|
||||
Reference in New Issue
Block a user