feat: add strict typed identity model v2

scripts/canonical_typed_identity.py

@@ -0,0 +1,137 @@
+from __future__ import annotations
+import hashlib
+import json
+import re
+import unicodedata
+from dataclasses import dataclass
+from typing import Any
+SCHEMA = "contract_identity.v2"
+VERSION = "structured_tokens.v2"
+OPS = frozenset({">=", "<=", ">", "<", "==", "!=", "===", "!==", "->", "=>", "::"})
+POLICY = frozenset({"required", "forbidden", "present", "absent", "enabled", "disabled", "valid", "invalid", "missing", "optional"})
+KINDS = frozenset({"identifier", "operator", "policy_state", "secret_ref"})
+CATEGORIES = frozenset({"bug", "contract", "logic", "performance", "reliability", "security"})
+SECRET_TYPES = frozenset({"credential", "authorization", "api_key", "session_token", "private_key"})
+SECRET_ROLES = frozenset({"auth", "header", "query", "environment", "body"})
+OWNER = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,37}[A-Za-z0-9])?$")
+REPO = re.compile(r"^[A-Za-z0-9._-]{1,100}$")
+IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*(?:\(\))?$")
+DIGEST = re.compile(r"^[0-9a-f]{64}$")
+MAX_ITEMS = 32
+MAX_BYTES = 64 * 1024
+class IdentityError(ValueError):
+    pass
+@dataclass(frozen=True)
+class CanonicalIdentity:
+    _payload_json: str
+    contract_key: str
+    behavior_digest: str
+    fingerprint_v2: str
+    @property
+    def payload(self) -> dict[str, Any]:
+        return json.loads(self._payload_json)
+    def as_record(self) -> dict[str, Any]:
+        record = self.payload
+        record.update(contract_key=self.contract_key, behavior_digest=self.behavior_digest, fingerprint_v2=self.fingerprint_v2)
+        return record
+def _obj(value: Any, required: set[str], optional: set[str] = set()) -> dict[str, Any]:
+    keys = set(value) if isinstance(value, dict) else set()
+    if not isinstance(value, dict) or len(value) > len(required) + len(optional) or not required <= keys or not keys <= required | optional:
+        raise IdentityError("invalid object fields")
+    return value
+def _text(value: Any, limit: int = 512) -> str:
+    if not isinstance(value, str) or len(value) > limit:
+        raise IdentityError("invalid bounded text")
+    try:
+        raw_size = len(value.encode())
+    except UnicodeEncodeError as exc:
+        raise IdentityError("invalid unicode text") from exc
+    if raw_size > limit:
+        raise IdentityError("invalid bounded text")
+    value = unicodedata.normalize("NFC", value)
+    try:
+        normalized_size = len(value.encode())
+    except UnicodeEncodeError as exc:
+        raise IdentityError("invalid unicode text") from exc
+    if any(unicodedata.category(char).startswith("C") for char in value) or normalized_size > limit:
+        raise IdentityError("invalid text")
+    return value
+def _scope(value: Any) -> dict[str, str]:
+    scope = _obj(value, {"repo", "file", "category"})
+    repo = _text(scope["repo"], 140)
+    parts = repo.split("/")
+    if len(parts) != 2 or not OWNER.fullmatch(parts[0]) or "--" in parts[0] or not REPO.fullmatch(parts[1]) or parts[1] in {".", ".."}:
+        raise IdentityError("invalid repo")
+    path = _text(scope["file"], 1024)
+    if path.startswith("/") or "\\" in path or any(part in {"", ".", ".."} for part in path.split("/")):
+        raise IdentityError("invalid path")
+    category = _text(scope["category"], 32)
+    if category not in CATEGORIES:
+        raise IdentityError("invalid category")
+    return {"repo": f"{parts[0].lower()}/{parts[1].lower()}", "file": path, "category": category}
+def _token(value: Any) -> dict[str, Any]:
+    token = _obj(value, {"kind", "value"})
+    kind = _text(token["kind"], 32)
+    if kind not in KINDS:
+        raise IdentityError("unknown token kind")
+    if kind == "secret_ref":
+        ref = _obj(token["value"], {"type", "role", "position"})
+        typ, role, position = ref["type"], ref["role"], ref["position"]
+        if not isinstance(typ, str) or not isinstance(role, str):
+            raise IdentityError("invalid secret reference")
+        typ, role = _text(typ, 32), _text(role, 32)
+        if typ not in SECRET_TYPES or role not in SECRET_ROLES or isinstance(position, bool) or not isinstance(position, int) or not 0 <= position <= 1024:
+            raise IdentityError("invalid secret reference")
+        return {"kind": kind, "value": {"type": typ, "role": role, "position": position}}
+    item = _text(token["value"])
+    if kind == "operator" and item not in OPS or kind == "policy_state" and item not in POLICY or kind == "identifier" and not IDENTIFIER.fullmatch(item):
+        raise IdentityError("invalid typed token")
+    return {"kind": kind, "value": item}
+def _anchors(value: Any) -> list[dict[str, Any]]:
+    if not isinstance(value, list) or not value or len(value) > MAX_ITEMS:
+        raise IdentityError("invalid anchors")
+    result = [_token(item) for item in value]
+    for index, token in enumerate(result):
+        if token["kind"] != ("identifier" if index % 2 == 0 else "operator") or token["kind"] == "operator" and token["value"] != "::":
+            raise IdentityError("invalid anchor grammar")
+    if result[-1]["kind"] != "identifier":
+        raise IdentityError("anchor must end with identifier")
+    return result
+def _clauses(value: Any, required: bool) -> list[list[dict[str, Any]]]:
+    if not isinstance(value, list) or len(value) > MAX_ITEMS or required and not value:
+        raise IdentityError("invalid clauses")
+    result = []
+    for clause in value:
+        if not isinstance(clause, list) or not clause or len(clause) > MAX_ITEMS:
+            raise IdentityError("invalid clause")
+        result.append([_token(item) for item in clause])
+    return result
+def _json(value: Any) -> str:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+def _hash(value: Any) -> str:
+    return hashlib.sha256(_json(value).encode()).hexdigest()
+def validate_identity(payload: Any) -> CanonicalIdentity:
+    required = {"schema", "canonicalizer_version", "scope", "anchors", "predicates", "required_behavior", "forbidden_behavior", "ordering_constraints"}
+    fields = _obj(payload, required)
+    if fields["schema"] != SCHEMA or fields["canonicalizer_version"] != VERSION:
+        raise IdentityError("unsupported schema")
+    canonical = {"schema": SCHEMA, "canonicalizer_version": VERSION, "scope": _scope(fields["scope"]), "anchors": _anchors(fields["anchors"]), "predicates": _clauses(fields["predicates"], True), "required_behavior": _clauses(fields["required_behavior"], True), "forbidden_behavior": _clauses(fields["forbidden_behavior"], False), "ordering_constraints": _clauses(fields["ordering_constraints"], False)}
+    key = _hash({name: canonical[name] for name in ("schema", "canonicalizer_version", "scope", "anchors", "predicates")})
+    behavior = _hash({"contract_key": key, "required_behavior": canonical["required_behavior"], "forbidden_behavior": canonical["forbidden_behavior"], "ordering_constraints": canonical["ordering_constraints"]})
+    identity = CanonicalIdentity(_json(canonical), key, behavior, _hash({"contract_key": key, "behavior_digest": behavior}))
+    if len(_json(identity.as_record()).encode()) > MAX_BYTES:
+        raise IdentityError("identity too large")
+    return identity
+def verify_identity_record(record: Any) -> CanonicalIdentity:
+    payload_fields = {"schema", "canonicalizer_version", "scope", "anchors", "predicates", "required_behavior", "forbidden_behavior", "ordering_constraints"}
+    digest_fields = {"contract_key", "behavior_digest", "fingerprint_v2"}
+    if not isinstance(record, dict) or set(record) != payload_fields | digest_fields or any(not isinstance(record[name], str) or not DIGEST.fullmatch(record[name]) for name in digest_fields):
+        raise IdentityError("invalid verified record")
+    identity = validate_identity({name: record[name] for name in payload_fields})
+    if (identity.contract_key, identity.behavior_digest, identity.fingerprint_v2) != tuple(record[name] for name in ("contract_key", "behavior_digest", "fingerprint_v2")) or identity.as_record() != record:
+        raise IdentityError("record mismatch")
+    return identity
+def canonical_json(identity: CanonicalIdentity) -> str:
+    if not isinstance(identity, CanonicalIdentity):
+        raise IdentityError("expected identity")
+    return _json(identity.as_record())

