feat(safety): wire remaining rules, fix import, add detection eval

- Wire tool-fs-system-dir-write (open(...,'w') into system dirs) and
  tool-secret-private-key (embedded PEM literal anywhere in source) in the
  python scanner; these rule ids were previously defined but never fired
- Add `from typing import Optional` to _shell_parse (it was referenced only
  inside a non-evaluated local annotation, but keep it correct/explicit)
- Add manifest samples 15/16 covering the two newly wired rules
- Add scripts/tool_safety_eval.py: detection / false-positive rate eval for
  issue #90 acceptance #2 (>=90% detection, <=10% FP) -> 100% / 0% on manifest

All 73 safety tests pass.

Author: wzxsdf

scripts/tool_safety_eval.py

@@ -0,0 +1,82 @@
+# Copyright (C) 2026 Tencent. All rights reserved.
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Detection-rate / false-positive-rate eval against the sample manifest.
+
+Satisfies issue #90 acceptance #2 (high-risk detection >= 90%, safe-sample
+false-positive <= 10%) with a runnable, reproducible measurement.
+
+A "positive" sample is one whose manifest expected_decision != ALLOW (it should
+be flagged as risky); "detected" means the scan also returns != ALLOW. A
+"negative" sample expects ALLOW; a false positive is one the scan blocks.
+
+Usage:
+    python scripts/tool_safety_eval.py [--manifest m.yaml] [--policy p.yaml]
+"""
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+from typing import NamedTuple
+
+import yaml
+
+from trpc_agent_sdk.tools.safety._policy import load_policy
+from trpc_agent_sdk.tools.safety._scanner import scan
+
+_DEFAULT_MANIFEST = (
+    Path(__file__).resolve().parent.parent
+    / "tests" / "tools" / "safety" / "samples" / "manifest.yaml"
+)
+
+
+class Outcome(NamedTuple):
+    name: str
+    expected: str
+    actual: str
+    correct: bool  # scan agrees with manifest on allow-vs-block
+
+
+def _evaluate(manifest: Path, policy_path: str | None) -> list[Outcome]:
+    data = yaml.safe_load(manifest.read_text(encoding="utf-8"))
+    policy = load_policy(policy_path)
+    out: list[Outcome] = []
+    for s in data["samples"]:
+        actual = scan(policy, s["script"], language=s["language"]).decision.name
+        expected = s["expected_decision"]
+        correct = (actual == "ALLOW") == (expected == "ALLOW")
+        out.append(Outcome(s["name"], expected, actual, correct))
+    return out
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--manifest", default=str(_DEFAULT_MANIFEST))
+    ap.add_argument("--policy", default=None)
+    args = ap.parse_args()
+
+    outcomes = _evaluate(Path(args.manifest), args.policy)
+    positives = [o for o in outcomes if o.expected != "ALLOW"]
+    negatives = [o for o in outcomes if o.expected == "ALLOW"]
+    detected = [o for o in positives if o.actual != "ALLOW"]
+    false_pos = [o for o in negatives if o.actual != "ALLOW"]
+
+    det_rate = len(detected) / len(positives) if positives else 1.0
+    fp_rate = len(false_pos) / len(negatives) if negatives else 0.0
+
+    for o in outcomes:
+        flag = "OK " if o.correct else "ERR"
+        print(f"  [{flag}] {o.name:<24} expected={o.expected:<14} actual={o.actual}")
+    print()
+    print(f"positives={len(positives)} detected={len(detected)} "
+          f"detection_rate={det_rate:.0%}")
+    print(f"negatives={len(negatives)} false_pos={len(false_pos)} "
+          f"false_positive_rate={fp_rate:.0%}")
+
+    ok = det_rate >= 0.90 and fp_rate <= 0.10
+    print("RESULT:", "PASS" if ok else "FAIL",
+          "(acceptance #2: detection>=90%, false_positive<=10%)")
+    return 0 if ok else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/tools/safety/samples/manifest.yaml

@@ -109,4 +109,20 @@ samples:
       from concurrent.futures import ThreadPoolExecutor
       ThreadPoolExecutor(max_workers=10000)
     expected_decision: NEEDS_REVIEW
-    required_rule_ids: [tool-res-concurrent-flood]
+    required_rule_ids: [tool-res-concurrent-flood]
+
+  - name: 15_system_dir_write
+    language: python
+    script: |
+      with open("/etc/passwd", "w") as f:
+          f.write("hacked")
+    expected_decision: DENY
+    required_rule_ids: [tool-fs-system-dir-write]
+
+  - name: 16_private_key
+    language: python
+    script: |
+      key = "-----BEGIN RSA PRIVATE KEY-----\nMIIEoQIBAA\n-----END RSA PRIVATE KEY-----"
+      print(key)
+    expected_decision: DENY
+    required_rule_ids: [tool-secret-private-key]

trpc_agent_sdk/tools/safety/_python_scanner.py

@@ -15,13 +15,15 @@
 from trpc_agent_sdk.tools.safety._rules import R_CODE_UNSAFE_EXEC
 from trpc_agent_sdk.tools.safety._rules import R_FS_RECURSIVE_DELETE
 from trpc_agent_sdk.tools.safety._rules import R_FS_READ_CREDENTIALS
+from trpc_agent_sdk.tools.safety._rules import R_FS_SYSTEM_DIR
 from trpc_agent_sdk.tools.safety._rules import R_NET_HTTP
 from trpc_agent_sdk.tools.safety._rules import R_NET_SOCKET
 from trpc_agent_sdk.tools.safety._rules import R_PROC_SUBPROCESS
 from trpc_agent_sdk.tools.safety._rules import R_RES_CONCURRENT_FLOOD
 from trpc_agent_sdk.tools.safety._rules import R_RES_INFINITE_LOOP
 from trpc_agent_sdk.tools.safety._rules import R_RES_LARGE_WRITE
 from trpc_agent_sdk.tools.safety._rules import R_SECRET_LOGGING
+from trpc_agent_sdk.tools.safety._rules import R_SECRET_PRIVATE_KEY
 from trpc_agent_sdk.tools.safety._types import Decision
 from trpc_agent_sdk.tools.safety._types import Finding
 from trpc_agent_sdk.tools.safety._types import RiskLevel
@@ -30,6 +32,9 @@
 _URL_RE = re.compile(r"https?://([^/\s'\"']+)", re.I)
 _SECRET_NAME_RE = re.compile(r"(api[_-]?key|secret|token|password|passwd|private[_-]?key)", re.I)
 _PRIVATE_KEY_RE = re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----")
+# System directories: writing here can brick the runtime (issue risk class #1).
+_SYSTEM_DIRS = ("/etc/", "/usr/", "/bin/", "/sbin/", "/boot/", "/sys/",
+                "/proc/", "/lib/", "/lib64/", "/var/", "/dev/")
 
 # attribute path -> rule fired when called/used
 _DANGEROUS_ATTR = {
@@ -125,16 +130,17 @@ def resolve_attr(node: ast.AST) -> str:
         # infinite loop
         if isinstance(node, (ast.While,)) and _is_truthy(node.test):
             add(R_RES_INFINITE_LOOP, "while True:", "infinite loop; review.")
-        # secret logging: assignment of secret-named var + later print
+        # secret logging: assignment to a secret-named variable.
         if isinstance(node, ast.Assign):
             for t in node.targets:
                 if isinstance(t, ast.Name) and _SECRET_NAME_RE.search(t.id):
-                    if _PRIVATE_KEY_RE.search(_literal(node.value)):
-                        add(R_SECRET_LOGGING, f"private key in {t.id}",
-                            "private key literal detected.")
-                    else:
-                        add(R_SECRET_LOGGING, f"secret assigned to {t.id}",
-                            "secret-like variable; avoid logging.")
+                    add(R_SECRET_LOGGING, f"secret assigned to {t.id}",
+                        "secret-like variable; avoid logging.")
+        # private-key literal embedded anywhere (independent of variable name).
+        if isinstance(node, ast.Constant) and isinstance(node.value, str) \
+                and _PRIVATE_KEY_RE.search(node.value):
+            add(R_SECRET_PRIVATE_KEY, "private key literal",
+                "embedded private key; refuse.")
 
     return findings
 
@@ -161,6 +167,11 @@ def _check_open_path(call: ast.Call, policy: Policy, add) -> None:
         path = arg.value
     elif isinstance(arg, ast.JoinedStr):
         path = "".join(v.value for v in arg.values if isinstance(v, ast.Constant))
+    mode = _open_mode(call)
+    if path and _is_write_mode(mode) and path.startswith(_SYSTEM_DIRS):
+        add(R_FS_SYSTEM_DIR, f"open('{path}','{mode}')",
+            f"writing to system directory {path}; refuse.")
+        return
     if path and _CRED_PATH_RE.search(path):
         add(R_FS_READ_CREDENTIALS, f"open('{path}')",
             f"reading credential path {path}; review.")
@@ -218,12 +229,24 @@ def _is_whitelisted(url: str, policy: Policy) -> bool:
     return root in {d.lower() for d in policy.whitelisted_domains} or host in policy.whitelisted_domains
 
 
-def _literal(node: ast.AST) -> str:
-    if isinstance(node, ast.Constant) and isinstance(node.value, str):
-        return node.value
+def _open_mode(call: ast.Call) -> str:
+    """Return the literal mode string of an open() call, else ''."""
+    if len(call.args) >= 2:
+        m = call.args[1]
+        if isinstance(m, ast.Constant) and isinstance(m.value, str):
+            return m.value
+    for kw in call.keywords:
+        if kw.arg == "mode" and isinstance(kw.value, ast.Constant) \
+                and isinstance(kw.value.value, str):
+            return kw.value.value
     return ""
 
 
+def _is_write_mode(mode: str) -> bool:
+    """True if the mode can create/modify a file (w x a +)."""
+    return bool(set(mode) & set("wxa+"))
+
+
 def _heuristic_fallback(policy: Policy, script: str) -> list[Finding]:
     """Best-effort when the script is not valid Python AST."""
     findings: list[Finding] = []

trpc_agent_sdk/tools/safety/_shell_parse.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import shlex
+from typing import Optional
 
 
 def split_tokens(cmd: str) -> list[str]: