From a046bdf06663fae4faa2852d746de362c3e6d1d6 Mon Sep 17 00:00:00 2001 From: Evgeny Kiriyak <224408464+evkir@users.noreply.github.com> Date: Wed, 29 Jul 2026 08:36:48 +0200 Subject: [PATCH 1/4] feat(core): add HTTP endpoint nodes to the KB graph --- cyberai/core/kb_graph.py | 39 +++++++++++++- tests/unit/test_kb_graph_web.py | 96 +++++++++++++++++++++++++++++++++ 2 files changed, 134 insertions(+), 1 deletion(-) create mode 100644 tests/unit/test_kb_graph_web.py diff --git a/cyberai/core/kb_graph.py b/cyberai/core/kb_graph.py index 4542129..3fcce66 100644 --- a/cyberai/core/kb_graph.py +++ b/cyberai/core/kb_graph.py @@ -1,7 +1,8 @@ """In-memory knowledge-base graph. Builds a directed relationship graph from KnowledgeBase entries so a planner -can reason over multi-step attack paths (host -> port -> service -> CVE). +can reason over multi-step attack paths (host -> port -> service -> CVE, +and host -> HTTP endpoint for web surface). Backed by networkx and held entirely in memory — no external graph database — to keep the pipeline air-gapped. @@ -26,6 +27,7 @@ SERVICE = "service" CVE = "cve" LLM_ENDPOINT = "llm_endpoint" +HTTP_ENDPOINT = "http_endpoint" # Edge relations. HAS_PORT = "HAS_PORT" @@ -33,6 +35,7 @@ VULN_TO = "VULN_TO" SUBDOMAIN = "SUBDOMAIN" EXPOSES = "EXPOSES" +SERVES = "SERVES" def _add_node(g: nx.DiGraph, ntype: str, name: Any, **attrs: Any) -> Node: @@ -92,10 +95,44 @@ def build_kb_graph(kb: KnowledgeBase, target: Optional[str] = None) -> nx.DiGrap continue g.add_edge(root, _add_node(g, LLM_ENDPOINT, url), rel=EXPOSES) + _add_http_endpoints(g, kb, root) _add_cves(g, kb, root) return g +def _add_http_endpoints(g: nx.DiGraph, kb: KnowledgeBase, root: Node) -> None: + """Add HTTP endpoint nodes from the web surface recon discovered. + + Identity is method plus URL, because the surface keys endpoints by that + pair: one path can expose different parameters per method, and collapsing + them would lose a route. Parameterless routes are recorded as well -- they + describe reachable surface even though nothing can be injected into them + yet, and a consumer can tell them apart by an empty ``params``. + """ + surface = kb.get("recon.web_surface") or {} + if not isinstance(surface, dict): + return + entries = list(surface.get("endpoints") or []) + list(surface.get("routes") or []) + for entry in entries: + if not isinstance(entry, dict): + continue + url = entry.get("url") + if not url: + continue + method = str(entry.get("method") or "GET").upper() + node = _add_node( + g, + HTTP_ENDPOINT, + f"{method} {url}", + url=str(url), + method=method, + params=list(entry.get("params") or []), + body_params=list(entry.get("body_params") or []), + source=entry.get("source"), + ) + g.add_edge(root, node, rel=SERVES) + + def _add_cves(g: nx.DiGraph, kb: KnowledgeBase, root: Node) -> None: service_nodes = {d["name"]: n for n, d in g.nodes(data=True) if d.get("ntype") == SERVICE} diff --git a/tests/unit/test_kb_graph_web.py b/tests/unit/test_kb_graph_web.py new file mode 100644 index 0000000..6e47af9 --- /dev/null +++ b/tests/unit/test_kb_graph_web.py @@ -0,0 +1,96 @@ +"""HTTP endpoint nodes in the KB graph.""" + +from __future__ import annotations + +from cyberai.core.kb_graph import ( + HOST, + HTTP_ENDPOINT, + SERVES, + build_kb_graph, + nodes_by_type, +) +from cyberai.core.scan_session import ScanSession + +TARGET = "acme.tld" +BASE = "http://acme.tld" + + +def _session(surface=None) -> ScanSession: + s = ScanSession(target=TARGET) + s.kb.set("recon.result", {"target": TARGET}) + if surface is not None: + s.kb.set("recon.web_surface", surface) + return s + + +def _attrs(g, name): + return g.nodes[(HTTP_ENDPOINT, name)] + + +def test_endpoints_and_routes_become_nodes(): + g = build_kb_graph( + _session( + { + "endpoints": [ + { + "url": f"{BASE}/switch_personal_path", + "method": "GET", + "params": ["path"], + "body_params": ["path"], + "source": "openapi", + } + ], + "routes": [{"url": f"{BASE}/done", "method": "GET", "params": []}], + } + ).kb + ) + names = sorted(n[1] for n in nodes_by_type(g, HTTP_ENDPOINT)) + assert names == [f"GET {BASE}/done", f"GET {BASE}/switch_personal_path"] + + ep = _attrs(g, f"GET {BASE}/switch_personal_path") + assert ep["url"] == f"{BASE}/switch_personal_path" + assert ep["method"] == "GET" + assert ep["params"] == ["path"] + assert ep["body_params"] == ["path"] + assert ep["source"] == "openapi" + + # A parameterless route is still surface, distinguishable by empty params. + assert _attrs(g, f"GET {BASE}/done")["params"] == [] + + +def test_endpoints_hang_off_the_root_host(): + g = build_kb_graph(_session({"endpoints": [{"url": f"{BASE}/x", "params": ["q"]}]}).kb) + edge = g.edges[(HOST, TARGET), (HTTP_ENDPOINT, f"GET {BASE}/x")] + assert edge["rel"] == SERVES + + +def test_method_is_part_of_identity(): + g = build_kb_graph( + _session( + { + "endpoints": [ + {"url": f"{BASE}/item", "method": "GET", "params": ["id"]}, + {"url": f"{BASE}/item", "method": "POST", "params": ["name"]}, + ] + } + ).kb + ) + assert len(nodes_by_type(g, HTTP_ENDPOINT)) == 2 + assert _attrs(g, f"POST {BASE}/item")["params"] == ["name"] + + +def test_method_defaults_to_get_and_is_upper_cased(): + g = build_kb_graph( + _session({"endpoints": [{"url": f"{BASE}/a", "method": "post", "params": ["q"]}]}).kb + ) + assert _attrs(g, f"POST {BASE}/a")["method"] == "POST" + + g = build_kb_graph(_session({"endpoints": [{"url": f"{BASE}/b", "params": ["q"]}]}).kb) + assert _attrs(g, f"GET {BASE}/b")["method"] == "GET" + + +def test_missing_or_malformed_surface_adds_nothing(): + assert nodes_by_type(build_kb_graph(_session().kb), HTTP_ENDPOINT) == [] + assert nodes_by_type(build_kb_graph(_session("not-a-dict").kb), HTTP_ENDPOINT) == [] + surface = {"endpoints": [None, {"method": "GET"}, {"url": ""}], "routes": ["nope"]} + assert nodes_by_type(build_kb_graph(_session(surface).kb), HTTP_ENDPOINT) == [] From 116d6b7e329f1ff5117d9b04f2b590dc3436a680 Mon Sep 17 00:00:00 2001 From: Evgeny Kiriyak <224408464+evkir@users.noreply.github.com> Date: Wed, 29 Jul 2026 08:39:27 +0200 Subject: [PATCH 2/4] feat(planner): emit web-exploit subtasks from HTTP endpoint nodes --- cyberai/agents/planner/agent.py | 25 ++++++++- tests/unit/test_planner_web_subtasks.py | 70 +++++++++++++++++++++++++ 2 files changed, 93 insertions(+), 2 deletions(-) create mode 100644 tests/unit/test_planner_web_subtasks.py diff --git a/cyberai/agents/planner/agent.py b/cyberai/agents/planner/agent.py index a56cb0d..58fcad4 100644 --- a/cyberai/agents/planner/agent.py +++ b/cyberai/agents/planner/agent.py @@ -2,9 +2,11 @@ Turns the knowledge-base relationship graph into an ordered list of concrete subtasks (a TODO plan) that later phases can act on. Deterministic by default: -CVE nodes become exploitation subtasks ranked by score, LLM/RAG endpoints +CVE nodes become exploitation subtasks ranked by score, HTTP endpoints with +injectable parameters become web-exploitation subtasks, LLM/RAG endpoints become injection-fuzzing subtasks, and remaining services become enumeration -subtasks. The plan and a serialised graph are written to the KB under ``plan``. +subtasks. Which endpoints matter is decided here; the order payloads are tried +in stays with the exploitation layer, so there is one source of truth for each. The plan and a serialised graph are written to the KB under ``plan``. """ from __future__ import annotations @@ -15,6 +17,7 @@ from cyberai.core.kb_graph import ( CVE, HOST, + HTTP_ENDPOINT, LLM_ENDPOINT, SERVICE, attack_paths, @@ -68,6 +71,24 @@ def _plan_from_graph(self, graph: Any, target: str) -> List[Dict[str, Any]]: } ) + for node in nodes_by_type(graph, HTTP_ENDPOINT): + attrs = graph.nodes[node] + params = attrs.get("params") or [] + if not params: + # A route with no parameters offers nothing to inject; it stays + # in the graph as reachable surface but earns no subtask. + continue + subtasks.append( + { + "action": "web-exploit", + "target": attrs.get("url"), + "method": attrs.get("method"), + "params": list(params), + "body_params": list(attrs.get("body_params") or []), + "path": [target, attrs.get("url")], + } + ) + for node in nodes_by_type(graph, LLM_ENDPOINT): subtasks.append( { diff --git a/tests/unit/test_planner_web_subtasks.py b/tests/unit/test_planner_web_subtasks.py new file mode 100644 index 0000000..a43144f --- /dev/null +++ b/tests/unit/test_planner_web_subtasks.py @@ -0,0 +1,70 @@ +"""The planner turns discovered HTTP endpoints into web-exploit subtasks.""" + +from __future__ import annotations + +from typing import Any, Dict, List + +from cyberai.agents.planner.agent import PlannerAgent +from cyberai.core.config import CyberAIConfig +from cyberai.core.scan_session import ScanSession + +TARGET = "acme.tld" +BASE = "http://acme.tld" + +SURFACE = { + "endpoints": [ + { + "url": f"{BASE}/switch_personal_path", + "method": "GET", + "params": ["path"], + "body_params": ["path"], + "source": "openapi", + }, + {"url": f"{BASE}/search", "method": "POST", "params": ["q", "page"]}, + ], + "routes": [{"url": f"{BASE}/done", "method": "GET", "params": []}], +} + + +def _todo(surface: Any = None, ranked_cves: Any = None) -> List[Dict[str, Any]]: + s = ScanSession(target=TARGET) + s.kb.set("recon.result", {"target": TARGET}) + if surface is not None: + s.kb.set("recon.web_surface", surface) + if ranked_cves is not None: + s.kb.set("intel.ranked_cves", ranked_cves) + return PlannerAgent(CyberAIConfig(), s).run(TARGET)["todo"] + + +def _web(todo: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + return [t for t in todo if t["action"] == "web-exploit"] + + +def test_endpoints_become_web_exploit_subtasks(): + tasks = _web(_todo(SURFACE)) + assert [t["target"] for t in tasks] == [ + f"{BASE}/switch_personal_path", + f"{BASE}/search", + ] + first = tasks[0] + assert first["method"] == "GET" + assert first["params"] == ["path"] + assert first["body_params"] == ["path"] + assert first["path"] == [TARGET, f"{BASE}/switch_personal_path"] + assert tasks[1]["body_params"] == [] + + +def test_parameterless_routes_produce_no_subtask(): + assert all(t["target"] != f"{BASE}/done" for t in _web(_todo(SURFACE))) + + +def test_subtasks_are_numbered_with_the_rest_of_the_plan(): + todo = _todo(SURFACE, ranked_cves=[{"cve_id": "CVE-2024-9", "composite_score": 9.5}]) + assert [t["id"] for t in todo] == list(range(1, len(todo) + 1)) + assert todo[0]["action"] == "exploit" + assert [t["action"] for t in todo[1:3]] == ["web-exploit", "web-exploit"] + + +def test_no_web_surface_yields_no_web_subtasks(): + assert _web(_todo()) == [] + assert _web(_todo({"endpoints": []})) == [] From 87170c0dcb63c555e119f572ac90cf4d87cf3821 Mon Sep 17 00:00:00 2001 From: Evgeny Kiriyak <224408464+evkir@users.noreply.github.com> Date: Wed, 29 Jul 2026 08:42:14 +0200 Subject: [PATCH 3/4] feat(exploit): let a caller prioritise endpoints in exploit_surface --- cyberai/agents/exploit/web_exploit.py | 51 +++++++++++++++- tests/unit/test_web_exploit_priority.py | 79 +++++++++++++++++++++++++ 2 files changed, 127 insertions(+), 3 deletions(-) create mode 100644 tests/unit/test_web_exploit_priority.py diff --git a/cyberai/agents/exploit/web_exploit.py b/cyberai/agents/exploit/web_exploit.py index 05c607e..92841ec 100644 --- a/cyberai/agents/exploit/web_exploit.py +++ b/cyberai/agents/exploit/web_exploit.py @@ -18,6 +18,7 @@ from __future__ import annotations import logging +from collections.abc import Sequence from dataclasses import dataclass, field from typing import Any, Callable, Optional @@ -179,13 +180,52 @@ def _promise(endpoint: dict[str, Any]) -> tuple[int, int]: return (rank, len(names)) -def _by_promise(endpoints: list[dict[str, Any]]) -> list[dict[str, Any]]: +def _endpoint_key(endpoint: dict[str, Any]) -> tuple[str, str]: + """Identity of an endpoint: its URL and method, as recon keys them.""" + return (str(endpoint.get("url") or ""), str(endpoint.get("method") or "GET").upper()) + + +def _plan_rank(priority: Optional[Sequence[Any]]) -> dict[tuple[str, str], int]: + """Map each prioritised (url, method) pair to its position. + + Malformed entries are dropped rather than raised on: a caller's plan is + advisory, and a bad entry must not cost the whole run. + """ + rank: dict[tuple[str, str], int] = {} + for idx, item in enumerate(priority or []): + try: + url, method = item + except (TypeError, ValueError): + continue + if not url: + continue + key = (str(url), str(method or "GET").upper()) + rank.setdefault(key, idx) + return rank + + +def _by_promise( + endpoints: list[dict[str, Any]], priority: Optional[Sequence[Any]] = None +) -> list[dict[str, Any]]: """Most promising endpoints first, original order breaking ties. A spec can declare a hundred routes while the request budget covers a fraction of them, so this order decides what actually gets attacked. + + A caller that knows more than the parameter names -- a planner reading the + knowledge-base graph -- may name endpoints to try first. That selection + outranks the name heuristic, but only as a grouping: within the prioritised + set and within the rest, `_promise` still decides, so there remains exactly + one place that ranks an endpoint by what its parameters promise. """ - return sorted(endpoints, key=_promise) + rank = _plan_rank(priority) + if not rank: + return sorted(endpoints, key=_promise) + tail = len(rank) + return sorted( + endpoints, + key=lambda e: (rank.get(_endpoint_key(e), tail), *_promise(e)), + ) def _baseline_params(params: list[str], skip: str) -> dict[str, str]: @@ -200,12 +240,17 @@ def exploit_surface( timeout: float = DEFAULT_TIMEOUT, max_requests: int = 400, json_sender: Optional[SendFn] = None, + priority: Optional[Sequence[Any]] = None, ) -> WebExploitReport: """Attack every parameter on a discovered surface; return proven findings. One confirmation per parameter is enough: further payloads on an already proven parameter add requests and noise without adding evidence. + `priority` is an ordered sequence of (url, method) pairs to attack first; + endpoints it does not name follow. With a limited request budget, what is + reached at all depends on this order. + A parameter recon marked as `body_params` is delivered as a JSON body rather than a query string. Without that, a route declaring a required body answers every payload with a validation error, and the whole @@ -215,7 +260,7 @@ def exploit_surface( send_json = json_sender or _default_json_sender(timeout) report = WebExploitReport() - ordered = _by_promise(list(surface.get("endpoints", []))) + ordered = _by_promise(list(surface.get("endpoints", [])), priority) for endpoint in ordered: url = endpoint.get("url", "") method = (endpoint.get("method") or "GET").upper() diff --git a/tests/unit/test_web_exploit_priority.py b/tests/unit/test_web_exploit_priority.py new file mode 100644 index 0000000..9bb6fa1 --- /dev/null +++ b/tests/unit/test_web_exploit_priority.py @@ -0,0 +1,79 @@ +"""A caller may prioritise endpoints; `_promise` still ranks within groups.""" + +from __future__ import annotations + +from typing import Any, Dict, List + +from cyberai.agents.exploit.web_exploit import exploit_surface + +BASE = "http://acme.tld" +DULL = f"{BASE}/dull" +HINTED = f"{BASE}/read" +OTHER = f"{BASE}/other" + +SURFACE: Dict[str, Any] = { + "endpoints": [ + {"url": DULL, "method": "GET", "params": ["zzz"]}, + {"url": HINTED, "method": "GET", "params": ["path"]}, + {"url": OTHER, "method": "POST", "params": ["yyy"]}, + ] +} + + +def _order(priority: Any = None, surface: Any = None) -> List[str]: + """URLs in the order they were actually attacked.""" + hit: List[str] = [] + + def send(url: str, method: str, params: Dict[str, str]) -> str: + hit.append(url) + return "" + + exploit_surface( + surface if surface is not None else SURFACE, + sender=send, + max_requests=10000, + priority=priority, + ) + return list(dict.fromkeys(hit)) + + +def test_without_priority_the_name_hint_decides(): + assert _order()[0] == HINTED + + +def test_priority_outranks_the_name_hint(): + assert _order([(DULL, "GET")])[0] == DULL + + +def test_priority_order_is_followed(): + assert _order([(OTHER, "POST"), (DULL, "GET")])[:2] == [OTHER, DULL] + + +def test_unprioritised_endpoints_follow_in_promise_order(): + assert _order([(OTHER, "POST")]) == [OTHER, HINTED, DULL] + + +def test_method_is_part_of_the_match(): + # Same URL, wrong method: not a match, so the hint ordering stands. + assert _order([(DULL, "POST")])[0] == HINTED + + +def test_malformed_priority_is_ignored(): + for bad in ([], None, [None, "nope", (), ("", "GET"), (DULL,)]): + assert _order(bad)[0] == HINTED + + +def test_duplicate_priority_entries_keep_the_first_position(): + assert _order([(OTHER, "POST"), (DULL, "GET"), (OTHER, "POST")])[:2] == [OTHER, DULL] + + +def test_priority_decides_what_a_small_budget_reaches(): + hit: List[str] = [] + report = exploit_surface( + SURFACE, + sender=lambda u, m, p: hit.append(u) or "", + max_requests=1, + priority=[(DULL, "GET")], + ) + assert hit == [DULL] + assert report.budget_exhausted is True From ce6b3fe698a5e9b0b6d69e9f3c72a757cfb5c0cf Mon Sep 17 00:00:00 2001 From: Evgeny Kiriyak <224408464+evkir@users.noreply.github.com> Date: Wed, 29 Jul 2026 08:44:38 +0200 Subject: [PATCH 4/4] feat(exploit): honour the plan when attacking the web surface --- cyberai/agents/exploit/agent.py | 32 +++++++++++++++-- cyberai/core/config.py | 3 ++ tests/unit/test_plan_consumption.py | 55 +++++++++++++++++++++++++++++ 3 files changed, 88 insertions(+), 2 deletions(-) diff --git a/cyberai/agents/exploit/agent.py b/cyberai/agents/exploit/agent.py index 685b5c7..b658a61 100644 --- a/cyberai/agents/exploit/agent.py +++ b/cyberai/agents/exploit/agent.py @@ -4,7 +4,7 @@ import json import re -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from rich.console import Console from rich.table import Table @@ -173,7 +173,7 @@ def _run_web_exploit(self, target: str) -> Dict[str, Any]: self._log("No HTTP surface in KB — web exploitation skipped") return {"confirmed": 0, "endpoints_tested": 0, "findings": []} - report = exploit_surface(surface) + report = exploit_surface(surface, priority=self._plan_web_priority()) self.kb.set("exploit.web", report.to_dict(), agent=self.AGENT_NAME) for wf in report.findings: @@ -198,6 +198,34 @@ def _run_web_exploit(self, target: str) -> Dict[str, Any]: ) return report.to_dict() + def _plan_web_priority(self) -> Optional[List[Tuple[str, str]]]: + """Endpoints the planner named, as (url, method) pairs to attack first. + + The planner decides which endpoints are worth reaching from the KB + graph; how hard each one is pressed stays with the exploitation layer. + Returns None when the flag is off or the plan says nothing usable, so + the deterministic order is what runs unless a plan improves on it. + """ + if not getattr(self.config, "use_plan_web_order", False): + return None + plan = self.kb.get("plan") or {} + todo = plan.get("todo") if isinstance(plan, dict) else None + if not isinstance(todo, list): + return None + + priority: List[Tuple[str, str]] = [] + for task in todo: + if not isinstance(task, dict) or task.get("action") != "web-exploit": + continue + url = task.get("target") + if not url: + continue + priority.append((str(url), str(task.get("method") or "GET").upper())) + if not priority: + return None + self._log(f"plan order applied to {len(priority)} web endpoint(s)") + return priority + def _apply_plan_order(self, ranked_cves: List[Dict]) -> List[Dict]: """Reorder CVEs to follow the planner TODO when a plan is present. diff --git a/cyberai/core/config.py b/cyberai/core/config.py index 59af209..2d9388f 100644 --- a/cyberai/core/config.py +++ b/cyberai/core/config.py @@ -120,6 +120,8 @@ class CyberAIConfig: use_web_exploit: bool = False # Flag-gated: read spec and JS bundles when the HTML shell exposes nothing. use_api_discovery: bool = False + # Flag-gated: attack the endpoints the planner named before the rest. + use_plan_web_order: bool = False exploit_memory_path: Optional[str] = None # Flag-gated: parse local practice-lab machine artifacts and detect flags. use_lab_dogfood: bool = False @@ -181,6 +183,7 @@ def from_env(cls) -> "CyberAIConfig": use_web_recon=_env_bool("CYBERAI_USE_WEB_RECON", False), use_web_exploit=_env_bool("CYBERAI_USE_WEB_EXPLOIT", False), use_api_discovery=_env_bool("CYBERAI_USE_API_DISCOVERY", False), + use_plan_web_order=_env_bool("CYBERAI_USE_PLAN_WEB_ORDER", False), 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), diff --git a/tests/unit/test_plan_consumption.py b/tests/unit/test_plan_consumption.py index 3fed2eb..5b8d57f 100644 --- a/tests/unit/test_plan_consumption.py +++ b/tests/unit/test_plan_consumption.py @@ -67,3 +67,58 @@ def test_malformed_plan_keeps_order(): assert _ids(_agent("not-a-dict")._apply_plan_order(_cves())) == BASE assert _ids(_agent({"todo": []})._apply_plan_order(_cves())) == BASE assert _ids(_agent({"todo": "nope"})._apply_plan_order(_cves())) == BASE + + +# ── web-exploit ordering ────────────────────────────────────────────── + + +def _web_agent(plan: Any = None, flag: bool = True) -> ExploitAgent: + session = ScanSession(target="acme.tld") + if plan is not None: + session.kb_set("plan", plan) + return ExploitAgent(CyberAIConfig(use_plan_web_order=flag), session) + + +WEB_PLAN = { + "todo": [ + {"action": "exploit", "target": "CVE-1"}, + {"action": "web-exploit", "target": "http://acme.tld/b", "method": "post"}, + {"action": "web-exploit", "target": "http://acme.tld/a"}, + ] +} + + +def test_web_priority_follows_plan_order(): + assert _web_agent(WEB_PLAN)._plan_web_priority() == [ + ("http://acme.tld/b", "POST"), + ("http://acme.tld/a", "GET"), + ] + + +def test_web_priority_off_by_default(): + assert ( + ExploitAgent(CyberAIConfig(), ScanSession(target="acme.tld"))._plan_web_priority() is None + ) + assert _web_agent(WEB_PLAN, flag=False)._plan_web_priority() is None + + +def test_web_priority_none_without_usable_plan(): + assert _web_agent()._plan_web_priority() is None + assert _web_agent("not-a-dict")._plan_web_priority() is None + assert _web_agent({"todo": "nope"})._plan_web_priority() is None + assert _web_agent({"todo": []})._plan_web_priority() is None + assert ( + _web_agent({"todo": [{"action": "exploit", "target": "CVE-1"}]})._plan_web_priority() + is None + ) + + +def test_web_priority_skips_malformed_entries(): + plan = { + "todo": [ + None, + {"action": "web-exploit", "target": None}, + {"action": "web-exploit", "target": "http://acme.tld/a"}, + ] + } + assert _web_agent(plan)._plan_web_priority() == [("http://acme.tld/a", "GET")]