Tighten graph checks and trust-principal lint

src/open_range/lint.py

@@ -12,6 +12,7 @@ Usage::
 from __future__ import annotations
 
 import argparse
+import re
 import sys
 from pathlib import Path
 from typing import Any
@@ -134,22 +135,28 @@ def _check_business_process_flows(manifest: Manifest) -> list[str]:
     return errors
 
 
+_PRINCIPAL_RE = re.compile(r"^[A-Za-z0-9._@-]+$")
+
+
 def _check_trust_relationships(manifest: Manifest) -> list[str]:
-    """All trust relationships must reference valid usernames."""
-    user_names = {u.username for u in manifest.users}
+    """Trust principals must be well-formed identifiers.
+
+    Trust edges may reference people who are not login accounts. Those are
+    normalized into the canonical principal catalog at build time, so lint
+    should validate identifier quality rather than requiring every principal to
+    appear in ``users``.
+    """
     errors: list[str] = []
     for rel in manifest.trust_relationships:
-        if rel.source and rel.source not in user_names:
+        if rel.source and not _PRINCIPAL_RE.match(rel.source):
             errors.append(
-                f"Trust relationship source '{rel.source}' "
-                f"is not in the users list. "
-                f"Valid usernames: {sorted(user_names)}"
+                f"Trust relationship source '{rel.source}' is not a valid "
+                "principal identifier"
             )
-        if rel.target and rel.target not in user_names:
+        if rel.target and not _PRINCIPAL_RE.match(rel.target):
             errors.append(
-                f"Trust relationship target '{rel.target}' "
-                f"is not in the users list. "
-                f"Valid usernames: {sorted(user_names)}"
+                f"Trust relationship target '{rel.target}' is not a valid "
+                "principal identifier"
             )
     return errors
 
@@ -165,7 +172,7 @@ ALL_CHECKS = [
     ("NPC persona usernames", _check_npc_usernames),
     ("data inventory hosts", _check_data_inventory_hosts),
     ("business process data flows", _check_business_process_flows),
-    ("trust relationship usernames", _check_trust_relationships),
+    ("trust relationship principals", _check_trust_relationships),
 ]
 
 

src/open_range/validator/graph_reward_grounding.py

@@ -2,10 +2,9 @@
 
 from __future__ import annotations
 
-from collections import defaultdict, deque
-
 from open_range.protocols import CheckResult, ContainerSet, SnapshotSpec
 from open_range.validator.graphs import compile_snapshot_graphs
+from open_range.validator.path_solvability import build_host_adjacency, has_host_path
 
 
 class GraphRewardGroundingCheck:
@@ -22,7 +21,7 @@ class GraphRewardGroundingCheck:
                 error="snapshot has no flags to ground",
             )
 
-        adjacency = _adjacency(compiled)
+        adjacency = build_host_adjacency(snapshot, compiled)
         vuln_hosts = {v.host for v in snapshot.truth_graph.vulns if v.host}
         for flag in snapshot.flags:
             if flag.host not in compiled.hosts:
@@ -32,7 +31,9 @@ class GraphRewardGroundingCheck:
             if flag.host in vuln_hosts:
                 continue
 
-            if vuln_hosts and not any(_has_path(source, flag.host, adjacency) for source in vuln_hosts):
+            if vuln_hosts and not any(
+                has_host_path(source, flag.host, adjacency) for source in vuln_hosts
+            ):
                 issues.append(
                     f"flag '{flag.id}' on '{flag.host}' is not reachable from any vuln host"
                 )
@@ -44,27 +45,3 @@ class GraphRewardGroundingCheck:
             details={"issues": issues},
             error="" if passed else "; ".join(issues),
         )
-
-
-def _adjacency(compiled) -> dict[str, set[str]]:
-    adjacency: dict[str, set[str]] = defaultdict(set)
-    for source, target in compiled.dependency_edges:
-        adjacency[source].add(target)
-    return adjacency
-
-
-def _has_path(start: str, target: str, adjacency: dict[str, set[str]]) -> bool:
-    if start == target:
-        return True
-    queue: deque[str] = deque([start])
-    seen = {start}
-    while queue:
-        current = queue.popleft()
-        for neighbor in adjacency.get(current, set()):
-            if neighbor == target:
-                return True
-            if neighbor in seen:
-                continue
-            seen.add(neighbor)
-            queue.append(neighbor)
-    return False

src/open_range/validator/path_solvability.py

@@ -33,7 +33,7 @@ class PathSolvabilityCheck:
                 error="snapshot has no vuln or flag hosts to solve toward",
             )
 
