Skip to content

Commit 650167b

Browse files
Pigbibicodex
Pigbibi
and
codex
committed
feat: add bounded identity record validator
Co-Authored-By: Codex <noreply@openai.com>
1 parent 6cc545a commit 650167b

2 files changed

Lines changed: 198 additions & 0 deletions

File tree

‎scripts/canonical_typed_identity.py‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
from __future__ import annotations
2+
import hashlib
3+
import json
4+
import re
5+
import unicodedata
6+
from dataclasses import dataclass
7+
from typing import Any
8+
SCHEMA = "contract_identity.v2"
9+
VERSION = "structured_tokens.v1"
10+
OPS = frozenset({">=", "<=", ">", "<", "==", "!=", "===", "!==", "->", "=>", "::"})
11+
POLICY = frozenset({"required", "forbidden", "present", "absent", "enabled", "disabled", "valid", "invalid", "missing", "optional"})
12+
KINDS = frozenset({"identifier", "operator", "policy_state", "secret_ref"})
13+
CATEGORIES = frozenset({"bug", "contract", "logic", "performance", "reliability", "security"})
14+
SEVERITIES = frozenset({"critical", "high", "medium", "low"})
15+
OWNER = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,37}[A-Za-z0-9])?$")
16+
REPO = re.compile(r"^[A-Za-z0-9._-]{1,100}$")
17+
IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*(?:\(\))?$")
18+
DIGEST = re.compile(r"^[0-9a-f]{64}$")
19+
MARKERS = ("github_pat_", "ghp_", "ghs_", "aws_secret_access_key", "akia", "asia", "sk-", "eyj")
20+
MAX_ITEMS = 32
21+
MAX_BYTES = 64 * 1024
22+
class IdentityError(ValueError):
23+
pass
24+
@dataclass(frozen=True)
25+
class CanonicalIdentity:
26+
_payload_json: str
27+
contract_key: str
28+
behavior_digest: str
29+
fingerprint_v2: str
30+
@property
31+
def payload(self) -> dict[str, Any]:
32+
return json.loads(self._payload_json)
33+
def as_record(self) -> dict[str, Any]:
34+
result = self.payload
35+
result.update(contract_key=self.contract_key, behavior_digest=self.behavior_digest, fingerprint_v2=self.fingerprint_v2)
36+
return result
37+
def _obj(value: Any, required: set[str], optional: set[str] = set()) -> dict[str, Any]:
38+
if not isinstance(value, dict) or len(value) > len(required) + len(optional) or not required <= set(value) or not set(value) <= required | optional:
39+
raise IdentityError("invalid object fields")
40+
return value
41+
def _text(value: Any, limit: int = 512) -> str:
42+
if not isinstance(value, str) or len(value) > limit or len(value.encode()) > limit:
43+
raise IdentityError("invalid bounded text")
44+
value = unicodedata.normalize("NFC", value)
45+
if any(unicodedata.category(char).startswith("C") for char in value) or len(value.encode()) > limit:
46+
raise IdentityError("invalid text")
47+
return value
48+
def _scope(value: Any) -> dict[str, str]:
49+
scope = _obj(value, {"repo", "file", "category"})
50+
repo = _text(scope["repo"], 140)
51+
parts = repo.split("/")
52+
if len(parts) != 2 or not OWNER.fullmatch(parts[0]) or "--" in parts[0] or not REPO.fullmatch(parts[1]) or parts[1] in {".", ".."}:
53+
raise IdentityError("invalid repo")
54+
path = _text(scope["file"], 1024)
55+
if path.startswith("/") or "\\" in path or any(part in {"", ".", ".."} for part in path.split("/")):
56+
raise IdentityError("invalid path")
57+
category = _text(scope["category"], 32)
58+
if category not in CATEGORIES:
59+
raise IdentityError("invalid category")
60+
return {"repo": f"{parts[0].lower()}/{parts[1].lower()}", "file": path, "category": category}
61+
def _token(value: Any) -> dict[str, Any]:
62+
token = _obj(value, {"kind", "value"})
63+
kind = _text(token["kind"], 32)
64+
if kind not in KINDS:
65+
raise IdentityError("unknown token kind")
66+
if kind == "secret_ref":
67+
ref = _obj(token["value"], {"type", "role", "position"})
68+
typ, role, position = _text(ref["type"], 32), _text(ref["role"], 32), ref["position"]
69+
if not re.fullmatch(r"[a-z][a-z0-9_.-]{0,31}", typ) or not re.fullmatch(r"[a-z][a-z0-9_.-]{0,31}", role) or isinstance(position, bool) or not isinstance(position, int) or not 0 <= position <= 1024:
70+
raise IdentityError("invalid secret reference")
71+
return {"kind": kind, "value": {"type": typ, "role": role, "position": position}}
72+
item = _text(token["value"])
73+
if any(marker in item.lower() for marker in MARKERS):
74+
raise IdentityError("reserved secret marker")
75+
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):
76+
raise IdentityError("invalid typed token")
77+
return {"kind": kind, "value": item}
78+
def _anchors(value: Any) -> list[dict[str, Any]]:
79+
if not isinstance(value, list) or not value or len(value) > MAX_ITEMS:
80+
raise IdentityError("invalid anchors")
81+
result = [_token(item) for item in value]
82+
for index, token in enumerate(result):
83+
if token["kind"] != ("identifier" if index % 2 == 0 else "operator") or token["kind"] == "operator" and token["value"] != "::":
84+
raise IdentityError("invalid anchor grammar")
85+
if result[-1]["kind"] != "identifier":
86+
raise IdentityError("anchor must end with identifier")
87+
return result
88+
def _clauses(value: Any, required: bool) -> list[list[dict[str, Any]]]:
89+
if not isinstance(value, list) or len(value) > MAX_ITEMS or required and not value:
90+
raise IdentityError("invalid clauses")
91+
result = []
92+
for clause in value:
93+
if not isinstance(clause, list) or not clause or len(clause) > MAX_ITEMS:
94+
raise IdentityError("invalid clause")
95+
result.append([_token(item) for item in clause])
96+
return result
97+
def _json(value: Any) -> str:
98+
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
99+
def _hash(value: Any) -> str:
100+
return hashlib.sha256(_json(value).encode()).hexdigest()
101+
def validate_identity(payload: Any) -> CanonicalIdentity:
102+
fields = _obj(payload, {"schema", "canonicalizer_version", "scope", "anchors", "predicates", "required_behavior", "forbidden_behavior", "ordering_constraints"}, {"severity"})
103+
if fields["schema"] != SCHEMA or fields["canonicalizer_version"] != VERSION:
104+
raise IdentityError("unsupported schema")
105+
if "severity" in fields and _text(fields["severity"], 16) not in SEVERITIES:
106+
raise IdentityError("invalid severity")
107+
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)}
108+
key = _hash({name: canonical[name] for name in ("schema", "canonicalizer_version", "scope", "anchors", "predicates")})
109+
behavior = _hash({"contract_key": key, "required_behavior": canonical["required_behavior"], "forbidden_behavior": canonical["forbidden_behavior"], "ordering_constraints": canonical["ordering_constraints"]})
110+
identity = CanonicalIdentity(_json(canonical), key, behavior, _hash({"contract_key": key, "behavior_digest": behavior}))
111+
if len(_json(identity.as_record()).encode()) > MAX_BYTES:
112+
raise IdentityError("identity too large")
113+
return identity
114+
def verify_identity_record(record: Any) -> CanonicalIdentity:
115+
fields = {"schema", "canonicalizer_version", "scope", "anchors", "predicates", "required_behavior", "forbidden_behavior", "ordering_constraints", "contract_key", "behavior_digest", "fingerprint_v2"}
116+
if not isinstance(record, dict) or set(record) != fields or any(not isinstance(record[name], str) or not DIGEST.fullmatch(record[name]) for name in fields - {"schema", "canonicalizer_version", "scope", "anchors", "predicates", "required_behavior", "forbidden_behavior", "ordering_constraints"}):
117+
raise IdentityError("invalid verified record")
118+
identity = validate_identity({name: record[name] for name in fields - {"contract_key", "behavior_digest", "fingerprint_v2"}})
119+
if (identity.contract_key, identity.behavior_digest, identity.fingerprint_v2) != tuple(record[name] for name in ("contract_key", "behavior_digest", "fingerprint_v2")):
120+
raise IdentityError("digest mismatch")
121+
return identity
122+
def canonical_json(identity: CanonicalIdentity) -> str:
123+
if not isinstance(identity, CanonicalIdentity):
124+
raise IdentityError("expected identity")
125+
return _json(identity.as_record())

