From 943a3ca3455af8f5d3a182580b52d98d3afdf2ce Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 13 Jul 2026 18:22:31 +0800 Subject: [PATCH 1/2] feat: add strict typed identity v2 Co-Authored-By: Codex --- scripts/canonical_typed_identity.py | 134 +++++++++++++++++++++ tests/test_canonical_typed_identity_r1c.py | 86 +++++++++++++ 2 files changed, 220 insertions(+) create mode 100644 scripts/canonical_typed_identity.py create mode 100644 tests/test_canonical_typed_identity_r1c.py diff --git a/scripts/canonical_typed_identity.py b/scripts/canonical_typed_identity.py new file mode 100644 index 0000000..45e1469 --- /dev/null +++ b/scripts/canonical_typed_identity.py @@ -0,0 +1,134 @@ +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 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()) diff --git a/tests/test_canonical_typed_identity_r1c.py b/tests/test_canonical_typed_identity_r1c.py new file mode 100644 index 0000000..f6cb5dd --- /dev/null +++ b/tests/test_canonical_typed_identity_r1c.py @@ -0,0 +1,86 @@ +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) + 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() From ed084d611432aa9f5c1af466ef4ed4fc0ff92861 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 13 Jul 2026 19:16:31 +0800 Subject: [PATCH 2/2] fix: validate secret reference metadata types Co-Authored-By: Codex --- scripts/canonical_typed_identity.py | 3 +++ tests/test_canonical_typed_identity_r1c.py | 4 ++++ 2 files changed, 7 insertions(+) diff --git a/scripts/canonical_typed_identity.py b/scripts/canonical_typed_identity.py index 45e1469..aafae97 100644 --- a/scripts/canonical_typed_identity.py +++ b/scripts/canonical_typed_identity.py @@ -77,6 +77,9 @@ def _token(value: Any) -> dict[str, Any]: 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}} diff --git a/tests/test_canonical_typed_identity_r1c.py b/tests/test_canonical_typed_identity_r1c.py index f6cb5dd..8e07ac7 100644 --- a/tests/test_canonical_typed_identity_r1c.py +++ b/tests/test_canonical_typed_identity_r1c.py @@ -33,6 +33,10 @@ def test_secret_ref_is_finite_and_has_no_raw_value(self): 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()