tests/test_canonical_typed_identity_r1c.py

@@ -0,0 +1,90 @@
+import copy
+import json
+import unittest
+from scripts.canonical_typed_identity import IdentityError, canonical_json, validate_identity, verify_identity_record
+class R1cIdentityTests(unittest.TestCase):
+    def tok(self, kind, value):
+        return {"kind": kind, "value": value}
+    def payload(self):
+        return {
+            "schema": "contract_identity.v2",
+            "canonicalizer_version": "structured_tokens.v2",
+            "scope": {"repo": "AcMe/Audit-Bridge", "file": "service/review.py", "category": "contract"},
+            "anchors": [self.tok("identifier", "Namespace"), self.tok("operator", "::"), self.tok("identifier", "validate()")],
+            "predicates": [[self.tok("identifier", "score"), self.tok("operator", ">="), self.tok("identifier", "threshold")]],
+            "required_behavior": [[self.tok("policy_state", "required")]],
+            "forbidden_behavior": [],
+            "ordering_constraints": [],
+        }
+    def invalid(self, value):
+        with self.assertRaises(IdentityError):
+            validate_identity(value)
+    def test_near_miss_identifiers_are_structural_not_secret_classified(self):
+        for name in ("ghs_database", "github_pat_validator", "Eurasia", "keyJson", "secret_manager", "secret_ref_validator", "api_config"):
+            value = self.payload()
+            value["anchors"] = [self.tok("identifier", name)]
+            validate_identity(value)
+    def test_secret_ref_is_finite_and_has_no_raw_value(self):
+        value = self.payload()
+        value["predicates"] = [[self.tok("secret_ref", {"type": "credential", "role": "auth", "position": 0})]]
+        record = validate_identity(value).as_record()
+        self.assertEqual(record["predicates"][0][0]["value"], {"type": "credential", "role": "auth", "position": 0})
+        for ref in ({"type": "ghs_database", "role": "auth", "position": 0}, {"type": "credential", "role": "auth", "position": 0, "raw": "x"}, {"type": "credential", "role": "other", "position": 0}, {"type": "credential", "role": "auth", "position": 1025}):
+            bad = self.payload()
+            bad["predicates"] = [[self.tok("secret_ref", ref)]]
+            self.invalid(bad)
+        for typ, role in (([], "auth"), ({"name": "credential"}, "auth"), ("credential", []), ("credential", {"name": "auth"}), (1, "auth"), ("credential", 1)):
+            bad = self.payload()
+            bad["predicates"] = [[self.tok("secret_ref", {"type": typ, "role": role, "position": 0})]]
+            self.invalid(bad)
+    def test_wire_is_canonical_and_digests_are_exact(self):
+        identity = validate_identity(self.payload())
+        record = identity.as_record()
+        self.assertEqual(verify_identity_record(record), identity)
+        self.assertEqual(canonical_json(identity), canonical_json(verify_identity_record(json.loads(canonical_json(identity)))))
+        for field in ("contract_key", "behavior_digest", "fingerprint_v2"):
+            bad = copy.deepcopy(record)
+            bad[field] = "0" * 64
+            with self.assertRaises(IdentityError):
+                verify_identity_record(bad)
+        bad = copy.deepcopy(record)
+        bad["scope"]["repo"] = "AcMe/Audit-Bridge"
+        with self.assertRaises(IdentityError):
+            verify_identity_record(bad)
+    def test_unknown_evidence_severity_prose_assignment_and_controls_reject(self):
+        for field, value in (("evidence", {}), ("description", "raw prose"), ("severity", "high")):
+            bad = self.payload()
+            bad[field] = value
+            self.invalid(bad)
+        for text in ("raw prose", "password=secret", "bad\x00name", "bad\ud800name"):
+            bad = self.payload()
+            bad["anchors"] = [self.tok("identifier", text)]
+            self.invalid(bad)
+        bad = self.payload()
+        bad["anchors"] = [self.tok("mystery", "x")]
+        self.invalid(bad)
+    def test_v1_is_not_migrated_and_key_order_is_stable(self):
+        old = self.payload()
+        old["canonicalizer_version"] = "structured_tokens.v1"
+        self.invalid(old)
+        value = self.payload()
+        reordered = {key: value[key] for key in reversed(tuple(value))}
+        self.assertEqual(validate_identity(value).contract_key, validate_identity(reordered).contract_key)
+    def test_operators_order_anchor_shape_path_category_and_bounds(self):
+        for operator in (">=", "<=", ">", "<", "==", "!=", "===", "!==", "->", "=>", "::"):
+            value = self.payload()
+            value["predicates"][0][1]["value"] = operator
+            validate_identity(value)
+        for anchors in ([self.tok("operator", "::")], [self.tok("identifier", "A"), self.tok("operator", "::")], [self.tok("identifier", "A::B")]):
+            value = self.payload()
+            value["anchors"] = anchors
+            self.invalid(value)
+        for path in ("/abs.py", "../escape.py", "a//b.py", "a\\b.py"):
+            value = self.payload()
+            value["scope"]["file"] = path
+            self.invalid(value)
+        value = self.payload()
+        value["scope"]["file"] = "x" * 1025
+        self.invalid(value)
+if __name__ == "__main__":
+    unittest.main()