diff --git a/scripts/canonical_typed_identity.py b/scripts/canonical_typed_identity.py new file mode 100644 index 0000000..9db4d96 --- /dev/null +++ b/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 diff --git a/tests/test_canonical_typed_identity_r1d.py b/tests/test_canonical_typed_identity_r1d.py new file mode 100644 index 0000000..3348067 --- /dev/null +++ b/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()