diff --git a/.env.example b/.env.example index 43f74d9..8373e20 100644 --- a/.env.example +++ b/.env.example @@ -18,6 +18,7 @@ CYBERAI_USE_EXPLOIT_MEMORY=0 CYBERAI_USE_LAB_DOGFOOD=0 CYBERAI_WEB_ENABLE_BENCH_TRIGGER=0 CYBERAI_AIR_GAPPED=0 +CYBERAI_STRICT_SCOPE=0 CYBERAI_ENABLE_MODEL_ROUTING=0 # Numeric / path settings (optional) diff --git a/CHANGELOG.md b/CHANGELOG.md index a85a0a4..7fe7f28 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,12 @@ All notable changes to CyberAI are documented here. ### Changed +- **Development status: Alpha to Beta.** The classifier said Alpha while the + package carried a wired trust boundary, decontaminated benchmark proofs, an + architecture-tested README and 2200-odd tests. Beta claims the interfaces + are stable enough to build against and that the failure modes are known and + written down -- not that the tool is finished. + - **Licence: MIT to Apache-2.0.** Releases up to and including v1.5.0 were published under the MIT License and remain available under those terms permanently, on both GitHub and PyPI; the change is not retroactive and diff --git a/README.md b/README.md index 5d854a5..25c813f 100644 --- a/README.md +++ b/README.md @@ -278,6 +278,7 @@ Every setting can be driven from the environment (or a `.env` file - see | `CYBERAI_ENABLE_REPLAN` | Critic-driven phase replan | | `CYBERAI_USE_EXPLOIT_MEMORY` | Recall similar past exploit chains | | `CYBERAI_AIR_GAPPED` | Force local-only (no-egress) LLM path | +| `CYBERAI_STRICT_SCOPE` | Refuse the exploit phase when no scope was given | | `CYBERAI_ENABLE_MODEL_ROUTING` | Per-phase model selection | | `CYBERAI_MAX_COST_USD` | LLM spend budget (0 = disabled) | | `CYBERAI_OUTPUT_DIR` | Report output directory | diff --git a/cyberai/__main__.py b/cyberai/__main__.py index 0218686..2ee666a 100644 --- a/cyberai/__main__.py +++ b/cyberai/__main__.py @@ -58,6 +58,7 @@ def _apply_feature_overrides( replan: bool | None = None, planner: bool | None = None, air_gapped: bool | None = None, + strict_scope: bool | None = None, web_recon: bool | None = None, web_exploit: bool | None = None, oob: bool | None = None, @@ -100,6 +101,8 @@ def _apply_feature_overrides( config.enable_planner = planner if air_gapped is not None: config.air_gapped = air_gapped + if strict_scope is not None: + config.strict_scope = strict_scope if web_recon is not None: config.use_web_recon = web_recon if web_exploit is not None: @@ -162,6 +165,11 @@ def cli() -> None: @click.option("--model", default=None, help="LLM model (overrides provider default)") @click.option("--dry-run", is_flag=True, help="Run pipeline without real network calls") @click.option("--scope", multiple=True, help="Authorized scope entry (repeatable)") +@click.option( + "--strict-scope/--no-strict-scope", + default=None, + help="Refuse to exploit when no scope was given, instead of warning", +) @click.option( "--auth", multiple=True, @@ -266,6 +274,7 @@ def scan( replan: bool | None, planner: bool | None, air_gapped: bool | None, + strict_scope: bool | None, web_recon: bool | None, web_exploit: bool | None, oob: bool | None, @@ -305,6 +314,7 @@ def scan( replan=replan, planner=planner, air_gapped=air_gapped, + strict_scope=strict_scope, web_recon=web_recon, web_exploit=web_exploit, oob=oob, diff --git a/cyberai/agents/exploit/agent.py b/cyberai/agents/exploit/agent.py index da8e6d6..4f97287 100644 --- a/cyberai/agents/exploit/agent.py +++ b/cyberai/agents/exploit/agent.py @@ -218,6 +218,10 @@ def _oob_confirmer(self, target: str): confirmation entirely rather than spending its wait on every parameter to learn the same thing each time. """ + if not self._scope_ok(target): + self._log(f"target {target} out of authorized scope -- no OOB confirmation") + return None + configured = self.config.phantom.grid_url or "" parsed = urlparse(configured) port = f":{parsed.port}" if parsed.port else "" @@ -254,6 +258,10 @@ def _run_web_exploit(self, target: str, classes: Optional[List[Any]] = None) -> """ from cyberai.core.scan_session import Severity + if not self._scope_ok(target): + self._log(f"target {target} out of authorized scope -- skipping web exploitation") + return {"confirmed": 0, "endpoints_tested": 0, "findings": []} + surface = self.kb.get("recon.web_surface", {}) or {} endpoints = surface.get("endpoints", []) if not endpoints: @@ -494,8 +502,15 @@ def _scope_ok(self, target: str) -> bool: """True if `target` is authorized for live actions. Empty scope preserves prior behaviour (no gate -> allowed); a non-empty - scope the target does not match blocks live actions (nuclei / OOB), - closing the bypass around the orchestrator-level scope gate. + scope the target does not match blocks live actions, closing the bypass + around the orchestrator-level scope gate. + + Every method that puts a packet on the wire consults this. The guard + was written to cover nuclei and the out-of-band path and reached only + nuclei: `_oob_confirmer` built a collector for a target the scope + excluded, and the HTTP walk sent its whole payload corpus without ever + asking. The orchestrator gates `session.target`, which is the same + value on a normal run and not the value these methods are handed. """ scope = getattr(getattr(self, "session", None), "authorized_scope", None) or [] if not scope: diff --git a/cyberai/agents/exploit/safety_validator.py b/cyberai/agents/exploit/safety_validator.py index 5089737..6a3d0c3 100644 --- a/cyberai/agents/exploit/safety_validator.py +++ b/cyberai/agents/exploit/safety_validator.py @@ -49,6 +49,7 @@ def validate_exploit_scope( target: str, authorized_scope: List[str] = None, attack_paths: List[Dict[str, Any]] = None, + strict: bool = False, ) -> ValidationResult: """ Validates that target and attack paths are within authorized scope. @@ -79,7 +80,19 @@ def validate_exploit_scope( violations.extend(ip_violations) if not authorized_scope: - warnings.append("No authorized_scope provided — proceeding without scope check") + # Absence of scope is not authorisation. Under strict it is a refusal: + # the caller asked for a run that cannot touch anything it was not + # pointed at, and an empty list points at nothing. The default stays + # a warning, because the pipeline has always proceeded here and a + # flag that changes behaviour for people who did not ask for it is + # not a safety feature, it is a surprise. + if strict: + violations.append( + "No authorized_scope provided and --strict-scope is set " + "— refusing to exploit an unscoped target" + ) + else: + warnings.append("No authorized_scope provided — proceeding without scope check") elif not in_scope: violations.append(f"Target '{target}' is NOT in authorized scope: {authorized_scope}") diff --git a/cyberai/agents/intel/agent.py b/cyberai/agents/intel/agent.py index d107c18..d91dd46 100644 --- a/cyberai/agents/intel/agent.py +++ b/cyberai/agents/intel/agent.py @@ -64,12 +64,26 @@ def _register_tools(self) -> None: ) ) - def _enrich_tls_findings(self) -> None: - """Attach CVE context to the TLS findings recon left in the KB. + def _enrich_tls_findings(self, target: str) -> None: + """Turn the TLS problems recon found into findings, with CVE context. TLSCVEMapper existed with no product caller: recon wrote recon.tls, - the mapper knew how to read it, and nothing joined them. The findings - list from TLSTool.run is already the shape the mapper expects. + the mapper knew how to read it, and nothing joined them. Joining them + into the knowledge base reproduced the same defect one level up -- + `intel.tls_findings` had no reader either, and the report renders + session.findings, not the KB. So an expired certificate, a negotiated + TLS 1.0 and RC4 in the suite were all probed, classified, matched to + CVEs, and then left out of the document the client reads. + + The finding is raised here rather than in recon because the CVE + context is what makes it actionable, and recon has no mapper. Recon + publishes no TLS finding at all, so there is nothing to duplicate. + + Severity travels from the TLS classifier rather than being decided + again here: it already knows a deprecated protocol is worse than a + weak cipher. An unrecognised level degrades to INFO instead of + raising, because a report that loses a finding over a spelling is + worse than one that under-rates it. """ tls_data = self.kb.get("recon.tls", {}) or {} findings = tls_data.get("findings", []) if isinstance(tls_data, dict) else [] @@ -77,6 +91,26 @@ def _enrich_tls_findings(self) -> None: return enriched = TLSCVEMapper().enrich(findings) self.kb.set("intel.tls_findings", enriched, agent=self.AGENT_NAME) + + for item in enriched: + level = str(item.get("severity") or "").upper() + severity = getattr(Severity, level, Severity.INFO) + cves = [c for c in (item.get("cves") or []) if c] + evidence = [item.get("detail", "")] + if item.get("remediation"): + evidence.append(f"Remediation: {item['remediation']}") + if not cves: + evidence.append("No CVE maps to this condition; it is a configuration issue") + self.session.add_finding( + severity=severity, + title=item.get("issue") or "TLS configuration issue", + description=item.get("detail", ""), + agent=self.AGENT_NAME, + target=target, + cve_ids=cves, + evidence=evidence, + ) + with_cves = sum(1 for f in enriched if f.get("cves")) self._log(f"TLS: {with_cves}/{len(enriched)} findings mapped to CVEs") @@ -85,7 +119,8 @@ def run(self, target: str, context: Optional[Dict[str, Any]] = None) -> Dict[str # This runs before the port checks below because those return early: # an HTTPS target with no open ports in the nmap result would # otherwise drop TLS data that recon already put in the KB. - self._enrich_tls_findings() + + self._enrich_tls_findings(target) nmap_data = self.kb.get("recon.nmap", {}) or {} ports = nmap_data.get("ports", []) if isinstance(nmap_data, dict) else [] diff --git a/cyberai/core/config.py b/cyberai/core/config.py index 688a7c4..5ff9404 100644 --- a/cyberai/core/config.py +++ b/cyberai/core/config.py @@ -181,6 +181,10 @@ class CyberAIConfig: routing: "RoutingConfig" = field(default_factory=lambda: RoutingConfig()) # Flag-gated: force all LLM calls onto a local endpoint, assert no egress. air_gapped: bool = False + # Flag-gated: an empty authorized scope becomes a violation instead of a + # warning, so the exploit phase refuses to run against a target nobody + # named. Off by default -- see the validator for why. + strict_scope: bool = False # Cap nmap scan rate (packets/sec) on external/legal targets; None = uncapped. max_rps: Optional[int] = None # Headers every web request carries. A target that refuses an anonymous @@ -257,4 +261,5 @@ def from_env(cls) -> "CyberAIConfig": use_lab_dogfood=_env_bool("CYBERAI_USE_LAB_DOGFOOD", False), web_enable_bench_trigger=_env_bool("CYBERAI_WEB_ENABLE_BENCH_TRIGGER", False), air_gapped=_env_bool("CYBERAI_AIR_GAPPED", False), + strict_scope=_env_bool("CYBERAI_STRICT_SCOPE", False), ) diff --git a/cyberai/core/orchestrator.py b/cyberai/core/orchestrator.py index 6e1fe25..0a21660 100644 --- a/cyberai/core/orchestrator.py +++ b/cyberai/core/orchestrator.py @@ -19,6 +19,8 @@ from rich.panel import Panel from cyberai.core.config import CyberAIConfig +from cyberai.core.cost_tracker import BudgetExceeded +from cyberai.core.egress_guard import EgressViolation from cyberai.core.llm_usage import llm_usage_record, llm_zero_reason from cyberai.core.logger import AuditLogger, get_logger from cyberai.core.scan_session import ScanPhase, ScanSession @@ -26,6 +28,33 @@ console = Console() log = get_logger("orchestrator") +# A phase is allowed to fail. Two things are not phase failures at all: they +# are the end of a guarantee the run was given, and continuing past them +# turns that guarantee into a suggestion. +# +# BudgetExceeded means a spending cap was crossed. Swallowing it per phase +# leaves the next phase free to spend again, so the cap stops being a cap +# exactly when it starts to matter. EgressViolation means the air-gapped +# path was asked to talk to a remote provider; a run that continues after +# that has already broken the property it was started for. +# +# Everything else stays caught. The broad except below is deliberate and +# stays: a phase dispatches nmap, httpx, subprocesses, three LLM providers +# and several external binaries, and no honest list of what they raise can +# be written. Narrowing it would trade a failed phase for a failed run. +FATAL_TO_THE_RUN = (BudgetExceeded, EgressViolation) + + +def _describe(exc: BaseException) -> str: + """Name the failure as well as describe it. + + A recorded phase error of "refused" reads the same whether a target + dropped the connection or a defect raised it. The session export, the + critic that decides on a retry, and whoever opens session.json all get + the type as well, at the cost of one prefix. + """ + return f"{type(exc).__name__}: {exc}" + class Orchestrator: """ @@ -223,8 +252,14 @@ def _run_phase(self, session: ScanSession, phase: ScanPhase) -> None: session.record_phase(phase, success=True, started=started, data=data) console.print(f"[green]✓ {phase.value} done[/green]") + except FATAL_TO_THE_RUN as exc: + session.record_phase(phase, success=False, started=started, error=_describe(exc)) + console.print(f"[bold red]✗ {phase.value} aborted the run: {exc}[/bold red]") + log.error(f"Phase {phase.value} hit a run-level limit", exc_info=True) + raise + except Exception as exc: # noqa: BLE001 — pipeline must survive one bad phase - session.record_phase(phase, success=False, started=started, error=str(exc)) + session.record_phase(phase, success=False, started=started, error=_describe(exc)) console.print(f"[red]✗ {phase.value} error: {exc}[/red]") log.error(f"Phase {phase.value} raised", exc_info=True) @@ -328,7 +363,12 @@ def _run_exploit(self, session: ScanSession) -> Dict: from cyberai.agents.exploit.safety_validator import validate_exploit_scope paths = session.kb_get("intel", {}).get("ranked_cves", []) - v = validate_exploit_scope(session.target, session.authorized_scope, paths) + v = validate_exploit_scope( + session.target, + session.authorized_scope, + paths, + strict=self.config.strict_scope, + ) if not v.passed: raise RuntimeError(f"Scope check failed: {v.violations}") @@ -441,8 +481,14 @@ async def _run_phase_async(self, session: ScanSession, phase: ScanPhase) -> None self._check_phase_injection(session, phase, data) session.record_phase(phase, success=True, started=started, data=data) console.print(f"[green]✓ {phase.value} done[/green]") + except FATAL_TO_THE_RUN as exc: + session.record_phase(phase, success=False, started=started, error=_describe(exc)) + console.print(f"[bold red]✗ {phase.value} aborted the run: {exc}[/bold red]") + log.error(f"Phase {phase.value} hit a run-level limit", exc_info=True) + raise + except Exception as exc: # noqa: BLE001 - session.record_phase(phase, success=False, started=started, error=str(exc)) + session.record_phase(phase, success=False, started=started, error=_describe(exc)) console.print(f"[red]✗ {phase.value} error: {exc}[/red]") log.error(f"Phase {phase.value} raised", exc_info=True) diff --git a/pyproject.toml b/pyproject.toml index bf59be1..66abb99 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,7 @@ authors = [ ] keywords = ["pentest", "security", "ai", "multi-agent", "offensive-security"] classifiers = [ - "Development Status :: 3 - Alpha", + "Development Status :: 4 - Beta", "Intended Audience :: Information Technology", "License :: OSI Approved :: Apache Software License", "Programming Language :: Python :: 3", diff --git a/tests/unit/test_exploit_scope_guard.py b/tests/unit/test_exploit_scope_guard.py index aabc3be..d22d321 100644 --- a/tests/unit/test_exploit_scope_guard.py +++ b/tests/unit/test_exploit_scope_guard.py @@ -6,6 +6,8 @@ before any network tool is touched. """ +from unittest.mock import MagicMock, patch + from cyberai.agents.exploit.agent import ExploitAgent from cyberai.core.config import CyberAIConfig from cyberai.core.scan_session import ScanSession @@ -32,3 +34,100 @@ def test_run_nuclei_skips_out_of_scope(): # Guard returns before NucleiEngine() -- no binary dependency. result = _agent(["*.acme.com"])._run_nuclei("evil.example.com", [{"cve_id": "CVE-2021-1"}]) assert result == [] + + +# The guard's docstring promised nuclei and OOB and delivered nuclei. Measured: +# _scope_ok returned False for evil.example.com while _oob_confirmer built a +# collector for it anyway, and the HTTP walk sent its whole payload corpus +# without asking at all. The orchestrator gates session.target, which is not +# the value these two are handed. + + +def _oob_agent(scope): + config = CyberAIConfig() + config.use_oob = True + session = ScanSession(target="evil.example.com", authorized_scope=scope) + return ExploitAgent(config, session) + + +def test_oob_confirmer_is_not_built_for_an_out_of_scope_target(): + """No collector, no callback, no request to a host nobody authorised.""" + assert _oob_agent(["*.acme.com"])._oob_confirmer("evil.example.com") is None + + +_SURFACE = { + "endpoints": [ + { + "method": "GET", + "url": "http://evil.example.com/search", + "params": ["q"], + } + ] +} + + +def test_web_exploitation_skips_an_out_of_scope_target(): + """The walk is the loudest thing the agent does; it asked permission last. + + The surface has to be in the knowledge base for this to prove anything. + An agent with an empty KB returns the same zeros whether the guard fires + or not -- the first version of this test passed with the guard deleted, + because it was reading the "no HTTP surface" branch and calling it a + refusal. exploit_surface is patched so a guard that failed to fire is + visible as a call rather than as a network request. + """ + agent = _agent(["*.acme.com"]) + agent.kb.set("recon.web_surface", _SURFACE) + + with patch("cyberai.agents.exploit.agent.exploit_surface") as walk: + result = agent._run_web_exploit("evil.example.com") + + walk.assert_not_called() + assert result["endpoints_tested"] == 0 + assert result["findings"] == [] + + +def test_web_exploitation_runs_for_a_target_inside_the_scope(): + """The other side of the same guard: in scope, the walk happens.""" + session = ScanSession(target="api.acme.com", authorized_scope=["*.acme.com"]) + agent = ExploitAgent(CyberAIConfig(), session) + agent.kb.set("recon.web_surface", _SURFACE) + + report = MagicMock() + report.findings = [] + report.oob_confirmed_params = [] + report.params_oob_confirmed = 0 + report.to_dict.return_value = {"confirmed": 0, "endpoints_tested": 1, "findings": []} + + with patch("cyberai.agents.exploit.agent.exploit_surface", return_value=report) as walk: + agent._run_web_exploit("api.acme.com") + + walk.assert_called_once() + + +def test_an_empty_scope_still_builds_the_confirmer(): + """No-regression: a bench run passes no scope and must be unaffected. + + The collector is stubbed as reachable. Without that this test passes on a + machine with phantom-grid running and fails in CI, and a green that + depends on what else is listening on the host is not evidence about the + scope guard. + """ + live_grid = MagicMock() + live_grid.available = True + with patch("cyberai.agents.exploit.agent.PhantomGridClient", return_value=live_grid): + assert _oob_agent([])._oob_confirmer("anything.example.org") is not None + + +def test_the_scope_guard_runs_before_the_collector_is_contacted(): + """Order matters: an out-of-scope target must not even probe the grid. + + If the guard sat below the availability check, an unauthorised host would + still cost a request to the collector, and the run would depend on + something unrelated to authorisation. + """ + grid = MagicMock() + grid.available = True + with patch("cyberai.agents.exploit.agent.PhantomGridClient", return_value=grid) as constructed: + assert _oob_agent(["*.acme.com"])._oob_confirmer("evil.example.com") is None + constructed.assert_not_called() diff --git a/tests/unit/test_run_level_failures.py b/tests/unit/test_run_level_failures.py new file mode 100644 index 0000000..2f3e3b4 --- /dev/null +++ b/tests/unit/test_run_level_failures.py @@ -0,0 +1,126 @@ +"""Some failures end a phase; two end the run. + +T4 in STANDOFF-KEY asked for the bare `except Exception` in the orchestrator +to be replaced with typed handling of network, LLM and tool errors. Measured +against the code, that item is the wrong fix. The clause wraps `_dispatch`, +which reaches nmap, httpx, subprocesses, three LLM providers and several +external binaries; no honest list of what they raise can be written, and a +narrower clause would trade a failed phase for a failed run. The docstring +that calls it deliberate is right. + +The real defect was the opposite one, and it was found by running it rather +than by reading. `BudgetExceeded` and `EgressViolation` were caught here too. +A run whose spending cap was crossed recorded a failed phase and moved to the +next one, free to spend again -- the cap stopped being a cap exactly when it +began to matter. A run whose air-gapped path was asked to reach a remote +provider carried on after the property it exists for had already been broken. + +Both are now re-raised, with the phase still recorded as failed so the +session says where the run ended. Everything else is still caught, which is +what the last test here pins: this change must not turn a dead target into a +dead run. +""" + +import asyncio +from unittest.mock import AsyncMock, patch + +import pytest + +from cyberai.core.config import CyberAIConfig +from cyberai.core.cost_tracker import BudgetExceeded +from cyberai.core.egress_guard import EgressViolation +from cyberai.core.orchestrator import AsyncOrchestrator, Orchestrator +from cyberai.core.scan_session import ScanPhase, ScanSession + + +def _run_two_phases(exc: BaseException) -> tuple[ScanSession, BaseException | None]: + """Drive two phases through _run_phase with the same failure on each.""" + orch = Orchestrator(CyberAIConfig()) + session = ScanSession(target="t.local") + escaped: BaseException | None = None + try: + with patch.object(Orchestrator, "_dispatch", side_effect=exc): + orch._run_phase(session, ScanPhase.RECON) + orch._run_phase(session, ScanPhase.INTEL) + except BaseException as caught: # noqa: BLE001 -- the point of the test + escaped = caught + return session, escaped + + +def test_a_crossed_budget_stops_the_run() -> None: + session, escaped = _run_two_phases(BudgetExceeded(1.5, 1.0)) + assert isinstance(escaped, BudgetExceeded) + # One phase, not two: the second never got the chance to spend again. + assert len(session.phases) == 1 + assert session.phases[0].success is False + + +def test_an_egress_violation_stops_the_run() -> None: + session, escaped = _run_two_phases(EgressViolation("provider is not local")) + assert isinstance(escaped, EgressViolation) + assert len(session.phases) == 1 + assert session.phases[0].success is False + + +def test_the_phase_is_recorded_before_the_exception_leaves() -> None: + """A run that ends must still say where it ended.""" + session, _ = _run_two_phases(BudgetExceeded(2.0, 1.0)) + assert session.phases[0].phase is ScanPhase.RECON + assert "budget exceeded" in (session.phases[0].error or "") + + +def test_an_ordinary_failure_still_only_ends_its_phase() -> None: + """The guarantee this change must not break.""" + session, escaped = _run_two_phases(ConnectionError("refused")) + assert escaped is None + assert len(session.phases) == 2 + assert not any(p.success for p in session.phases) + + +def _run_two_phases_async(exc: BaseException) -> tuple[ScanSession, BaseException | None]: + """The same drive through AsyncOrchestrator. + + _run_phase_async is a separate method on a separate class carrying its own + copy of the clause. Testing only the synchronous one would leave the copy + free to drift back, and a run driven through the async entry point is the + same run under the same spending cap. + """ + orch = AsyncOrchestrator(config=CyberAIConfig(), dry_run=False) + session = ScanSession(target="t.local") + escaped: BaseException | None = None + + async def drive() -> None: + await orch._run_phase_async(session, ScanPhase.RECON) + await orch._run_phase_async(session, ScanPhase.INTEL) + + with patch.object(orch, "_dispatch_async", new_callable=AsyncMock, side_effect=exc): + try: + asyncio.run(drive()) + except BaseException as caught: # noqa: BLE001 -- the point of the test + escaped = caught + return session, escaped + + +def test_the_async_path_stops_on_a_crossed_budget_too() -> None: + session, escaped = _run_two_phases_async(BudgetExceeded(1.5, 1.0)) + assert isinstance(escaped, BudgetExceeded) + assert len(session.phases) == 1 + assert session.phases[0].success is False + + +def test_the_async_path_still_survives_an_ordinary_failure() -> None: + session, escaped = _run_two_phases_async(ConnectionError("refused")) + assert escaped is None + assert len(session.phases) == 2 + assert not any(p.success for p in session.phases) + + +@pytest.mark.parametrize( + "exc", + [ConnectionError("refused"), BudgetExceeded(1.5, 1.0)], + ids=["ordinary", "fatal"], +) +def test_the_recorded_error_names_its_type(exc: BaseException) -> None: + """ "refused" alone does not say whether a target or a defect produced it.""" + session, _ = _run_two_phases(exc) + assert (session.phases[0].error or "").startswith(f"{type(exc).__name__}: ") diff --git a/tests/unit/test_strict_scope.py b/tests/unit/test_strict_scope.py new file mode 100644 index 0000000..b06e861 --- /dev/null +++ b/tests/unit/test_strict_scope.py @@ -0,0 +1,106 @@ +"""An empty scope refuses the exploit phase when the caller asked it to. + +T3 in STANDOFF-KEY. The validator warned on an absent scope and let the run +proceed, which is defensible as a default -- the pipeline has always behaved +that way -- and indefensible as the only option. Absence of authorisation is +not authorisation, and an engagement where that distinction matters had no +way to say so. + +The default is deliberately unchanged. Eighteen existing call sites pass no +strict argument and every one of them still gets a warning; a flag that +alters behaviour for people who did not set it is a surprise, not a safety +feature. + +The chain is tested end to end rather than at the validator alone, because +each link has failed independently before: a flag declared in click and never +read, a config field with no consumer, an orchestrator that computes a +verdict and ignores it. Here the CLI option reaches the config, the config +reaches the validator, and the orchestrator turns the violation into a +failed phase. +""" + +import pytest + +from cyberai.agents.exploit.safety_validator import validate_exploit_scope +from cyberai.core.config import CyberAIConfig + + +def test_empty_scope_warns_by_default() -> None: + v = validate_exploit_scope("scanme.nmap.org", [], []) + assert v.passed + assert any("proceeding without scope check" in w for w in v.warnings) + assert not v.violations + + +def test_empty_scope_is_a_violation_under_strict() -> None: + v = validate_exploit_scope("scanme.nmap.org", [], [], strict=True) + assert not v.passed + assert any("strict-scope" in x for x in v.violations) + assert not any("proceeding without scope check" in w for w in v.warnings) + + +def test_strict_does_not_touch_a_run_that_has_a_scope() -> None: + lenient = validate_exploit_scope("scanme.nmap.org", ["scanme.nmap.org"], []) + strict = validate_exploit_scope("scanme.nmap.org", ["scanme.nmap.org"], [], strict=True) + assert lenient.passed and strict.passed + assert strict.violations == lenient.violations + assert strict.warnings == lenient.warnings + + +def test_strict_does_not_rescue_an_out_of_scope_target() -> None: + """A named scope the target misses already fails; strict changes nothing.""" + v = validate_exploit_scope("evil.example.com", ["scanme.nmap.org"], [], strict=True) + assert not v.passed + assert any("NOT in authorized scope" in x for x in v.violations) + + +def test_config_field_defaults_off_and_reads_the_environment( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("CYBERAI_STRICT_SCOPE", raising=False) + assert CyberAIConfig.from_env().strict_scope is False + monkeypatch.setenv("CYBERAI_STRICT_SCOPE", "1") + assert CyberAIConfig.from_env().strict_scope is True + + +def test_the_orchestrator_hands_the_flag_to_the_validator() -> None: + """The link the other tests cannot see. + + Removing `strict=self.config.strict_scope` from _run_exploit left all six + of the tests above green: they drive the validator directly and the CLI + directly, and neither notices that the phase in between stopped passing + the value on. A flag that reaches the config and dies there is the shape + this project calls a producer without a consumer, and it survives review + precisely because every piece has a test. + + So this drives the phase. Under strict with no scope the exploit phase + must refuse; the same session without strict must get past the check. + Reaching the agent is not the point and would need a network, so the + second case asserts on what the validator decided rather than on a run. + """ + from cyberai.core.orchestrator import Orchestrator + from cyberai.core.scan_session import ScanSession + + config = CyberAIConfig() + config.strict_scope = True + orch = Orchestrator(config=config, dry_run=True) + session = ScanSession(target="scanme.nmap.org", authorized_scope=[]) + + with pytest.raises(RuntimeError, match="Scope check failed"): + orch._run_exploit(session) + + config.strict_scope = False + lenient = validate_exploit_scope( + session.target, session.authorized_scope, [], strict=config.strict_scope + ) + assert lenient.passed + + +def test_cli_flag_overrides_the_environment(monkeypatch: pytest.MonkeyPatch) -> None: + from cyberai.__main__ import _apply_feature_overrides + + monkeypatch.setenv("CYBERAI_STRICT_SCOPE", "1") + config = CyberAIConfig.from_env() + assert config.strict_scope is True + assert _apply_feature_overrides(config, strict_scope=False).strict_scope is False + assert _apply_feature_overrides(config, strict_scope=None).strict_scope is False diff --git a/tests/unit/test_tls_findings_reach_the_report.py b/tests/unit/test_tls_findings_reach_the_report.py new file mode 100644 index 0000000..a831c28 --- /dev/null +++ b/tests/unit/test_tls_findings_reach_the_report.py @@ -0,0 +1,115 @@ +"""TLS problems must reach the document, not just the knowledge base. + +The mapper had no caller; a previous fix gave it one and wrote the enriched +list to `intel.tls_findings`. That key had no reader either -- a grep for it +across cyberai/ returns the write and nothing else -- and the report renders +session.findings, not the KB. So an expired certificate, a negotiated +TLS 1.0 and RC4 in the cipher suite were probed, classified, matched to CVEs, +and then absent from the document a client reads. The same disease as the +one being cured, one level up. + +tests/unit/test_intel_tls_context.py already pins the KB write and would +stay green through all of that, which is the point: it tests the producer. +This file tests the consumer. + +The findings are raised through IntelAgent.run() rather than by calling the +private method, because the two early returns above it -- no ports, and a +mass-open port list -- are exactly the runs where a TLS-only target ends up, +and a fix that worked on the full path only would miss them both. +""" + +from __future__ import annotations + +from cyberai.agents.intel.agent import IntelAgent +from cyberai.core.config import CyberAIConfig +from cyberai.core.scan_session import ScanSession, Severity + +_DEPRECATED_TLS = { + "severity": "HIGH", + "issue": "Deprecated TLS version", + "detail": "Server negotiated TLSv1.0 — deprecated since RFC 8996", +} + +_WEAK_CIPHER = { + "severity": "MEDIUM", + "issue": "Weak cipher suite", + "detail": "RC4 in negotiated suite ECDHE-RSA-RC4-SHA", +} + +_NO_CVE = { + "severity": "LOW", + "issue": "Certificate chain includes an extra intermediate", + "detail": "No known CVE for this condition", +} + + +def _session(nmap: dict, *tls: dict) -> ScanSession: + session = ScanSession(target="example.com") + session.kb.set("recon.nmap", nmap) + if tls: + session.kb.set("recon.tls", {"domain": "example.com", "findings": list(tls)}) + return session + + +def test_a_tls_problem_becomes_a_finding_with_its_cves() -> None: + session = _session({"ports": []}, _DEPRECATED_TLS) + IntelAgent(CyberAIConfig(), session).run("example.com") + + tls = [f for f in session.findings if f.title == _DEPRECATED_TLS["issue"]] + assert len(tls) == 1 + assert tls[0].severity is Severity.HIGH + assert "CVE-2011-3389" in tls[0].cve_ids + + +def test_severity_comes_from_the_tls_classifier() -> None: + """Not re-decided here: it already knows deprecated beats weak cipher.""" + session = _session({"ports": []}, _DEPRECATED_TLS, _WEAK_CIPHER) + IntelAgent(CyberAIConfig(), session).run("example.com") + + by_title = {f.title: f.severity for f in session.findings} + assert by_title[_DEPRECATED_TLS["issue"]] is Severity.HIGH + assert by_title[_WEAK_CIPHER["issue"]] is Severity.MEDIUM + + +def test_a_mass_open_scan_still_reports_its_tls_problems() -> None: + """The run where this matters most. + + A fake-ip proxy answers every port, so the CVE lookup is skipped and the + port list is noise -- but the TLS handshake was real and what it revealed + is still true. + """ + session = _session( + {"ports": [{"port": 443, "service": "https"}], "mass_open": True, "open_count": 900}, + _DEPRECATED_TLS, + ) + result = IntelAgent(CyberAIConfig(), session).run("example.com") + + assert result["reason"] == "mass_open" + assert any(f.title == _DEPRECATED_TLS["issue"] for f in session.findings) + + +def test_a_condition_with_no_cve_still_reaches_the_report() -> None: + """An unmapped condition is a configuration problem, not a non-problem.""" + session = _session({"ports": []}, _NO_CVE) + IntelAgent(CyberAIConfig(), session).run("example.com") + + tls = [f for f in session.findings if f.title == _NO_CVE["issue"]] + assert len(tls) == 1 + assert tls[0].cve_ids == [] + assert any("No CVE maps" in e for e in tls[0].evidence) + + +def test_the_remediation_travels_with_the_finding() -> None: + session = _session({"ports": []}, _DEPRECATED_TLS) + IntelAgent(CyberAIConfig(), session).run("example.com") + + tls = next(f for f in session.findings if f.title == _DEPRECATED_TLS["issue"]) + assert any("Remediation:" in e for e in tls.evidence) + + +def test_a_target_with_no_tls_data_raises_nothing() -> None: + """Negative arm: an implementation that always adds a finding fails here.""" + session = _session({"ports": []}) + IntelAgent(CyberAIConfig(), session).run("example.com") + + assert session.findings == []