‎tests/test_canonical_typed_identity_r1b.py‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
import copy
2+
import unittest
3+
from scripts.canonical_typed_identity import IdentityError, validate_identity, verify_identity_record
4+
class R1bIdentityTests(unittest.TestCase):
5+
def tok(self, kind, value):
6+
return {"kind": kind, "value": value}
7+
def payload(self, severity="high"):
8+
return {
9+
"schema": "contract_identity.v2",
10+
"canonicalizer_version": "structured_tokens.v1",
11+
"scope": {"repo": "AcMe/Audit-Bridge", "file": "service/review.py", "category": "contract"},
12+
"anchors": [self.tok("identifier", "Namespace"), self.tok("operator", "::"), self.tok("identifier", "validate()")],
13+
"predicates": [[self.tok("identifier", "score"), self.tok("operator", ">="), self.tok("identifier", "threshold")]],
14+
"required_behavior": [[self.tok("policy_state", "required")]],
15+
"forbidden_behavior": [],
16+
"ordering_constraints": [],
17+
"severity": severity,
18+
}
19+
def invalid(self, value):
20+
with self.assertRaises(IdentityError):
21+
validate_identity(value)
22+
def test_reserved_markers_anywhere_and_safe_identifiers(self):
23+
for safe in ("secret_manager", "secret_ref_validator"):
24+
value = self.payload()
25+
value["anchors"] = [self.tok("identifier", safe)]
26+
validate_identity(value)
27+
for marker in ("ghs_123", "env.ghs_123", "ASIA123", "key.ASIA123", "api.sk-secret", "env.eyJtoken"):
28+
value = self.payload()
29+
value["anchors"] = [self.tok("identifier", marker)]
30+
self.invalid(value)
31+
def test_anchor_grammar_requires_explicit_namespace_tokens(self):
32+
for anchors in (
33+
[self.tok("operator", "::")],
34+
[self.tok("identifier", "A"), self.tok("operator", "::")],
35+
[self.tok("identifier", "A"), self.tok("operator", "::"), self.tok("operator", "::"), self.tok("identifier", "B")],
36+
[self.tok("identifier", "A"), self.tok("operator", "=>"), self.tok("identifier", "B")],
37+
[self.tok("identifier", "A::B")],
38+
):
39+
value = self.payload()
40+
value["anchors"] = anchors
41+
self.invalid(value)
42+
def test_severity_is_not_verified_metadata(self):
43+
identity = validate_identity(self.payload())
44+
critical = validate_identity(self.payload("critical"))
45+
self.assertEqual((identity.contract_key, identity.behavior_digest, identity.fingerprint_v2), (critical.contract_key, critical.behavior_digest, critical.fingerprint_v2))
46+
record = identity.as_record()
47+
self.assertNotIn("severity", record)
48+
self.assertEqual(verify_identity_record(record), identity)
49+
for severity in (None, "critical"):
50+
tampered = copy.deepcopy(record)
51+
tampered["severity"] = severity
52+
with self.assertRaises(IdentityError):
53+
verify_identity_record(tampered)
54+
def test_repo_is_lowercase_but_file_and_identifier_case_remain(self):
55+
identity = validate_identity(self.payload())
56+
self.assertEqual(identity.payload["scope"]["repo"], "acme/audit-bridge")
57+
self.assertEqual(identity.payload["scope"]["file"], "service/review.py")
58+
self.assertEqual(identity.payload["anchors"][0]["value"], "Namespace")
59+
lower = copy.deepcopy(self.payload())
60+
lower["scope"]["repo"] = "acme/audit-bridge"
61+
self.assertEqual(identity.contract_key, validate_identity(lower).contract_key)
62+
def test_unknown_evidence_secret_ref_and_digest_tamper_fail_closed(self):
63+
value = self.payload()
64+
value["predicates"] = [[self.tok("secret_ref", {"type": "credential", "role": "auth", "position": 0})]]
65+
record = validate_identity(value).as_record()
66+
self.assertNotIn("secret", record["predicates"][0][0]["value"])
67+
for field in ("evidence", "unknown"):
68+
invalid = self.payload()
69+
invalid[field] = {}
70+
self.invalid(invalid)
71+
record["contract_key"] = "0" * 64
72+
with self.assertRaises(IdentityError):
73+
verify_identity_record(record)

0 commit comments

Comments
 (0)