From 8eeb7dbc4b85e9511d31b2c2fb16a00550c369d0 Mon Sep 17 00:00:00 2001 From: liuhaolinqi <2164800367@qq.com> Date: Sat, 24 Oct 2026 13:10:15 +0800 Subject: [PATCH 1/2] fix: scanner false positives, symlink dodge; perf: policy cache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit findings addressed: - scanner: 'sk-' pattern had no word boundary or minimum length, so 'task-123' / 'risk-2026' matched as leaked keys and blocked clean model calls (the serialized loop payload contains the user task). Now mirrors the dots_client redaction shape; patterns precompiled. - middleware: check the blocked-name list against the resolved path as well, so a benign-named symlink cannot dodge blocked_files. - policy: cache parsed policies per path, revalidated against (st_mtime_ns, st_size) — ~5.5x faster on hot loops, identical change-on-disk semantics, errors never cached. - main: deterministic demo now exits 2 on block, matching --llm. Tests: 30 -> 35 (all pass; symlink case skips only where the platform cannot create symlinks, e.g. Windows without developer mode). --- CHANGELOG.md | 29 ++++++++++++++ main.py | 6 ++- security/middleware.py | 11 +++++- security/policy.py | 28 +++++++++++++ security/scanner.py | 23 ++++++----- test/test_security.py | 90 ++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 173 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a4e859f..b01ae4d 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/main.py b/main.py index f40cac2..75054bd 100644 --- a/main.py +++ b/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) diff --git a/security/middleware.py b/security/middleware.py index 2ebf462..33c2060 100644 --- a/security/middleware.py +++ b/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", diff --git a/security/policy.py b/security/policy.py index d008009..31494fb 100644 --- a/security/policy.py +++ b/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": [], diff --git a/security/scanner.py b/security/scanner.py index 4429a6c..5edbc84 100644 --- a/security/scanner.py +++ b/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) diff --git a/test/test_security.py b/test/test_security.py index 9dae7fb..0510d50 100644 --- a/test/test_security.py +++ b/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 / "real.env").write_text("PASSWORD=demo-only\n", encoding="utf-8") + link = root / "notes.txt" + try: + link.symlink_to(root / "real.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")) From b60bd1385b2862a696f4d2c9b6470bf2c9e154f9 Mon Sep 17 00:00:00 2001 From: liuhaolinqi <2164800367@qq.com> Date: Sat, 24 Oct 2026 13:15:59 +0800 Subject: [PATCH 2/2] fix(test): symlink target must actually carry the blocked name The fixture named the target real.env, but blocks_file matches filenames exactly, so '.env' never matched and the case asserted the wrong expectation. The target is now .env; the benign-named link is notes.txt as before. --- test/test_security.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/test_security.py b/test/test_security.py index 0510d50..a8bb7c5 100644 --- a/test/test_security.py +++ b/test/test_security.py @@ -64,10 +64,10 @@ 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 / "real.env").write_text("PASSWORD=demo-only\n", encoding="utf-8") + (root / ".env").write_text("PASSWORD=demo-only\n", encoding="utf-8") link = root / "notes.txt" try: - link.symlink_to(root / "real.env") + link.symlink_to(root / ".env") except OSError: self.skipTest("symlinks unavailable on this platform") policy_path = Path(self.temp_directory.name) / "policy.yaml"