fix: scanner false positives & symlink dodge; perf: policy cache

CHANGELOG.md

@@ -2,6 +2,35 @@
 
 All notable changes to AgentShield are documented in this file.
 
+## Unreleased
+
+### Fixed
+
+- **Scanner false positives**: the `sk-` key pattern had no word boundary or
+  minimum length, so ordinary words like `task-123` or `risk-2026` matched and
+  were blocked as `sensitive_data` — killing LLM-loop model calls whose
+  serialized payload contains the user task text. The pattern now mirrors the
+  redaction patterns in `model/dots_client.py` (`\bsk-[a-zA-Z0-9_-]{8,}\b`).
+- **Symlink dodge of `blocked_files`**: blocked-name matching only saw the
+  caller-supplied path, so a benign-named symlink (e.g. `notes.txt` → `.env`)
+  could reach a blocked target. The middleware now checks the resolved path
+  against the blocked-name list too.
+
+### Changed
+
+- `load_policy` caches parsed policies per path, revalidated against the
+  file's `(st_mtime_ns, st_size)` fingerprint — ~5.5x faster on hot
+  middleware loops with identical on-disk-change semantics; errors are never
+  cached.
+- The deterministic demo (`python main.py` without `--llm`) now exits with
+  code 2 when a call is blocked, matching the `--llm` mode.
+
+### Added
+
+- Regression tests for the scanner fix (unit and model-payload level), the
+  symlink dodge (skipped where the platform cannot create symlinks), and the
+  policy cache (change detection, object reuse).
+
 ## v1.6.1 — 2026-10-01
 
 ### Fixed

main.py

@@ -30,8 +30,10 @@ def main() -> int:
     args = parser.parse_args()
 
     if not args.llm:
-        agent(args.task, policy_path=args.policy, audit_path=args.audit_log)
-        return 0
+        completed = agent(
+            args.task, policy_path=args.policy, audit_path=args.audit_log
+        )
+        return 0 if completed else 2
 
     try:
         client = DotsClient(base_url=args.base_url, model=args.model)

security/middleware.py

@@ -3,6 +3,7 @@
 from collections.abc import Mapping
 from dataclasses import dataclass
 from os import PathLike
+from pathlib import Path
 from typing import Any
 
 from security.audit import DEFAULT_AUDIT_PATH, write_audit_event
@@ -113,7 +114,13 @@ def check_tool_call(
                 audit_path=audit_path,
                 capability=capability,
             )
