feat: implement frozen strict typed identity contract

scripts/canonical_typed_identity.py

@@ -0,0 +1,173 @@
+"""Pure structured_tokens.v2 validation, canonicalization, and digesting."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import unicodedata
+from typing import Any
+
+SCHEMA = "contract_identity.v2"
+VERSION = "structured_tokens.v2"
+MAX_RECORD_BYTES = 65536
+PAYLOAD_FIELDS = ("schema", "canonicalizer_version", "scope", "anchors", "predicates", "required_behavior", "forbidden_behavior", "ordering_constraints")
+RECORD_FIELDS = set(PAYLOAD_FIELDS) | {"contract_key", "behavior_digest", "fingerprint_v2"}
+IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*(?:\(\))?$")
+OWNER_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,37}[A-Za-z0-9])?$")
+NAME_RE = re.compile(r"^[A-Za-z0-9._-]{1,100}$")
+PREDICATE_OPERATORS = {">=", "<=", ">", "<", "==", "!=", "===", "!=="}
+ORDERING_OPERATORS = {"->", "=>"}
+POLICY_STATES = {"required", "forbidden", "present", "absent", "enabled", "disabled", "valid", "invalid", "missing", "optional"}
+SECRET_TYPES = {"credential", "authorization", "api_key", "session_token", "private_key"}
+SECRET_ROLES = {"auth", "header", "query", "environment", "body"}
+CATEGORIES = {"bug", "contract", "logic", "performance", "reliability", "security"}
+
+
+class IdentityError(ValueError):
+    """The typed identity is not valid structured_tokens.v2 data."""
+
+
+def _fail(message: str) -> None:
+    raise IdentityError(message)
+
+
+def _object(value: Any, fields: set[str], label: str) -> dict[str, Any]:
+    if type(value) is not dict or set(value) != fields:
+        _fail(f"{label} fields are invalid")
+    return value
+
+
+def _text(value: Any, chars: int, size: int, label: str) -> str:
+    if type(value) is not str:
+        _fail(f"{label} must be a string")
+    try:
+        raw = value.encode("utf-8")
+    except UnicodeEncodeError:
+        _fail(f"{label} is not valid UTF-8")
+    if len(value) > chars or len(raw) > size:
+        _fail(f"{label} exceeds its limit")
+    normalized = unicodedata.normalize("NFC", value)
+    if any(unicodedata.category(char) in {"Cc", "Cf"} for char in normalized):
+        _fail(f"{label} contains a control character")
+    if len(normalized) > chars or len(normalized.encode("utf-8")) > size:
+        _fail(f"{label} exceeds its normalized limit")
+    return normalized
+
+
+def _scope(value: Any) -> dict[str, str]:
+    value = _object(value, {"repo", "file", "category"}, "scope")
+    repo = _text(value["repo"], 140, 140, "scope.repo")
+    if repo.count("/") != 1:
+        _fail("scope.repo must be owner/name")
+    owner, name = repo.split("/")
+    if not OWNER_RE.fullmatch(owner) or "--" in owner or not NAME_RE.fullmatch(name) or name in {".", ".."}:
+        _fail("scope.repo is invalid")
+    path = _text(value["file"], 1024, 1024, "scope.file")
+    if path.startswith("/") or "\\" in path or any(part in {"", ".", ".."} for part in path.split("/")):
+        _fail("scope.file is invalid")
+    category = _text(value["category"], 32, 32, "scope.category")
+    if category not in CATEGORIES:
+        _fail("scope.category is invalid")
+    return {"repo": f"{owner.lower()}/{name.lower()}", "file": path, "category": category}
+
+
+def _token(value: Any, expected: str, operators: set[str] | None = None) -> dict[str, Any]:
+    value = _object(value, {"kind", "value"}, "token")
+    if value["kind"] != expected:
+        _fail(f"expected {expected} token")
+    item = value["value"]
+    if expected == "identifier":
+        item = _text(item, 512, 512, "identifier")
+        if not IDENTIFIER_RE.fullmatch(item):
+            _fail("identifier grammar is invalid")
+    elif expected == "operator":
+        if type(item) is not str or item not in (operators or set()):
+            _fail("operator is invalid here")
+    elif expected == "policy_state":
+        if type(item) is not str or item not in POLICY_STATES:
+            _fail("policy_state is invalid")
+    elif expected == "secret_ref":
+        item = _object(item, {"type", "role", "position"}, "secret_ref")
+        if type(item["type"]) is not str or item["type"] not in SECRET_TYPES or type(item["role"]) is not str or item["role"] not in SECRET_ROLES:
+            _fail("secret_ref metadata is invalid")
+        if type(item["position"]) is not int or not 0 <= item["position"] <= 1024:
+            _fail("secret_ref position is invalid")
+        item = dict(item)
+    return {"kind": expected, "value": item}
+
+
+def _items(value: Any, minimum: int, label: str) -> list[Any]:
+    if type(value) is not list or not minimum <= len(value) <= 32:
+        _fail(f"{label} item count is invalid")
+    return value
+
+
+def _anchors(value: Any) -> list[dict[str, Any]]:
+    result = []
+    for index, item in enumerate(_items(value, 1, "anchors")):
+        result.append(_token(item, "identifier") if index % 2 == 0 else _token(item, "operator", {"::"}))
+    if len(result) % 2 == 0:
+        _fail("anchors must end with an identifier")
+    return result
+
+
+def _clauses(value: Any, label: str, kind: str, minimum: int) -> list[list[dict[str, Any]]]:
+    result = []
+    for clause in _items(value, minimum, label):
+        if type(clause) is not list:
+            _fail(f"{label} clause must be an array")
+        if kind == "predicate" and len(clause) == 3:
+            parsed = [_token(clause[0], "identifier"), _token(clause[1], "operator", PREDICATE_OPERATORS), _token(clause[2], "identifier")]
+        elif kind == "ordering" and len(clause) == 3:
+            parsed = [_token(clause[0], "identifier"), _token(clause[1], "operator", ORDERING_OPERATORS), _token(clause[2], "identifier")]
+        elif kind == "behavior" and len(clause) == 1:
+            parsed = [_token(clause[0], "policy_state")]
+        elif kind == "behavior" and len(clause) == 3:
+            parsed = [_token(clause[0], "identifier"), _token(clause[1], "operator", {"==", "!="})]
+            if type(clause[2]) is not dict or type(clause[2].get("kind")) is not str or clause[2].get("kind") not in {"policy_state", "secret_ref"}:
+                _fail(f"{label} final operand is invalid")
+            parsed.append(_token(clause[2], clause[2]["kind"]))
+        else:
+            _fail(f"{label} clause grammar is invalid")
+        result.append(parsed)
+    return result
+
+
+def _json(value: Any) -> bytes:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+
+def _digest(value: Any) -> str:
+    return hashlib.sha256(_json(value)).hexdigest()
+
+
+def build_identity_record(payload: Any) -> dict[str, Any]:
+    payload = _object(payload, set(PAYLOAD_FIELDS), "payload")
+    if payload["schema"] != SCHEMA or payload["canonicalizer_version"] != VERSION:
+        _fail("schema or canonicalizer version is unsupported")
+    canonical = {
+        "schema": SCHEMA, "canonicalizer_version": VERSION, "scope": _scope(payload["scope"]),
+        "anchors": _anchors(payload["anchors"]),
+        "predicates": _clauses(payload["predicates"], "predicates", "predicate", 1),
+        "required_behavior": _clauses(payload["required_behavior"], "required_behavior", "behavior", 1),
+        "forbidden_behavior": _clauses(payload["forbidden_behavior"], "forbidden_behavior", "behavior", 0),
+        "ordering_constraints": _clauses(payload["ordering_constraints"], "ordering_constraints", "ordering", 0),
+    }
+    contract = {key: canonical[key] for key in PAYLOAD_FIELDS[:5]}
+    record = dict(canonical)
+    record["contract_key"] = _digest(contract)
+    behavior = {"contract_key": record["contract_key"], **{key: canonical[key] for key in PAYLOAD_FIELDS[5:]}}
+    record["behavior_digest"] = _digest(behavior)
+    record["fingerprint_v2"] = _digest({"contract_key": record["contract_key"], "behavior_digest": record["behavior_digest"]})
+    if len(_json(record)) > MAX_RECORD_BYTES:
+        _fail("canonical record exceeds 65536 bytes")
+    return record
+
+
+def verify_identity_record(record: Any) -> dict[str, Any]:
+    record = _object(record, RECORD_FIELDS, "record")
+    expected = build_identity_record({key: record[key] for key in PAYLOAD_FIELDS})
+    if record != expected:
+        _fail("record is not the exact canonical record")
+    return expected

tests/test_canonical_typed_identity_r1d.py

@@ -0,0 +1,127 @@
+import copy
+import hashlib
+import json
+import unittest
+
+from scripts.canonical_typed_identity import IdentityError, build_identity_record, verify_identity_record
+
+
+def token(kind, value=None):
+    values = {"identifier": "subject", "policy_state": "required", "secret_ref": {"type": "credential", "role": "auth", "position": 0}}
+    return {"kind": kind, "value": values.get(kind) if value is None else value}
+
+
+def clause(parts):
+    return [token(part) if part in {"identifier", "policy_state", "secret_ref"} else token("operator", part) for part in parts]
+
+
+def payload():
+    return {
+        "schema": "contract_identity.v2",
+        "canonicalizer_version": "structured_tokens.v2",
+        "scope": {"repo": "QuantStrategyLab/AIAuditBridge", "file": "service/Auth.py", "category": "contract"},
+        "anchors": [token("identifier", "Auth")],
+        "predicates": [clause(["identifier", ">=", "identifier"])],
+        "required_behavior": [clause(["policy_state"])],
+        "forbidden_behavior": [],
+        "ordering_constraints": [],
+    }
+
+
+def modified(path, value):
+    result = payload()
+    target = result
+    for key in path[:-1]:
+        target = target[key]
+    target[path[-1]] = value
+    return result
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+
+CORPUS = [
+    ("near_miss_ghs_database", "anchor", {"kind": "identifier", "value": "ghs_database"}, True),
+    ("near_miss_github_pat_validator", "anchor", {"kind": "identifier", "value": "github_pat_validator"}, True),
+    ("near_miss_ordinary_identifiers", "anchors", ["Eurasia", "keyJson", "secret_manager", "secret_ref_validator"], True),
+    ("valid_predicate", "predicates", ["identifier", ">=", "identifier"], True),
+    ("valid_policy_behavior", "required_behavior", ["policy_state"], True),
+    ("valid_secret_final_behavior", "required_behavior", ["identifier", "==", "secret_ref"], True),
+    ("valid_ordering", "ordering_constraints", ["identifier", "->", "identifier"], True),
+    ("reject_predicate_policy_state", "predicates", ["policy_state", "==", "identifier"], False),
+    ("reject_predicate_namespace", "predicates", ["identifier", "::", "identifier"], False),
+    ("reject_trailing_operator", "predicates", ["identifier", ">="], False),
+    ("reject_secret_lhs", "required_behavior", ["secret_ref", "==", "identifier"], False),
+    ("reject_secret_middle", "required_behavior", ["identifier", "==", "secret_ref", "==", "identifier"], False),
+    ("reject_ordering_comparison", "ordering_constraints", ["identifier", "==", "identifier"], False),
+    ("reject_unknown_field", "top", {"evidence": {}}, False),
+    ("reject_prose", "anchor", {"kind": "identifier", "value": "raw prose"}, False),
+    ("reject_assignment", "anchor", {"kind": "identifier", "value": "password=secret"}, False),
+    ("reject_control", "anchor", {"kind": "identifier", "value": "bad\\u0000name"}, False),
+    ("reject_v1", "top", {"canonicalizer_version": "structured_tokens.v1"}, False),
+]
+
+
+class CanonicalTypedIdentityR1dTests(unittest.TestCase):
+    def test_corpus_and_oracle(self):
+        for case_id, target, value, accepted in CORPUS:
+            with self.subTest(case_id=case_id):
+                candidate = payload()
+                if target == "anchor":
+                    candidate["anchors"] = [value]
+                elif target == "anchors":
+                    for identifier in value:
+                        candidate["anchors"] = [token("identifier", identifier)]
+                        build_identity_record(candidate)
+                    continue
+                elif target == "top":
+                    candidate.update(value)
+                else:
+                    candidate[target] = [clause(value)]
+                if accepted:
+                    build_identity_record(candidate)
+                else:
+                    with self.assertRaises(IdentityError):
+                        build_identity_record(candidate)
+
+    def test_canonicalization_digests_and_exact_verification(self):
+        source = payload()
+        source["scope"]["file"] = "Cafe\u0301.py"
+        record = build_identity_record(source)
+        self.assertEqual(record["scope"]["repo"], "quantstrategylab/aiauditbridge")
+        self.assertEqual(record["scope"]["file"], "Caf\u00e9.py")
+        contract = {key: record[key] for key in ("schema", "canonicalizer_version", "scope", "anchors", "predicates")}
+        self.assertEqual(record["contract_key"], digest(contract))
+        behavior = {"contract_key": record["contract_key"], **{key: record[key] for key in ("required_behavior", "forbidden_behavior", "ordering_constraints")}}
+        self.assertEqual(record["behavior_digest"], digest(behavior))
+        self.assertEqual(record["fingerprint_v2"], digest({"contract_key": record["contract_key"], "behavior_digest": record["behavior_digest"]}))
+        self.assertEqual(verify_identity_record(record), record)
+        for key in ("contract_key", "behavior_digest", "fingerprint_v2"):
+            tampered = copy.deepcopy(record)
+            tampered[key] = "0" * 64
+            with self.subTest(key=key), self.assertRaises(IdentityError):
+                verify_identity_record(tampered)
+
+    def test_strict_schema_unicode_bounds_and_secret_metadata(self):
+        invalid = [
+            *[modified(("scope", "repo"), value) for value in ("owner", "-owner/name", "owner--x/name", "owner/..")],
+            *[modified(("scope", "file"), value) for value in ("/absolute.py", "a\\b.py", "a/../b.py", "a//b.py")],
+            modified(("anchors",), [token("identifier", "bad\x00name")]),
+            modified(("anchors",), [token("identifier", "a" * 513)]),
+            modified(("required_behavior",), [[token("identifier"), token("operator", "=="), {"kind": "secret_ref", "value": {"type": "credential", "role": "auth", "position": 0, "raw": "forbidden"}}]]),
+            modified(("required_behavior",), [[token("identifier"), token("operator", "=="), token("secret_ref", {"type": "credential", "role": "auth", "position": True})]]),
+            modified(("anchors",), [token("identifier", "a"), token("identifier", "b")]),
+        ]
+        for candidate in invalid:
+            with self.subTest(candidate=candidate), self.assertRaises(IdentityError):
+                build_identity_record(candidate)
+
+        canonical = modified(("anchors",), [token("identifier", "Cafe\u0301")])
+        canonical["scope"]["repo"] = "Owner/Name"
+        with self.assertRaises(IdentityError):
+            build_identity_record(canonical)
+
+
+if __name__ == "__main__":
+    unittest.main()