-        adjacency = _adjacency(compiled)
+        adjacency = build_host_adjacency(snapshot, compiled)
         unreachable = [
             host
             for host in target_hosts
@@ -78,13 +78,64 @@ def _start_hosts(compiled: CompiledGraphs) -> set[str]:
     return set()
 
 
-def _adjacency(compiled: CompiledGraphs) -> dict[str, set[str]]:
+def build_host_adjacency(
+    snapshot: SnapshotSpec,
+    compiled: CompiledGraphs,
+) -> dict[str, set[str]]:
     adjacency: dict[str, set[str]] = defaultdict(set)
     for source, target in compiled.dependency_edges:
         adjacency[source].add(target)
+
+    principal_hosts = _principal_hosts(snapshot)
+    for source_principal, target_principal, _edge_type in compiled.trust_edges:
+        source_hosts = principal_hosts.get(source_principal, set())
+        target_hosts = principal_hosts.get(target_principal, set())
+        for source_host in source_hosts:
+            for target_host in target_hosts:
+                if source_host and target_host:
+                    adjacency[source_host].add(target_host)
     return adjacency
 
 
+def has_host_path(
+    start: str,
+    target: str,
+    adjacency: dict[str, set[str]],
+) -> bool:
+    return _has_path(start, target, adjacency)
+
+
+def _principal_hosts(snapshot: SnapshotSpec) -> dict[str, set[str]]:
+    topology = snapshot.topology or {}
+    mapping: dict[str, set[str]] = defaultdict(set)
+
+    raw_users = topology.get("users", [])
+    if isinstance(raw_users, list):
+        for raw in raw_users:
+            if not isinstance(raw, dict):
+                continue
+            username = str(raw.get("username", "")).strip()
+            if not username:
+                continue
+            for raw_host in raw.get("hosts", []):
+                host = str(raw_host).strip()
+                if host:
+                    mapping[username].add(host)
+
+    raw_catalog = topology.get("principal_catalog", {})
+    if isinstance(raw_catalog, dict):
+        for raw_name, raw_principal in raw_catalog.items():
+            name = str(raw_name).strip()
+            if not name or not isinstance(raw_principal, dict):
+                continue
+            for raw_host in raw_principal.get("hosts", []):
+                host = str(raw_host).strip()
+                if host:
+                    mapping[name].add(host)
+
+    return mapping
+
+
 def _reachable_from_any(
     target: str,
     starts: set[str],

tests/test_lint.py

@@ -106,9 +106,10 @@ class TestValidManifest:
             assert errors == [], f"Check '{check_name}' failed: {errors}"
 
     def test_tier1_manifest_loads(self):
-        """Tier 1 manifest should at least load without schema error."""
+        """Tier 1 manifest should load and pass lint checks."""
         result = lint_file(ROOT / "manifests" / "tier1_basic.yaml")
         assert result["schema_error"] is None, result["schema_error"]
+        assert result["valid"] is True, result["checks"]
 
 
 # ---------------------------------------------------------------------------
@@ -177,35 +178,35 @@ class TestInvalidUserRefs:
         assert len(errors) == 1
         assert "ghost_user" in errors[0]
 
-    def test_trust_relationship_invalid_source(self):
+    def test_trust_relationship_invalid_source_identifier(self):
         data = _minimal_manifest()
         data["trust_relationships"] = [
             {
                 "type": "delegates_access",
-                "from": "nobody",
+                "from": "bad actor!",
                 "to": "admin",
             },
         ]
         manifest = Manifest(**data)
         results = lint_manifest(manifest)
-        errors = results["trust relationship usernames"]
+        errors = results["trust relationship principals"]
         assert len(errors) == 1
-        assert "nobody" in errors[0]
+        assert "bad actor!" in errors[0]
 
-    def test_trust_relationship_invalid_target(self):
+    def test_trust_relationship_invalid_target_identifier(self):
         data = _minimal_manifest()
         data["trust_relationships"] = [
             {
                 "type": "delegates_access",
                 "from": "admin",
-                "to": "phantom",
+                "to": "phantom user",
             },
         ]
         manifest = Manifest(**data)
         results = lint_manifest(manifest)
-        errors = results["trust relationship usernames"]
+        errors = results["trust relationship principals"]
         assert len(errors) == 1
-        assert "phantom" in errors[0]
+        assert "phantom user" in errors[0]
 
 
 # ---------------------------------------------------------------------------

tests/test_validator.py

@@ -165,6 +165,42 @@ async def test_graph_reward_grounding_fails_when_flag_host_unreachable(mock_cont
     assert "not reachable from any vuln host" in result.error
 
 
+@pytest.mark.asyncio
+async def test_graph_checks_allow_trust_based_host_pivots(mock_containers):
+    from open_range.validator.graph_reward_grounding import GraphRewardGroundingCheck
+    from open_range.validator.path_solvability import PathSolvabilityCheck
+
+    spec = SnapshotSpec(
+        topology={
+            "hosts": ["attacker", "web", "db"],
+            "zones": {"external": ["attacker"], "dmz": ["web"], "internal": ["db"]},
+            "dependency_edges": [{"source": "attacker", "target": "web"}],
+            "trust_edges": [{"source": "websvc", "target": "dbsvc", "type": "credential_reuse"}],
+            "host_details": {
+                "attacker": {"services": ["nmap"]},
+                "web": {"services": ["nginx"]},
+                "db": {"services": ["mysql"]},
+            },
+            "principal_catalog": {
+                "websvc": {"username": "websvc", "hosts": ["web"], "is_login_account": False},
+                "dbsvc": {"username": "dbsvc", "hosts": ["db"], "is_login_account": False},
+            },
+        },
+        truth_graph=TruthGraph(
+            vulns=[Vulnerability(id="v1", type="credential_reuse", host="web", service="nginx")],
+        ),
+        flags=[FlagSpec(id="f1", value="FLAG{db}", path="/var/flags/flag1.txt", host="db")],
+        evidence_spec=[EvidenceItem(type="log_entry", location="db:/var/log/mysql.log")],
+        golden_path=[GoldenPathStep(step=1, command="scan", expect_in_stdout="ok")],
+        task=TaskSpec(red_briefing="go", blue_briefing="watch"),
+    )
+
+    path_result = await PathSolvabilityCheck().check(spec, mock_containers)
+    reward_result = await GraphRewardGroundingCheck().check(spec, mock_containers)
+    assert path_result.passed is True
+    assert reward_result.passed is True
+
+
 # ---------------------------------------------------------------------------
 # Check 1: BuildBoot
 # ---------------------------------------------------------------------------