-        if policy.blocks_file(path):
+        # Check the blocked-name list against the resolved path as well: a
+        # symlink with an innocuous name must not dodge a blocked target name.
+        resolved_path = Path(path)
+        if not resolved_path.is_absolute():
+            resolved_path = policy.base_directory / resolved_path
+        resolved_path = resolved_path.resolve()
+        if policy.blocks_file(path) or policy.blocks_file(str(resolved_path)):
             return _finish_decision(
                 allowed=False,
                 reason="blocked_file",
@@ -123,7 +130,7 @@ def check_tool_call(
                 audit_path=audit_path,
                 capability=capability,
             )
-        if not policy.allows_file(path):
+        if not policy.allows_file(resolved_path):
             return _finish_decision(
                 allowed=False,
                 reason="file_outside_allowed_roots",

security/policy.py

@@ -66,13 +66,41 @@ def allows_url(self, url: str) -> bool:
         return False
 
 
+#: Parsed policies keyed by path, validated against (mtime_ns, size) on each
+#: load. A stat is far cheaper than a re-read for hot middleware loops, and
+#: any write to the policy file changes the fingerprint and forces a reload.
+_POLICY_CACHE: dict[Path, tuple[tuple[int, int], SecurityPolicy]] = {}
+
+
 def load_policy(path: str | Path = DEFAULT_POLICY_PATH) -> SecurityPolicy:
     """Load the small, dependency-free YAML subset used by V1.5.
 
     The supported shape is three top-level keys whose values are string lists.
     Rejecting unknown syntax keeps this prototype predictable and fail-closed.
+
+    Results are cached per path and revalidated against the file's
+    ``(st_mtime_ns, st_size)`` fingerprint, so every call still observes a
+    policy file edited or replaced on disk. Parse and I/O errors are never
+    cached.
     """
     policy_path = Path(path)
+    try:
+        stat_result = policy_path.stat()
+    except OSError:
+        _POLICY_CACHE.pop(policy_path, None)
+        raise
+    fingerprint = (stat_result.st_mtime_ns, stat_result.st_size)
+
+    cached = _POLICY_CACHE.get(policy_path)
+    if cached is not None and cached[0] == fingerprint:
+        return cached[1]
+
+    policy = _parse_policy(policy_path)
+    _POLICY_CACHE[policy_path] = (fingerprint, policy)
+    return policy
+
+
+def _parse_policy(policy_path: Path) -> SecurityPolicy:
     values: dict[str, list[str]] = {
         "blocked_files": [],
         "allowed_file_roots": [],

security/scanner.py

@@ -3,18 +3,21 @@
 import re
 
 
-PATTERNS = (
-    r"sk-[a-zA-Z0-9]+",
-    r"\bak_[a-zA-Z0-9]{16,}\b",
-    r"-----BEGIN OPENSSH PRIVATE KEY-----",
-    r"AWS_SECRET",
-    r"PASSWORD",
+# Word boundaries and minimum lengths keep ordinary words that merely end in
+# "sk" (task-42, risk-2026) from matching; the shape mirrors the redaction
+# patterns in model.dots_client so scan and redact agree on what a key is.
+PATTERNS: tuple[re.Pattern[str], ...] = tuple(
+    re.compile(pattern, flags=re.IGNORECASE)
+    for pattern in (
+        r"\bsk-[a-zA-Z0-9_-]{8,}\b",
+        r"\bak_[a-zA-Z0-9]{16,}\b",
+        r"-----BEGIN OPENSSH PRIVATE KEY-----",
+        r"AWS_SECRET",
+        r"PASSWORD",
+    )
 )
 
 
 def scan_sensitive(data: str) -> bool:
     """Return True when data matches any configured sensitive pattern."""
-    return any(
-        re.search(pattern, data, flags=re.IGNORECASE) is not None
-        for pattern in PATTERNS
-    )
+    return any(pattern.search(data) is not None for pattern in PATTERNS)

test/test_security.py

@@ -32,6 +32,96 @@ def test_scanner_detects_secret_markers(self) -> None:
         self.assertTrue(scan_sensitive("ak_1234567890abcdef"))
         self.assertFalse(scan_sensitive("service started successfully"))
 
+    def test_scanner_does_not_flag_words_containing_sk(self) -> None:
+        """Ordinary words ending in "sk" must not match the key pattern."""
+        self.assertFalse(scan_sensitive("task-1234567890"))
+        self.assertFalse(scan_sensitive("risk-2026"))
+        self.assertFalse(scan_sensitive("desk-42 status report"))
+        self.assertTrue(scan_sensitive("sk-1234567890abcdef"))
+        self.assertTrue(scan_sensitive("SK-abcdefghijklmnop"))
+
+    def test_model_payload_with_task_id_is_not_flagged(self) -> None:
+        """The serialized loop payload contains the user task text, so a
+        task id like task-1234567890 must not read as a leaked API key."""
+        decision = check_tool_call(
+            "model",
+            {
+                "url": "https://note3-prev-api.askdiandian.com/v1/chat/completions",
+                "data": json.dumps(
+                    {
+                        "messages": [
+                            {"role": "user", "content": "summarize task-1234567890"}
+                        ]
+                    }
+                ),
+            },
+            audit_path=self.audit_path,
+        )
+        self.assertTrue(decision)
+        self.assertEqual(decision.reason, "policy_passed")
+
+    def test_symlink_to_blocked_file_is_blocked(self) -> None:
+        """A benign-named symlink must not dodge the blocked-target name."""
+        root = Path(self.temp_directory.name) / "root"
+        root.mkdir()
+        (root / ".env").write_text("PASSWORD=demo-only\n", encoding="utf-8")
+        link = root / "notes.txt"
+        try:
+            link.symlink_to(root / ".env")
+        except OSError:
+            self.skipTest("symlinks unavailable on this platform")
+        policy_path = Path(self.temp_directory.name) / "policy.yaml"
+        policy_path.write_text(
+            "blocked_files:\n"
+            "  - .env\n"
+            "allowed_file_roots:\n"
+            "  - root\n"
+            "allowed_domains:\n"
+            "  - example.com\n",
+            encoding="utf-8",
+        )
+
+        decision = check_tool_call(
+            "file",
+            {"path": str(link)},
+            policy_path=policy_path,
+            audit_path=self.audit_path,
+        )
+        self.assertFalse(decision)
+        self.assertEqual(decision.reason, "blocked_file")
+
+    def test_policy_cache_reloads_when_file_changes(self) -> None:
+        policy_path = Path(self.temp_directory.name) / "policy.yaml"
+        policy_path.write_text(
+            "blocked_files:\n"
+            "  - .env\n"
+            "allowed_file_roots:\n"
+            "  - test\n"
+            "allowed_domains:\n"
+            "  - example.com\n",
+            encoding="utf-8",
+        )
+        first = load_policy(policy_path)
+        self.assertTrue(first.allows_url("https://example.com"))
+        self.assertFalse(first.allows_url("https://github.com"))
+
+        # The extra line changes the (mtime_ns, size) fingerprint.
+        policy_path.write_text(
+            "blocked_files:\n"
+            "  - .env\n"
+            "allowed_file_roots:\n"
+            "  - test\n"
+            "allowed_domains:\n"
+            "  - example.com\n"
+            "  - github.com\n",
+            encoding="utf-8",
+        )
+        second = load_policy(policy_path)
+        self.assertTrue(second.allows_url("https://github.com"))
+
+    def test_policy_cache_returns_same_object_while_unchanged(self) -> None:
+        self.assertIs(load_policy(DEFAULT_POLICY_PATH), load_policy(DEFAULT_POLICY_PATH))
+
     def test_policy_blocks_files_and_restricts_domains(self) -> None:
         policy = load_policy(DEFAULT_POLICY_PATH)
         self.assertTrue(policy.blocks_file("test/secrets/.env"))