feat(cyber): webapp.secret_leak — NPC comms become the agent's objective

examples/secret_leak_eval.py

@@ -0,0 +1,119 @@
+"""Tutorial: a persona NPC's message becomes the agent's objective.
+
+Runs a live `cyber_webapp` episode of the `webapp.secret_leak` family twice on the
+same world — once with the social NPC speaking, once staying silent — to show that
+the NPC's in-world message is what makes the task winnable. The agent only ever
+reaches the app over HTTP, yet it recovers a secret an NPC posted to the team chat.
+
+No model required: the NPC is driven by a scripted backend so the demo is
+deterministic. Run: `uv run python examples/secret_leak_eval.py`
+"""
+
+from __future__ import annotations
+
+import json
+import tempfile
+from pathlib import Path
+from typing import Any
+from urllib.request import urlopen
+
+from cyber_webapp import WebappPack, WebappSecretLeak
+from openrange_pack_sdk import Snapshot
+
+from openrange.core.admit import admit
+from openrange.core.episode import EpisodeService
+
+MANIFEST = {
+    "world": {"goal": "recover a leaked secret"},
+    "pack": {"id": "webapp"},
+    "runtime": {"tick": {"mode": "off"}},
+    "npc": [
+        {
+            "type": "cyber.persona",
+            "config": {
+                "name": "Dana",
+                "role": "sysadmin",
+                "goal": "remind a colleague of the rotated admin key",
+                "tools": ["chat_post"],
+                "channels": ["team"],
+                "cadence_ticks": 1,
+            },
+        }
+    ],
+}
+
+
+class ScriptedNPC:
+    """Drives the persona deterministically: it posts the secret, or (silent=True)
+    says nothing — the A/B knob for this demo."""
+
+    def __init__(self, secret: str, *, silent: bool) -> None:
+        self._secret = secret
+        self._silent = silent
+
+    def preflight(self) -> None:
+        pass
+
+    def build_agent(self, *, system_prompt: str, tools: Any = ()) -> Any:
+        post = next(
+            (t for t in tools if getattr(t, "__name__", "") == "chat_post"), None
+        )
+
+        def agent(prompt: str) -> object:
+            if post is not None and not self._silent:
+                post(channel="team", text=f"reminder: the admin key is {self._secret}")
+            return {"message": "ok"}
+
+        return agent
+
+
+def run(
+    secret: str, task_id: str, snapshot: Snapshot, *, silent: bool
+) -> tuple[list[Any], bool]:
+    """Run one episode; return the served chat board and whether the secret is
+    recoverable from it (the agent's view)."""
+    svc = EpisodeService(
+        WebappPack(),
+        Path(tempfile.mkdtemp()),
+        npc_agent_backend=ScriptedNPC(secret, silent=silent),
+    )
+    try:
+        handle = svc.start_episode(snapshot, task_id)
+        svc.tick(handle)  # the persona acts (or stays silent)
+        board = json.loads(
+            urlopen(svc.base_url(handle) + "/team/chat", timeout=5).read()
+        )
+        recovered = any(secret in str(m.get("text", "")) for m in board)
+        svc.stop_episode(handle)
+        return board, recovered
+    finally:
+        svc.close()
+
+
+def main() -> int:
+    snapshot = admit(WebappPack(), MANIFEST)
+    assert isinstance(snapshot, Snapshot), snapshot
+    task = next(
+        t for t in snapshot.tasks if t.meta.get("family") == "webapp.secret_leak"
+    )
+    secret = snapshot.graph.nodes[task.goal_nodes[0]].attrs["value_ref"]
+    print(
+        f"task: {task.id}\ninstruction: {task.instruction}\nhidden secret: {secret}\n"
+    )
+
+    for label, silent in (("Dana speaks", False), ("Dana silent", True)):
+        board, recovered = run(secret, task.id, snapshot, silent=silent)
+        print(f"[{label}]  GET /team/chat -> {board}")
+        print(f"            secret recoverable by the agent: {recovered}\n")
+
+    print(
+        "The world, task, and agent are identical across both runs — the only\n"
+        "difference is whether the inhabitant spoke. The NPC's message is the\n"
+        "episode's win condition (graded via the attributed npc_chat store, which\n"
+        f"the agent cannot forge — see {WebappSecretLeak.__name__}.check_success)."
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

packs/cyber_webapp/cyber_webapp/__init__.py

@@ -15,7 +15,7 @@
 
 from cyber_webapp.builder import WebappBuilder
 from cyber_webapp.container import minimum_backing
-from cyber_webapp.families import WebappBuild, WebappPentest
+from cyber_webapp.families import WebappBuild, WebappPentest, WebappSecretLeak
 from cyber_webapp.invariants import (
     credential_reuse_binding,
     credential_value_binding,
@@ -83,7 +83,7 @@ def minimum_backing(self, graph: WorldGraph) -> Backing:
         return minimum_backing(graph)
 
     def task_families(self) -> list[TaskFamily]:
-        return [WebappBuild(), WebappPentest()]
+        return [WebappBuild(), WebappPentest(), WebappSecretLeak()]
 
 
 __all__ = [
@@ -95,6 +95,7 @@ def task_families(self) -> list[TaskFamily]:
     "WebappPack",
     "WebappPentest",
     "WebappRuntimeError",
+    "WebappSecretLeak",
     "WebappRuntime",
     "credential_reuse_binding",
     "credential_value_binding",

packs/cyber_webapp/cyber_webapp/builder.py

@@ -16,7 +16,7 @@
 )
 
 from cyber_webapp.difficulty import world_difficulty
-from cyber_webapp.families import WebappBuild, WebappPentest
+from cyber_webapp.families import WebappBuild, WebappPentest, WebappSecretLeak
 from cyber_webapp.priors import default_prior
 from cyber_webapp.sampling import (
     _DEFAULT_LOOT_WEIGHTS,
@@ -103,6 +103,7 @@ def sample(self, rng: random.Random, manifest: Manifest) -> BuildResult:
         tasks: list[TaskSpec] = []
         tasks.extend(WebappBuild().generate(graph, manifest, prior))
         tasks.extend(WebappPentest().generate(graph, manifest, prior))
+        tasks.extend(WebappSecretLeak().generate(graph, manifest, prior))
         return BuildResult(
             graph=graph,
             tasks=tasks,

packs/cyber_webapp/cyber_webapp/codegen/templates/app.py.j2

@@ -155,13 +155,32 @@ def _openapi(query, state):
     )
 
 
+# A shared team chat board. NPCs post cover traffic (and, in a leak scenario, a
+# secret) via POST; a direct request reads it back via GET. This is how an NPC's
+# in-world message becomes observable to an agent that can only reach the app.
+def _team_chat(query, state):
+    del query
+    return (200, {"Content-Type": "application/json"},
+            json.dumps(state.setdefault("team_chat", [])).encode())
+
+
+def _team_chat_post(query, state):
+    text = (query.get("text") or [""])[0]
+    if text:
+        state.setdefault("team_chat", []).append(
+            {"sender": (query.get("sender") or ["anon"])[0], "text": text})
+    return (200, {"Content-Type": "application/json"}, b'{"ok": true}')
+
+
 ROUTES = {
 {% for route in routes %}
     {{ route.path | tojson }}: {{ route.handler }},
 {% endfor %}
 }
 ROUTES.setdefault("/", _index)
 ROUTES.setdefault("/openapi.json", _openapi)
+ROUTES.setdefault("/team/chat", _team_chat)
+ROUTES.setdefault("/team/chat/post", _team_chat_post)
 
 # Internal endpoints, dispatched to only by the SSRF proxy — never served to a direct
 # request. This table enforces network segmentation when one process holds the whole
@@ -180,6 +199,7 @@ _ROUTE_METHODS = {
     {{ route.path | tojson }}: {{ (route.method | default("GET")) | tojson }},
 {% endfor %}
 }
+_ROUTE_METHODS.setdefault("/team/chat/post", "POST")
 
 
 class Handler(BaseHTTPRequestHandler):

packs/cyber_webapp/cyber_webapp/families/__init__.py

@@ -1,6 +1,7 @@
-"""TaskFamilies for the cyber webapp pack: webapp.build and webapp.pentest."""
+"""TaskFamilies for the cyber webapp pack: build, pentest, secret_leak."""
 
 from cyber_webapp.families.build import WebappBuild
 from cyber_webapp.families.pentest import WebappPentest
+from cyber_webapp.families.secret_leak import WebappSecretLeak
 
-__all__ = ["WebappBuild", "WebappPentest"]
+__all__ = ["WebappBuild", "WebappPentest", "WebappSecretLeak"]

packs/cyber_webapp/cyber_webapp/families/secret_leak.py

@@ -0,0 +1,118 @@
+"""`webapp.secret_leak` TaskFamily — recover a secret an NPC leaked into team chat.
+
+Unlike `webapp.pentest` (exploit a vuln chain), the win path here runs through
+the world's inhabitants: a persona NPC posts the flag into the shared `/team/chat`
+board (see the runtime bridge), and the agent recovers it from there. This is the
+family that makes persona cover-traffic observably change an episode — success
+depends on an NPC having spoken.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from graphschema import Node, Visibility, WorldGraph
+from openrange_pack_sdk import (
+    EpisodeResult,
+    FeasibilityVerdict,
+    Manifest,
+    PackPrior,
+    TaskFamily,
+    TaskSpec,
+)
+
+
+class WebappSecretLeak(TaskFamily):
+    id = "webapp.secret_leak"
+    pack_id = "webapp"
+
+    def generate(
+        self,
+        graph: WorldGraph,
+        manifest: Manifest,
+        prior: PackPrior | None,
+    ) -> list[TaskSpec]:
+        del prior
+        # Only offer the task when the world has an inhabitant that could leak —
+        # without an NPC the board stays empty and the task is unwinnable.
+        flag = self._find_flag_secret(graph)
+        entry = self._any_public_endpoint(graph)
+        if flag is None or entry is None or not manifest.get("npc"):
+            return []
+        return [
+            self.make_task(
+                instruction=(
+                    "A teammate leaked a secret into the internal chat board. Read "
+                    "GET /team/chat, recover the secret, and submit it."
+                ),
+                # The board rides the same app as this endpoint; the binding just
+                # gives the agent a foothold URL, the leak lives at /team/chat.
+                entrypoints=entry.id,
+                goal_nodes=flag.id,
+                difficulty=0.2,
+                meta={"flag_secret": flag.id, "channel": "/team/chat"},
+            ),
+        ]
+
+    def check_feasibility(
+        self,
+        graph: WorldGraph,
+        task: TaskSpec,
+    ) -> FeasibilityVerdict:
+        if not task.goal_nodes:
+            return FeasibilityVerdict(False, "missing goal")
+        flag = graph.nodes.get(task.goal_nodes[0])
+        if flag is None:
+            return FeasibilityVerdict(False, "goal node not in graph")
+        # The graph can host the task once it has a secret to leak; whether an NPC
+        # actually leaks it is a manifest property this hook can't see (it becomes
+        # the npc_leaked_secret subgoal at grading time).
+        if flag.attrs.get("value_ref") in (None, ""):
+            return FeasibilityVerdict(False, "flag has no value to leak")
+        return FeasibilityVerdict(True)
+
+    def check_success(
+        self,
+        graph: WorldGraph,
+        task: TaskSpec,
+        final_state: Mapping[str, Any],
+    ) -> EpisodeResult:
+        if not task.goal_nodes:
+            return EpisodeResult(success=False, reason="no goal node")
+        flag = graph.nodes.get(task.goal_nodes[0])
+        if flag is None:
+            return EpisodeResult(success=False, reason="goal not in graph")
+        expected = flag.attrs.get("value_ref")
+        submitted = final_state.get("flag_from_response")
+        ok = bool(expected) and submitted == expected
+        # The observability subgoal reads the attributed NPC chat directly — the
+        # agent can't forge it (it has no actor_id in the store), so this is a
+        # clean, separable signal that the leak actually happened in-world.
+        leaked_in_chat = any(
+            isinstance(m, Mapping) and expected and expected in str(m.get("text", ""))
+            for m in (final_state.get("npc_chat") or ())
+        )
+        return EpisodeResult(
+            success=ok,
+            subgoals={
+                "read_the_board": bool(final_state.get("requests_made")),
+                "npc_leaked_secret": leaked_in_chat,
+                "recovered_secret": ok,
+            },
+            reason="secret recovered"
+            if ok
+            else f"expected {expected!r}, got {submitted!r}",
+        )
+
+    def _find_flag_secret(self, graph: WorldGraph) -> Node | None:
+        for s in graph.by_kind("secret"):
+            if s.attrs.get("kind") == "flag" and s.visibility is Visibility.HIDDEN:
+                return s
+        return None
+
+    def _any_public_endpoint(self, graph: WorldGraph) -> Node | None:
+        for e in graph.by_kind("endpoint"):
+            if e.visibility is Visibility.PUBLIC and e.attrs.get("public_url"):
+                return e
+        return None

packs/cyber_webapp/cyber_webapp/realize.py

@@ -14,6 +14,7 @@
 from pathlib import Path
 from typing import Any, cast
 from urllib.error import URLError
+from urllib.parse import urlencode
 from urllib.request import urlopen
 
 from graphschema import WorldGraph
@@ -83,11 +84,27 @@ def http_get(path: object) -> bytes:
         def http_get_json(path: object) -> object:
             return json.loads(http_get(path).decode())
 
+        chat = dict(surface_chat(self._chat))
+        store_post = chat["chat_post"]
+
+        def chat_post(sender: object, channel: object, text: object) -> object:
+            # Mirror the message onto the served /team/chat board so an agent that
+            # can only reach the app observes it. The in-process store stays the
+            # attributed source of truth; forwarding is best-effort.
+            result = store_post(sender, channel, text)
+            try:
+                body = urlencode({"sender": str(sender), "text": str(text)}).encode()
+                urlopen(base_url + "/team/chat/post", data=body, timeout=5).read()
+            except OSError:
+                pass
+            return result
+
+        chat["chat_post"] = chat_post
         return {
             "http_get": http_get,
             "http_get_json": http_get_json,
             **surface_mailbox(self._mailbox),
-            **surface_chat(self._chat),
+            **chat,
         }
 
     def poll_events(self) -> tuple[Mapping[str, Any], ...]:

tests/test_cyber_webapp_v2.py

@@ -481,7 +481,7 @@ def test_credential_binding_rejects_when_producer_cannot_reach_gate() -> None:
 
 
 def test_real_webapp_pack_identity() -> None:
-    """The pack registers under id `webapp`, ships two families."""
+    """The pack registers under id `webapp`, ships three families."""
     from cyber_webapp import WebappPack
 
     pack = WebappPack()
@@ -491,6 +491,7 @@ def test_real_webapp_pack_identity() -> None:
     assert {f.id for f in pack.task_families()} == {
         "webapp.build",
         "webapp.pentest",
+        "webapp.secret_leak",
     }