セキュリティ特化モデルを個人開発の監査に組み込む — 抽出層と、再提出されない指摘の指紋設計

# scan_sinks.py — 外向きリクエストのシンクを抽出して指紋を付ける
import ast, hashlib, json, sys
from pathlib import Path
 
SINKS = {
    ("requests", "get"), ("requests", "post"),
    ("httpx", "get"), ("httpx", "post"),
    ("urllib.request", "urlopen"),
}
TAINT_ROOTS = ("request",)
 
 
def dotted(node):
    if isinstance(node, ast.Attribute):
        left = dotted(node.value)
        return f"{left}.{node.attr}" if left else node.attr
    if isinstance(node, ast.Name):
        return node.id
    return ""
 
 
def taint_key(node):
    """request.args.get("src") / request.json["url"] を正規化キーに変換する。"""
    if isinstance(node, ast.Call):
        base = dotted(node.func)
        if base.startswith(TAINT_ROOTS) and node.args and isinstance(node.args[0], ast.Constant):
            return f"{base}({node.args[0].value})"
        return None
    if isinstance(node, ast.Subscript):
        base = dotted(node.value)
        if base.startswith(TAINT_ROOTS) and isinstance(node.slice, ast.Constant):
            return f"{base}[{node.slice.value}]"
        return None
    if isinstance(node, ast.Attribute):
        base = dotted(node)
        return base if base.startswith(TAINT_ROOTS) else None
    return None
 
 
class FnScanner(ast.NodeVisitor):
    def __init__(self, relpath):
        self.relpath = relpath
        self.findings = []
 
    def visit_FunctionDef(self, fn):
        env = {}  # ローカル変数名 -> 汚染元キー
        for node in ast.walk(fn):
            if isinstance(node, ast.Assign):
                key = taint_key(node.value)
                if key:
                    for t in node.targets:
                        if isinstance(t, ast.Name):
                            env[t.id] = key
        for node in ast.walk(fn):
            if not isinstance(node, ast.Call):
                continue
            fq = dotted(node.func)
            if "." not in fq or not node.args:
                continue
            mod, _, fname = fq.rpartition(".")
            if (mod, fname) not in SINKS:
                continue
            arg = node.args[0]
            local = arg.id if isinstance(arg, ast.Name) else ""
            src = env.get(local) if local else taint_key(arg)
            if src is None:
                continue
            self.findings.append({
                "file": self.relpath, "line": node.lineno,
                "sink": f"{mod}.{fname}", "source": src,
                "local": local, "fn": fn.name,
            })
        self.generic_visit(fn)
 
 
def h(*parts):
    return hashlib.sha1("|".join(parts).encode()).hexdigest()[:12]
 
 
def scan(root: Path):
    out = []
    for p in sorted(root.rglob("*.py")):
        rel = str(p.relative_to(root))
        s = FnScanner(rel)
        s.visit(ast.parse(p.read_text()))
        for f in s.findings:
            f["fp_line"] = h(f["file"], str(f["line"]), f["sink"])          # 方式A
            f["fp_v1"] = h(f["file"], f["fn"], f["sink"], f["local"])       # 方式B
            f["fp_v2"] = h(f["sink"], f["source"], f["fn"])                 # 方式C
            f["fp_v3"] = h(f["sink"], f["source"])                          # 方式D
            out.append(f)
    return out
 
 
if __name__ == "__main__":
    print(json.dumps(scan(Path(sys.argv[1])), ensure_ascii=False, indent=2))

衝突: 97d2524cdca2 -> ['services/thumbnail.py:8  fetch_remote_thumbnail',
                       'services/thumbnail.py:16 fetch_allowlisted_thumbnail']

# verify_gate.py — 再現テストの落ち→通りだけで採用を決める
import subprocess, tempfile, pathlib
 
def run(cmd, cwd):
    return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True).returncode
 
def verify(repo: pathlib.Path, repro_test: str, patch: str) -> dict:
    """戻り値の accepted が True のときだけ台帳へ書き込む。"""
    test_path = repo / "tests" / "test_repro_generated.py"
    test_path.parent.mkdir(exist_ok=True)
    test_path.write_text(repro_test)
 
    before = run(["python", "-m", "pytest", "-q", str(test_path)], repo)
    if before == 0:
        # 修正前に通ってしまう再現テストは、再現できていない
        test_path.unlink()
        return {"accepted": False, "reason": "repro_did_not_fail"}
 
    with tempfile.NamedTemporaryFile("w", suffix=".patch", delete=False) as fp:
        fp.write(patch)
        patch_file = fp.name
 
    if run(["git", "apply", "--check", patch_file], repo) != 0:
        test_path.unlink()
        return {"accepted": False, "reason": "patch_does_not_apply"}
 
    run(["git", "apply", patch_file], repo)
    after = run(["python", "-m", "pytest", "-q", str(test_path)], repo)
    suite = run(["python", "-m", "pytest", "-q"], repo)
    run(["git", "checkout", "--", "."], repo)   # 必ず元に戻す
    test_path.unlink(missing_ok=True)
 
    return {
        "accepted": after == 0 and suite == 0,
        "reason": "ok" if after == 0 and suite == 0 else f"after={after} suite={suite}",
    }