非対話実行でターン境界を守る — コマンド分類テーブルと fail-closed ゲートの実装メモ

#!/usr/bin/env python3
"""シェルスクリプト中のコマンド起動を抽出し、静的スキャナが
「文字列リテラルとして読めるか」を分類する。"""
import re, sys, os, json
 
CMD_TOKEN = re.compile(r'(?:^|[;|&]{1,2}\s*|\$\(\s*|`\s*)\s*([A-Za-z_][\w./-]*)\s')
VAR_REF   = re.compile(r'\$\{?[A-Za-z_][\w]*\}?')
HEREDOC   = re.compile(r'<<-?\s*[\'"]?([A-Za-z_][\w]*)')
 
def classify_line(line):
    """literal / dynamic / None"""
    s = line.strip()
    if not s or s.startswith('#'):
        return None
    m = CMD_TOKEN.search(' ' + s)
    if not m:
        return None
    head = m.group(1)
    parts = s.split()
    if parts and VAR_REF.search(parts[0]):
        return ('dynamic', head)          # コマンド名自体が変数展開
    rest = s[m.end():]
    if VAR_REF.search(rest) or '$(' in rest or '`' in rest:
        return ('dynamic', head)          # 引数が実行時に決まる
    return ('literal', head)
 
def scan(path):
    stats = {'literal': 0, 'dynamic': 0, 'heredoc_blocks': 0, 'lines': 0}
    heads = {}
    try:
        text = open(path, encoding='utf-8', errors='replace').read()
    except OSError:
        return stats, heads
    stats['heredoc_blocks'] = len(HEREDOC.findall(text))
    for line in text.splitlines():
        stats['lines'] += 1
        r = classify_line(line)
        if not r:
            continue
        kind, head = r
        stats[kind] += 1
        heads.setdefault(head, {'literal': 0, 'dynamic': 0})[kind] += 1
    return stats, heads
 
def main(roots):
    total = {'literal': 0, 'dynamic': 0, 'heredoc_blocks': 0, 'lines': 0}
    allheads, files = {}, 0
    for root in roots:
        for dirpath, dirnames, filenames in os.walk(root):
            dirnames[:] = [d for d in dirnames
                           if d not in ('.git', 'node_modules', '.next')]
            for fn in filenames:
                if not fn.endswith(('.sh', '.bash', '.md', '.txt')):
                    continue
                p = os.path.join(dirpath, fn)
                try:
                    if os.path.getsize(p) > 400_000:
                        continue
                except OSError:
                    continue
                s, h = scan(p)
                if s['literal'] + s['dynamic'] == 0:
                    continue
                files += 1
                for k in total:
                    total[k] += s[k]
                for head, c in h.items():
                    a = allheads.setdefault(head, {'literal': 0, 'dynamic': 0})
                    a['literal'] += c['literal']
                    a['dynamic'] += c['dynamic']
    print(json.dumps({'files': files, 'total': total}, ensure_ascii=False))
    tops = sorted(allheads.items(),
                  key=lambda kv: -(kv[1]['literal'] + kv[1]['dynamic']))[:12]
    for head, c in tops:
        n = c['literal'] + c['dynamic']
        print(f"{head:14s} n={n:5d} dynamic={c['dynamic'] / n * 100:5.1f}%")
 
if __name__ == '__main__':
    main(sys.argv[1:])

#!/usr/bin/env python3
"""素朴な完全一致テーブルと、正規化つき分類器の取りこぼしを比較する。"""
import time, statistics
from turn_gate import classify, READONLY, TURN, INTERACTIVE_ONLY
 
CASES = [
    ("/usage", "readonly"), ("usage", "readonly"), (" /usage ", "readonly"),
    ("/usage --output-format json", "readonly"), ("/Usage", "readonly"),
    ("/mcp:list", "readonly"), ("/mcp list", "readonly"), ("/status", "readonly"),
    ("/cost", "readonly"), ("/context", "readonly"), ("/config get model", "readonly"),
    ("/model", "readonly"), ("/help", "readonly"),
    ("/init", "turn"), ("init", "turn"), ("/review --base main", "turn"),
    ("/compact", "turn"), ("/agent run x", "turn"), ("/fix", "turn"),
    ("/settings", "interactive"), ("/login", "interactive"),
    ("/resume", "interactive"), ("/clear", "interactive"),
    ("/telemetry", "unclassified"), ("/doctor", "unclassified"),
    ("", "unclassified"), ("   ", "unclassified"),
    ('/usage "unclosed', "unclassified"), ("--output-format json", "unclassified"),
]
 
def naive(raw):
    """正規化なしの完全一致テーブル"""
    if raw in {"/" + c for c in READONLY}:
        return "readonly"
    if raw in {"/" + c for c in TURN}:
        return "turn"
    if raw in {"/" + c for c in INTERACTIVE_ONLY}:
        return "interactive"
    return "unclassified"
 
def score(fn):
    ok = sum(1 for raw, exp in CASES if fn(raw) == exp)
    # 危険な誤り: ターン消費・対話専用を readonly と誤認する
    unsafe = sum(1 for raw, exp in CASES
                 if exp in ("turn", "interactive") and fn(raw) == "readonly")
    # 過剰拒否: readonly を unclassified と誤認する（安全側だが運用は止まる）
    overblock = sum(1 for raw, exp in CASES
                    if exp == "readonly" and fn(raw) == "unclassified")
    return ok, unsafe, overblock
 
N = len(CASES)
for label, fn in (("完全一致", naive), ("正規化", lambda r: classify(r)[0])):
    ok, unsafe, over = score(fn)
    print(f"{label:6s}: 正解 {ok}/{N} ({ok / N * 100:.1f}%)  "
          f"危険な誤分類 {unsafe}  過剰拒否 {over}")
 
t = []
for _ in range(5):
    s = time.perf_counter()
    for _ in range(20000):
        for raw, _e in CASES:
            classify(raw)
    t.append((time.perf_counter() - s) / (20000 * N) * 1e6)
print(f"1件あたりの分類コスト: {statistics.median(t):.3f} us (中央値, 5回)")