[verified] Check trusted-template AC verification classification

2026-10-12 16:28:55 +00:00
parent b687af03ec
commit c1f0034300
3 changed files with 253 additions and 1 deletions
@@ -642,6 +642,119 @@ def verify_publication(before, after, exact_review, result, allowed_paths, revie
    return {"review_id": rid, "state": expected, "review_records_added": 1,
            "timeline_events_added": 1, "projected_review_ids": projected, "read_only": True}

+def _classification_bytes(path):
+    """Bounded regular-file read; caller independently admits source provenance."""
+    fd = None
+    try:
+        fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+        if not stat.S_ISREG(os.fstat(fd).st_mode):
+            _fail("classification input unavailable")
+        with os.fdopen(fd, "rb") as stream:
+            fd = None
+            raw = stream.read(1048577)
+        if len(raw) > 1048576:
+            _fail("classification input too large")
+        raw.decode("utf-8")
+        return raw
+    except (OSError, UnicodeError, ValueError):
+        _fail("classification input unavailable")
+    finally:
+        if fd is not None:
+            os.close(fd)
+
+
+def _classification_lines(text):
+    if type(text) is not str or len(text.encode("utf-8")) > 1048576:
+        _fail("invalid classification text")
+    text = text.replace("\r\n", "\n")
+    if any(c in text for c in ("\r", "\x00", "\u2028", "\u2029")):
+        _fail("unsupported classification text")
+    visible, fence = [], None
+    for line in text.split("\n"):
+        match = re.fullmatch(r" {0,3}(`{3,}|~{3,})(.*)", line)
+        if fence is not None:
+            if match and match[1][0] == fence[0] and len(match[1]) >= fence[1] and not match[2].strip():
+                fence = None
+            continue
+        if match:
+            if match[1][0] == "`" and "`" in match[2]:
+                _fail("ambiguous classification fence")
+            fence = (match[1][0], len(match[1]))
+            continue
+        if '<!--' in line or '-->' in line or re.search(r'</?[A-Za-z][^>]*>', line):
+            _fail("unsupported classification HTML")
+        visible.append(line)
+    if fence is not None:
+        _fail("unterminated classification fence")
+    return visible
+
+
+def check_ac_classification(body, expected_template_sha256):
+    """Check only canonical section-7 AC classification, not full compliance."""
+    if type(expected_template_sha256) is not str or not HEX64.fullmatch(expected_template_sha256):
+        _fail("invalid trusted template digest")
+    template = _classification_bytes(Path(__file__).resolve().parents[5] / "docs/requirements/TEMPLATE.md")
+    if hashlib.sha256(template).hexdigest() != expected_template_sha256:
+        _fail("trusted template drift")
+    template_text = template.decode("utf-8")
+    try:
+        template_block = template_text.split("## 7. 인수 기준\n", 1)[1].split("## 8.", 1)[0]
+        classification = template_block.split("**검증 분류**\n", 1)[1].split("**상태 불변조건**", 1)[0]
+    except IndexError:
+        _fail("unsupported trusted template")
+    fields = re.findall(r"^- ([^:]+): \$\{[a-z_]+\}$", classification, flags=re.M)
+    if fields != ["수준", "백엔드", "케이스"]:
+        _fail("unsupported trusted template")
+    lines = _classification_lines(body)
+    section = "## 7. 인수 기준"
+    if lines.count(section) != 1:
+        _fail("ambiguous AC section")
+    start = lines.index(section) + 1
+    end = next((i for i in range(start, len(lines)) if lines[i].startswith("## ")), len(lines))
+    if end == len(lines) or lines[end] != "## 8. 오류 및 충돌 계약":
+        _fail("unsupported AC section boundary")
+    blocks, current = [], None
+    for line in lines[start:end]:
+        if line.startswith("### "):
+            match = re.fullmatch(r"### (REQ-[0-9]{3}\.[0-9]{2}-AC[0-9]{2}) — (\S.*)", line)
+            if not match:
+                _fail("unsupported AC heading")
+            current = (match[1], [])
+            blocks.append(current)
+        elif current is None:
+            if line.strip():
+                _fail("unsupported AC preamble")
+        else:
+            if line.lstrip().startswith("#"):
+                _fail("ambiguous AC heading")
+            current[1].append(line)
+    ids = [item[0] for item in blocks]
+    if not ids or len(ids) != len(set(ids)):
+        _fail("missing or duplicate AC")
+    for _, block in blocks:
+        marker = "**검증 분류**"
+        if block.count(marker) != 1:
+            _fail("missing or duplicate AC classification")
+        pos = block.index(marker) + 1
+        stop = next((i for i in range(pos, len(block)) if block[i].startswith("**")), len(block))
+        values = {}
+        for line in block[pos:stop]:
+            if not line.strip():
+                continue
+            match = re.fullmatch(r"- ([^:]+): (.*)", line)
+            if not match or match[1] not in fields or match[1] in values:
+                _fail("ambiguous AC classification field")
+            value = match[2].strip()
+            if not value or "${" in value:
+                _fail("empty or placeholder AC classification")
+            values[match[1]] = value
+        if set(values) != set(fields):
+            _fail("missing AC classification field")
+    return {"ac_ids": ids, "checked_fields": fields, "read_only": True,
+            "template_sha256": expected_template_sha256,
+            "document_sha256": hashlib.sha256(body.encode("utf-8")).hexdigest()}
+
+
 def _read(path):
    try:
        with open(path, encoding="utf-8", newline="") as stream:
@@ -670,6 +783,10 @@ def main(argv=None):
    q=sub.add_parser("check-cache"); q.add_argument("--contract-path",required=True); q.add_argument("--owner",required=True); q.add_argument("--repository",required=True); q.add_argument("--issue",required=True,type=int)
    q=sub.add_parser("check-pages"); q.add_argument("--pages",required=True); q.add_argument("--key",required=True)
    q.add_argument("--page-size", type=int, default=50); q.add_argument("--max-pages", type=int, default=10000)
+    q=sub.add_parser("check-ac-classification")
+    q.add_argument("--document", required=True)
+    q.add_argument("--expected-document-sha256", required=True)
+    q.add_argument("--expected-template-sha256", required=True)
    q=sub.add_parser("check-unchanged"); q.add_argument("--before",required=True); q.add_argument("--after",required=True)
    q=sub.add_parser("verify-publication");
    for n in ("before","after","exact-review","result","allowed-paths","inline-comments"): q.add_argument("--"+n,required=True)
@@ -703,6 +820,11 @@ def main(argv=None):
            items = collect_pages(f, a.key, a.page_size, a.max_pages)
            if len(used) != len(pages): _fail("extra saved pages")
            print(json.dumps(items,ensure_ascii=False,sort_keys=True))
+        elif a.cmd=="check-ac-classification":
+            raw = _classification_bytes(a.document)
+            if type(a.expected_document_sha256) is not str or not HEX64.fullmatch(a.expected_document_sha256) or hashlib.sha256(raw).hexdigest() != a.expected_document_sha256:
+                _fail("classification document drift")
+            print(json.dumps(check_ac_classification(raw.decode("utf-8"), a.expected_template_sha256), ensure_ascii=False, sort_keys=True))
        elif a.cmd=="check-unchanged": print(json.dumps({"unchanged":verify_unchanged(_json_file(a.before),_json_file(a.after))}))
        else:
            print(json.dumps(verify_publication(_json_file(a.before),_json_file(a.after),_json_file(a.exact_review),_json_file(a.result),_json_file(a.allowed_paths),a.reviewer_id,_json_file(a.inline_comments),_json_file(a.provider_contract) if a.provider_contract else None, expected_pr_url=a.expected_pr_url),sort_keys=True))