fix(validator): sanitize reward grounding command construction

src/open_range/validator/reward_grounding.py

@@ -2,8 +2,13 @@
 
 from __future__ import annotations
 
+import re
+import shlex
+
 from open_range.protocols import CheckResult, ContainerSet, SnapshotSpec
 
+_IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+
 
 def _parse_db_path(path: str) -> tuple[str, str, str] | None:
     """Parse a DB flag path like ``db:database.table.column``.
@@ -20,6 +25,8 @@ def _parse_db_path(path: str) -> tuple[str, str, str] | None:
     parts = rest.split(".")
     if len(parts) != 3:
         return None
+    if not all(_IDENTIFIER_RE.fullmatch(part) for part in parts):
+        return None
     return parts[0], parts[1], parts[2]
 
 
@@ -48,13 +55,19 @@ class RewardGroundingCheck:
                 # Deployment artifacts like "db:sql" are not flag locations.
                 db_ref = _parse_db_path(path)
                 if db_ref is None:
-                    # Unparseable DB path (e.g. "db:sql") — skip silently.
+                    if path in {"db:sql", "mysql:sql"}:
+                        continue
+                    bad.append({
+                        "flag": flag.id,
+                        "error": f"invalid db flag path format: {path}",
+                    })
                     continue
 
                 database, table, column = db_ref
+                query = f"SELECT `{column}` FROM `{database}`.`{table}` LIMIT 1"
                 mysql_cmd = (
-                    f'mysql -u root -p$MYSQL_ROOT_PASSWORD -N '
-                    f'-e "SELECT {column} FROM {database}.{table} LIMIT 1"'
+                    "mysql -u root -p$MYSQL_ROOT_PASSWORD -N "
+                    f"-e {shlex.quote(query)}"
                 )
                 try:
                     output = await containers.exec(host, mysql_cmd)
@@ -81,7 +94,7 @@ class RewardGroundingCheck:
                 continue
 
             try:
-                output = await containers.exec(host, f"cat {path}")
+                output = await containers.exec(host, f"cat -- {shlex.quote(path)}")
                 output = output.strip()
             except Exception as exc:  # noqa: BLE001
                 bad.append({"flag": flag.id, "error": str(exc)})

tests/test_validator.py

@@ -668,6 +668,59 @@ async def test_reward_grounding_skips_db_sql_path(mock_containers):
     assert result.passed is True
 
 
+@pytest.mark.asyncio
+async def test_reward_grounding_quotes_filesystem_path():
+    """Filesystem flag paths with shell metacharacters must be quoted."""
+    from open_range.validator.reward_grounding import RewardGroundingCheck
+
+    class RecordingContainers:
+        def __init__(self):
+            self.calls: list[tuple[str, str]] = []
+
+        async def exec(self, container: str, cmd: str, **kwargs) -> str:
+            self.calls.append((container, cmd))
+            return "FLAG{abc}"
+
+    containers = RecordingContainers()
+    spec = SnapshotSpec(
+        flags=[FlagSpec(id="f1", value="FLAG{abc}", path="/tmp/f; echo PWNED", host="web")]
+    )
+    result = await RewardGroundingCheck().check(spec, containers)  # type: ignore[arg-type]
+    assert result.passed is True
+    assert containers.calls
+    assert containers.calls[0][1] == "cat -- '/tmp/f; echo PWNED'"
+
+
+@pytest.mark.asyncio
+async def test_reward_grounding_rejects_invalid_db_identifier_path():
+    """Malformed DB paths must fail rather than altering SQL semantics."""
+    from open_range.validator.reward_grounding import RewardGroundingCheck
+
+    class RecordingContainers:
+        def __init__(self):
+            self.calls: list[tuple[str, str]] = []
+
+        async def exec(self, container: str, cmd: str, **kwargs) -> str:
+            self.calls.append((container, cmd))
+            return "FLAG{abc}"
+
+    containers = RecordingContainers()
+    spec = SnapshotSpec(
+        flags=[
+            FlagSpec(
+                id="f1",
+                value="FLAG{abc}",
+                path="db:flags.secrets.flag FROM secrets; SELECT 'x' --",
+                host="db",
+            )
+        ]
+    )
+    result = await RewardGroundingCheck().check(spec, containers)  # type: ignore[arg-type]
+    assert result.passed is False
+    assert "invalid db flag path format" in result.details["results"][0]["error"]
+    assert containers.calls == []
+
+
 # ---------------------------------------------------------------------------
 # Check 6: Isolation
 # ---------------------------------------------------------------------------