From c64eb8e2144f6d86a1afe2f57f4b3d67dce81bac Mon Sep 17 00:00:00 2001 From: Evgeny Kiriyak <224408464+evkir@users.noreply.github.com> Date: Wed, 26 Aug 2026 20:10:13 +0200 Subject: [PATCH 1/4] feat(scope): refuse an unscoped exploit phase under --strict-scope T3 in STANDOFF-KEY asked for an empty scope to be a violation rather than a warning when --strict-scope is set. The flag did not exist: a grep across cyberai/ for strict_scope returned nothing, so the item as written could not be done. It is introduced here. The validator warned on an absent scope and let the run proceed. That is defensible as a default -- the pipeline has always behaved that way and a flag that changes behaviour for people who did not set it is a surprise, not a safety feature -- 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. Under strict the phase now refuses: the orchestrator already raises on a failed verdict, so a violation is a failed exploit phase rather than a line of yellow text nobody reads. The switch travels as a parameter, not as an environment read inside the validator. A function that consults os.environ cannot be driven from a test without patching the process, and this one is otherwise pure. All eighteen existing call sites pass no strict argument and every one still gets a warning; the default is unchanged by construction and by test. One gap was found by mutation rather than by review. Deleting `strict=self.config.strict_scope` from _run_exploit left six new tests green: they drive the validator directly and the CLI directly, and neither can see that the phase in between stopped passing the value on. A flag that reaches the config and dies there is a producer without a consumer wearing full test coverage. The seventh test drives the phase itself, and both mutations -- dropping the argument, and neutering the raise -- now fail it. --- .env.example | 1 + README.md | 1 + cyberai/__main__.py | 10 ++ cyberai/agents/exploit/safety_validator.py | 15 ++- cyberai/core/config.py | 5 + cyberai/core/orchestrator.py | 7 +- tests/unit/test_strict_scope.py | 106 +++++++++++++++++++++ 7 files changed, 143 insertions(+), 2 deletions(-) create mode 100644 tests/unit/test_strict_scope.py 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/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/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/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..45ada1e 100644 --- a/cyberai/core/orchestrator.py +++ b/cyberai/core/orchestrator.py @@ -328,7 +328,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}") 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 From 40a9ed98d2f8feb8867c008bc96f0ac130dd6bb0 Mon Sep 17 00:00:00 2001 From: Evgeny Kiriyak <224408464+evkir@users.noreply.github.com> Date: Wed, 26 Aug 2026 20:18:07 +0200 Subject: [PATCH 2/4] fix(orchestrator): a crossed budget or a broken air gap ends the run T4 asked for the bare `except Exception` around a phase to be replaced with typed handling of network, LLM and tool errors. Measured against the code that is the wrong fix, and doing it would have made the pipeline worse. 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 trades a failed phase for a failed run. The comment calling it deliberate is correct and it stays. The real defect is the opposite one, and it was found by running the code rather than by reading it. Two exceptions were being caught here that are not phase failures at all: BudgetExceeded says a spending cap was crossed. Measured before the change: a run whose first phase hit the cap recorded a failed phase and moved to the next one, which was free to spend again. The cap stopped being a cap exactly when it began to matter. EgressViolation says the air-gapped path was asked to reach a remote provider. The run continued after the property it exists for had already been broken. Both now re-raise, in the sync path and in AsyncOrchestrator's copy of the same clause. The phase is still recorded as failed before the exception leaves, so the session says where the run ended rather than simply stopping. Everything else is still caught, and a test pins that: a dead target must not become a dead run. Recorded phase errors also carry their type now. "refused" alone reads the same whether a target dropped the connection or a defect raised it; the session export, the critic deciding on a retry, and whoever opens session.json get "ConnectionError: refused" instead. The format matches what mcp/client_probe.py already writes. Four mutations, four predicted victims: dropping either raise, emptying the fatal tuple, and reducing the description to str(exc). Dropping the sync raise leaves the async test green and vice versa, which is what proves the two copies are covered separately rather than by accident. --- cyberai/core/orchestrator.py | 45 ++++++++- tests/unit/test_run_level_failures.py | 126 ++++++++++++++++++++++++++ 2 files changed, 169 insertions(+), 2 deletions(-) create mode 100644 tests/unit/test_run_level_failures.py diff --git a/cyberai/core/orchestrator.py b/cyberai/core/orchestrator.py index 45ada1e..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) @@ -446,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/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__}: ") From 2e24c9ff8fbf6a5bb1adcbc6ae2b10ea6eaf85d9 Mon Sep 17 00:00:00 2001 From: Evgeny Kiriyak <224408464+evkir@users.noreply.github.com> Date: Wed, 26 Aug 2026 20:26:20 +0200 Subject: [PATCH 3/4] fix(intel): TLS problems reach the report instead of the knowledge base T2 asked for the intel agent to stop losing ports from the KB during CVE lookup. Measured, that does not happen: after the mass-open early return the port list is still in recon.nmap, untouched, and the skip is a deliberate trade the docstring explains -- a proxy answering every probe makes the service data noise, and spraying NVD with it returns garbage. The defect next to it is real. TLSCVEMapper had no product caller until a previous fix gave it one, and that fix wrote the enriched list to `intel.tls_findings`. Nothing reads that key: a grep across cyberai/ returns the write and nothing else, and the report renders session.findings, not the KB. Recon publishes no TLS finding either. So on a real target 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 the client reads. The cure reproduced the disease one level up. Each enriched TLS problem now becomes a finding. Severity travels from the TLS classifier rather than being decided again here: it already knows a deprecated protocol outranks a weak cipher. An unrecognised level degrades to INFO rather than raising, because losing a finding over a spelling is worse than under-rating one. A condition with no matching CVE still reaches the report and says so -- unmapped is a configuration problem, not the absence of one. The call keeps its position above both early returns, and a test now covers the mass-open path as well as the no-ports one. Those are exactly the runs a TLS-only target lands in, and a fix that only worked on the full path would miss both. tests/unit/test_intel_tls_context.py stays green through every mutation of the new code, which is the point: it pins the producer, and the new file pins the consumer. Removing the finding loop reddens five of six new tests and none of the four old ones. --- cyberai/agents/intel/agent.py | 45 ++++++- .../test_tls_findings_reach_the_report.py | 115 ++++++++++++++++++ 2 files changed, 155 insertions(+), 5 deletions(-) create mode 100644 tests/unit/test_tls_findings_reach_the_report.py 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/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 == [] From 7633bcb57f6a9759a73f96850698cae54ad38176 Mon Sep 17 00:00:00 2001 From: Evgeny Kiriyak <224408464+evkir@users.noreply.github.com> Date: Wed, 26 Aug 2026 20:33:13 +0200 Subject: [PATCH 4/4] fix(exploit): the scope guard covers every path that touches the target T1 asked for authorized_scope to reach the exploit agent. It already does -- _scope_ok has been reading session.authorized_scope since before this sprint. What the item missed is that the guard reached one caller out of three. Its own docstring promised nuclei and the out-of-band path. Measured: _scope_ok returned False for a target the scope excluded while _oob_confirmer built a collector for that same target and handed it to the walk. The HTTP walk itself never asked at all -- it sent the full payload corpus at whatever host it was given. The orchestrator gates session.target, which is the same value on a normal run and is not the value these methods receive, so a direct call bypassed authorisation entirely. That bypass is what the in-agent guard was added to close. Both now consult it, and the OOB check sits above the collector probe: an unauthorised host must not even cost a request to phantom-grid. An empty scope is still a no-op, which is what the bench profile relies on. One test in this file was worthless and mutation found it. The first version of the web-path test asserted zeros against an agent with an empty knowledge base, so it read the "no HTTP surface" branch and called it a refusal; with the guard deleted it stayed green. It now puts a surface in the KB and patches exploit_surface, so a guard that fails to fire shows up as a call. Also: Development Status classifier moves from Alpha to Beta. It said Alpha while the package shipped a wired trust boundary, decontaminated proofs, an architecture-tested README and 2280 tests. Beta claims the interfaces are stable enough to build against and that the failure modes are known and written down -- not that the work is done. --- CHANGELOG.md | 6 ++ cyberai/agents/exploit/agent.py | 19 ++++- pyproject.toml | 2 +- tests/unit/test_exploit_scope_guard.py | 99 ++++++++++++++++++++++++++ 4 files changed, 123 insertions(+), 3 deletions(-) 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/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/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()