Feat/plan web surface

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.
 

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()

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(
                 {

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),

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,13 +27,15 @@
 SERVICE = "service"
 CVE = "cve"
 LLM_ENDPOINT = "llm_endpoint"
+HTTP_ENDPOINT = "http_endpoint"
 
 # Edge relations.
 HAS_PORT = "HAS_PORT"
 RUNS = "RUNS"
 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}
 

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) == []