From 692b2594b86bfad2c3c4b265dfb6924d0beb11f8 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 13 Jul 2026 03:25:38 +0800 Subject: [PATCH 1/4] feat: define canonical contract identity schema Co-Authored-By: Codex --- scripts/contract_identity.py | 346 ++++++++++++++++++++++++++++++++ tests/test_contract_identity.py | 212 +++++++++++++++++++ 2 files changed, 558 insertions(+) create mode 100644 scripts/contract_identity.py create mode 100644 tests/test_contract_identity.py diff --git a/scripts/contract_identity.py b/scripts/contract_identity.py new file mode 100644 index 0000000..a75d53c --- /dev/null +++ b/scripts/contract_identity.py @@ -0,0 +1,346 @@ +"""Pure canonical contract identity schema and digest helpers.""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import re +import unicodedata +from collections import defaultdict +from dataclasses import dataclass +from typing import Any + +SCHEMA = "contract_identity.v2" +CANONICALIZER_VERSION = "operator_tokens.v1" +ALLOWED_CATEGORIES = frozenset( + {"bug", "contract", "logic", "performance", "reliability", "security"} +) +ALLOWED_SEVERITIES = frozenset({"critical", "high", "medium", "low"}) +ALLOWED_ANCHOR_KINDS = frozenset( + {"config", "endpoint", "field", "identifier", "schema", "symbol", "type"} +) + +MAX_REPO_BYTES = 200 +MAX_FILE_BYTES = 512 +MAX_ANCHOR_BYTES = 256 +MAX_TOKEN_BYTES = 256 +MAX_EVIDENCE_BYTES = 256 +MAX_ITEMS = 32 +MAX_TOKENS_PER_CLAUSE = 128 +MAX_CANONICAL_BYTES = 16_384 + +_TOP_REQUIRED = frozenset( + { + "schema", + "canonicalizer_version", + "scope", + "anchors", + "predicates", + "required_behavior", + "forbidden_behavior", + "ordering_constraints", + "evidence", + } +) +_DIGEST_FIELDS = frozenset({"contract_key", "behavior_digest", "fingerprint_v2"}) +_OPERATORS = ("===", "!==", ">=", "<=", "==", "!=", "->", "=>", "::", ">", "<") +_TOKEN_RE = re.compile( + r"|===|!==|>=|<=|==|!=|->|=>|::|>|<|[\w.$]+|[^\s]", + re.UNICODE, +) +_PLACEHOLDER_RE = re.compile(r"") +_SECRET_PATTERNS = ( + ("PRIVATE_KEY", re.compile(r"-----BEGIN [^-]+-----.*?-----END [^-]+-----", re.I | re.S)), + ("AWS", re.compile(r"\b(?:AKIA|ASIA|AIDA|AROA)[A-Z0-9]{16}\b")), + ("SLACK", re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,}\b")), + ("GITLAB", re.compile(r"\bglpat-[A-Za-z0-9_-]{10,}\b")), + ("JWT", re.compile(r"\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b")), + ("DSN", re.compile(r"\b[a-z][a-z0-9+.-]*://[^\s/@:]+:[^\s/@]+@[^\s]+", re.I)), + ("BEARER", re.compile(r"(?i)\bbearer\s+\S+")), + ("CREDENTIAL", re.compile( + r"(?i)(? dict[str, str]: + return {"repo": self.repo, "file": self.file, "category": self.category} + + +@dataclass(frozen=True) +class Anchor: + kind: str + value: str + + def as_dict(self) -> dict[str, str]: + return {"kind": self.kind, "value": self.value} + + +@dataclass(frozen=True) +class Evidence: + head_sha: str + diff_digest: str + file: str + location_or_hunk_digest: str + + def as_dict(self) -> dict[str, str]: + return { + "head_sha": self.head_sha, + "diff_digest": self.diff_digest, + "file": self.file, + "location_or_hunk_digest": self.location_or_hunk_digest, + } + + +@dataclass(frozen=True) +class ContractIdentity: + scope: Scope + anchors: tuple[Anchor, ...] + predicates: tuple[tuple[str, ...], ...] + required_behavior: tuple[tuple[str, ...], ...] + forbidden_behavior: tuple[tuple[str, ...], ...] + ordering_constraints: tuple[tuple[str, ...], ...] + evidence: Evidence + severity: str | None + contract_key: str + behavior_digest: str + fingerprint_v2: str + + def as_record(self) -> dict[str, Any]: + record: dict[str, Any] = { + "schema": SCHEMA, + "canonicalizer_version": CANONICALIZER_VERSION, + "scope": self.scope.as_dict(), + "anchors": [anchor.as_dict() for anchor in self.anchors], + "predicates": [list(clause) for clause in self.predicates], + "required_behavior": [list(clause) for clause in self.required_behavior], + "forbidden_behavior": [list(clause) for clause in self.forbidden_behavior], + "ordering_constraints": [list(clause) for clause in self.ordering_constraints], + "evidence": self.evidence.as_dict(), + "contract_key": self.contract_key, + "behavior_digest": self.behavior_digest, + "fingerprint_v2": self.fingerprint_v2, + } + if self.severity is not None: + record["severity"] = self.severity + return record + + +class _SecretRedactor: + def __init__(self, *, allow_placeholders: bool) -> None: + self.allow_placeholders = allow_placeholders + self.counts: dict[str, int] = defaultdict(int) + + def redact(self, value: str, field: str) -> str: + if " str: + self.counts[kind] += 1 + return f"" + + text = pattern.sub(replacement, text) + malformed = re.search(r"]*>", text) + if malformed and not _PLACEHOLDER_RE.fullmatch(malformed.group(0)): + raise IdentityValidationError(f"{field} contains an invalid secret placeholder") + return text + + +def _exact_fields(value: Any, required: frozenset[str], optional: frozenset[str], field: str) -> dict[str, Any]: + if not isinstance(value, dict) or set(value) != required | (set(value) & optional): + raise IdentityValidationError(f"{field} has missing or extra fields") + return value + + +def _text(value: Any, field: str, max_bytes: int, redactor: _SecretRedactor) -> str: + if not isinstance(value, str): + raise IdentityValidationError(f"{field} must be a string") + normalized = unicodedata.normalize("NFC", value) + if not normalized or any(unicodedata.category(char).startswith("C") for char in normalized): + raise IdentityValidationError(f"{field} is empty or contains control characters") + normalized = redactor.redact(normalized, field) + if len(normalized.encode("utf-8")) > max_bytes: + raise IdentityValidationError(f"{field} exceeds {max_bytes} bytes") + return normalized + + +def _repo(value: Any, field: str, redactor: _SecretRedactor) -> str: + repo = _text(value, field, MAX_REPO_BYTES, redactor) + if not re.fullmatch(r"[A-Za-z0-9._-]+/[A-Za-z0-9._-]+", repo): + raise IdentityValidationError(f"{field} must be owner/name") + return repo + + +def _relative_path(value: Any, field: str, redactor: _SecretRedactor) -> str: + path = _text(value, field, MAX_FILE_BYTES, redactor) + parts = path.split("/") + if ( + " str: + reference = _text(value, field, MAX_EVIDENCE_BYTES, redactor) + if " tuple[tuple[str, ...], ...]: + if not isinstance(value, list) or len(value) > MAX_ITEMS or (not value and not allow_empty): + raise IdentityValidationError(f"{field} must be a bounded list") + clauses: list[tuple[str, ...]] = [] + for index, clause in enumerate(value): + if not isinstance(clause, list) or not clause: + raise IdentityValidationError(f"{field}[{index}] must be a non-empty token list") + tokens: list[str] = [] + for raw_token in clause: + token = _text(raw_token, f"{field}[{index}]", MAX_TOKEN_BYTES, redactor) + tokens.extend(_TOKEN_RE.findall(token)) + if not tokens or len(tokens) > MAX_TOKENS_PER_CLAUSE: + raise IdentityValidationError(f"{field}[{index}] has invalid token count") + clauses.append(tuple(tokens)) + return tuple(clauses) + + +def _stable_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(_stable_json(value)).hexdigest() + + +def build_contract_identity(payload: dict[str, Any], *, _allow_placeholders: bool = False) -> ContractIdentity: + top = _exact_fields(payload, _TOP_REQUIRED, frozenset({"severity"}), "identity") + if top["schema"] != SCHEMA or top["canonicalizer_version"] != CANONICALIZER_VERSION: + raise IdentityValidationError("unsupported identity schema or canonicalizer") + redactor = _SecretRedactor(allow_placeholders=_allow_placeholders) + + raw_scope = _exact_fields(top["scope"], frozenset({"repo", "file", "category"}), frozenset(), "scope") + category = str(raw_scope["category"]).lower() if isinstance(raw_scope["category"], str) else "" + if category not in ALLOWED_CATEGORIES: + raise IdentityValidationError("scope.category is not controlled") + scope = Scope( + repo=_repo(raw_scope["repo"], "scope.repo", redactor), + file=_relative_path(raw_scope["file"], "scope.file", redactor), + category=category, + ) + + raw_anchors = top["anchors"] + if not isinstance(raw_anchors, list) or not raw_anchors or len(raw_anchors) > MAX_ITEMS: + raise IdentityValidationError("anchors must be a non-empty bounded list") + anchors: list[Anchor] = [] + for index, raw_anchor in enumerate(raw_anchors): + anchor = _exact_fields(raw_anchor, frozenset({"kind", "value"}), frozenset(), f"anchors[{index}]") + kind = str(anchor["kind"]).lower() if isinstance(anchor["kind"], str) else "" + if kind not in ALLOWED_ANCHOR_KINDS: + raise IdentityValidationError(f"anchors[{index}].kind is not controlled") + anchors.append(Anchor(kind, _text(anchor["value"], f"anchors[{index}].value", MAX_ANCHOR_BYTES, redactor))) + + predicates = _clauses(top["predicates"], "predicates", redactor, allow_empty=False) + required = _clauses( + top["required_behavior"], "required_behavior", redactor, allow_empty=False + ) + forbidden = _clauses( + top["forbidden_behavior"], "forbidden_behavior", redactor, allow_empty=True + ) + ordering = _clauses( + top["ordering_constraints"], "ordering_constraints", redactor, allow_empty=True + ) + + raw_evidence = _exact_fields( + top["evidence"], + frozenset({"head_sha", "diff_digest", "file", "location_or_hunk_digest"}), + frozenset(), + "evidence", + ) + head_sha = _text(raw_evidence["head_sha"], "evidence.head_sha", 64, redactor).lower() + diff_digest = _text(raw_evidence["diff_digest"], "evidence.diff_digest", 64, redactor).lower() + if not re.fullmatch(r"[0-9a-f]{7,64}", head_sha) or not re.fullmatch(r"[0-9a-f]{64}", diff_digest): + raise IdentityValidationError("evidence digests are malformed") + evidence = Evidence( + head_sha=head_sha, + diff_digest=diff_digest, + file=_relative_path(raw_evidence["file"], "evidence.file", redactor), + location_or_hunk_digest=_reference( + raw_evidence["location_or_hunk_digest"], + "evidence.location_or_hunk_digest", + redactor, + ), + ) + if evidence.file != scope.file: + raise IdentityValidationError("evidence.file must match scope.file") + severity = top.get("severity") + if severity is not None: + severity = str(severity).lower() if isinstance(severity, str) else "" + if severity not in ALLOWED_SEVERITIES: + raise IdentityValidationError("severity is not controlled") + + contract_payload = { + "schema": SCHEMA, + "canonicalizer_version": CANONICALIZER_VERSION, + "scope": scope.as_dict(), + "anchors": [anchor.as_dict() for anchor in anchors], + "predicates": [list(clause) for clause in predicates], + } + contract_key = _digest(contract_payload) + behavior_payload = { + "contract_key": contract_key, + "required_behavior": [list(clause) for clause in required], + "forbidden_behavior": [list(clause) for clause in forbidden], + "ordering_constraints": [list(clause) for clause in ordering], + } + behavior_digest = _digest(behavior_payload) + fingerprint_v2 = _digest( + {"contract_key": contract_key, "behavior_digest": behavior_digest} + ) + identity = ContractIdentity( + scope, tuple(anchors), predicates, required, forbidden, ordering, + evidence, severity, contract_key, behavior_digest, fingerprint_v2, + ) + if len(_stable_json(identity.as_record())) > MAX_CANONICAL_BYTES: + raise IdentityValidationError("canonical identity exceeds total byte limit") + return identity + + +def verify_persisted_identity(record: dict[str, Any]) -> ContractIdentity: + top = _exact_fields(record, _TOP_REQUIRED | _DIGEST_FIELDS, frozenset({"severity"}), "record") + expected = {field: top[field] for field in _DIGEST_FIELDS} + if any(not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value) for value in expected.values()): + raise IdentityValidationError("persisted identity digest is malformed") + payload = {key: value for key, value in top.items() if key not in _DIGEST_FIELDS} + identity = build_contract_identity(payload, _allow_placeholders=True) + if not all(hmac.compare_digest(expected[field], getattr(identity, field)) for field in _DIGEST_FIELDS): + raise IdentityValidationError("persisted identity digest mismatch") + return identity + + +def canonical_json(identity: ContractIdentity) -> str: + return _stable_json(identity.as_record()).decode("utf-8") + + +def operators() -> tuple[str, ...]: + return _OPERATORS diff --git a/tests/test_contract_identity.py b/tests/test_contract_identity.py new file mode 100644 index 0000000..aa528ec --- /dev/null +++ b/tests/test_contract_identity.py @@ -0,0 +1,212 @@ +from __future__ import annotations + +import copy +import json +import unittest + +from scripts.contract_identity import ( + MAX_ANCHOR_BYTES, + MAX_TOKEN_BYTES, + IdentityValidationError, + build_contract_identity, + canonical_json, + operators, + verify_persisted_identity, +) + + +class ContractIdentityTests(unittest.TestCase): + def payload(self) -> dict[str, object]: + return { + "schema": "contract_identity.v2", + "canonicalizer_version": "operator_tokens.v1", + "scope": { + "repo": "org/audit-bridge", + "file": "service/review.py", + "category": "contract", + }, + "anchors": [{"kind": "symbol", "value": "Review.validate()"}], + "predicates": [["score>=threshold"]], + "required_behavior": [["return", "blocked"]], + "forbidden_behavior": [], + "ordering_constraints": [["validate", "before", "dispatch"]], + "evidence": { + "head_sha": "deadbeef", + "diff_digest": "a" * 64, + "file": "service/review.py", + "location_or_hunk_digest": "hunk:review:1", + }, + "severity": "high", + } + + def test_operator_matrix_is_atomic_and_collision_resistant(self) -> None: + fingerprints = set() + for operator in operators(): + payload = self.payload() + payload["predicates"] = [[f"left{operator}right"]] + identity = build_contract_identity(payload) + self.assertIn(operator, identity.predicates[0]) + fingerprints.add(identity.fingerprint_v2) + self.assertEqual(len(fingerprints), len(operators())) + + def test_anchor_and_predicate_matrix_changes_contract_key(self) -> None: + schema = self.payload() + schema["anchors"] = [{"kind": "schema", "value": "schema_v2"}] + fingerprint = copy.deepcopy(schema) + fingerprint["anchors"] = [{"kind": "schema", "value": "fingerprint_v2"}] + self.assertNotEqual( + build_contract_identity(schema).contract_key, + build_contract_identity(fingerprint).contract_key, + ) + + auth = self.payload() + auth["predicates"] = [["validate()", "checks", "auth_header"]] + database = copy.deepcopy(auth) + database["predicates"] = [["validate()", "prevents", "database_leak"]] + self.assertNotEqual( + build_contract_identity(auth).contract_key, + build_contract_identity(database).contract_key, + ) + + unrelated = self.payload() + unrelated["anchors"] = [{"kind": "symbol", "value": "Audit.redact()"}] + self.assertNotEqual( + build_contract_identity(auth).contract_key, + build_contract_identity(unrelated).contract_key, + ) + + def test_severity_is_excluded_and_opposite_behavior_is_distinct(self) -> None: + high = build_contract_identity(self.payload()) + critical_payload = self.payload() + critical_payload["severity"] = "critical" + critical = build_contract_identity(critical_payload) + self.assertEqual(high.contract_key, critical.contract_key) + self.assertEqual(high.behavior_digest, critical.behavior_digest) + self.assertEqual(high.fingerprint_v2, critical.fingerprint_v2) + + opposite_payload = self.payload() + opposite_payload["required_behavior"] = [["return", "success"]] + opposite = build_contract_identity(opposite_payload) + self.assertEqual(high.contract_key, opposite.contract_key) + self.assertNotEqual(high.behavior_digest, opposite.behavior_digest) + + def test_limits_reject_instead_of_truncating_and_keep_long_tails(self) -> None: + exact = self.payload() + exact["anchors"] = [{"kind": "identifier", "value": "a" * MAX_ANCHOR_BYTES}] + self.assertEqual( + build_contract_identity(exact).anchors[0].value, + "a" * MAX_ANCHOR_BYTES, + ) + oversize = copy.deepcopy(exact) + oversize["anchors"][0]["value"] += "a" + with self.assertRaises(IdentityValidationError): + build_contract_identity(oversize) + + first = self.payload() + first["predicates"] = [["p" * 240, "tail_alpha"]] + second = copy.deepcopy(first) + second["predicates"] = [["p" * 240, "tail_beta"]] + self.assertNotEqual( + build_contract_identity(first).contract_key, + build_contract_identity(second).contract_key, + ) + first["predicates"] = [["p" * (MAX_TOKEN_BYTES + 1)]] + with self.assertRaises(IdentityValidationError): + build_contract_identity(first) + + def test_secret_literals_become_typed_placeholders_before_hashing(self) -> None: + payload = self.payload() + payload["required_behavior"] = [["token=secret-value"]] + identity = build_contract_identity(payload) + record = identity.as_record() + serialized = canonical_json(identity) + self.assertIn("", serialized) + self.assertNotIn("secret-value", serialized) + self.assertEqual(verify_persisted_identity(record), identity) + + payload["required_behavior"] = [[""]] + with self.assertRaises(IdentityValidationError): + build_contract_identity(payload) + record["required_behavior"] = [["token=secret-value"]] + with self.assertRaises(IdentityValidationError): + verify_persisted_identity(record) + + def test_schema_fields_category_unicode_and_controls_are_strict(self) -> None: + for mutate in ( + lambda value: value.pop("schema"), + lambda value: value.update({"description": "raw prose"}), + lambda value: value["scope"].update({"category": "unknown"}), + ): + payload = self.payload() + mutate(payload) + with self.assertRaises(IdentityValidationError): + build_contract_identity(payload) + + payload = self.payload() + payload["anchors"] = [{"kind": "identifier", "value": "Cafe\u0301"}] + self.assertEqual(build_contract_identity(payload).anchors[0].value, "Café") + payload["anchors"] = [{"kind": "identifier", "value": "bad\nanchor"}] + with self.assertRaises(IdentityValidationError): + build_contract_identity(payload) + + def test_repo_and_relative_path_boundaries(self) -> None: + for repo in ("QuantStrategyLab/AIAuditBridge", "owner/repo.name"): + payload = self.payload() + payload["scope"]["repo"] = repo + self.assertEqual(build_contract_identity(payload).scope.repo, repo) + for repo in ("owner", "/owner/repo", "owner/", "owner//repo", "owner/repo/extra"): + payload = self.payload() + payload["scope"]["repo"] = repo + with self.assertRaises(IdentityValidationError): + build_contract_identity(payload) + + payload = self.payload() + payload["scope"]["file"] = "src/nested/review.py" + payload["evidence"]["file"] = "src/nested/review.py" + self.assertEqual(build_contract_identity(payload).scope.file, "src/nested/review.py") + for path in ("/abs/review.py", "../escape.py", "a//b.py", "a\\b.py", "./review.py"): + invalid = self.payload() + invalid["scope"]["file"] = path + invalid["evidence"]["file"] = path + with self.assertRaises(IdentityValidationError): + build_contract_identity(invalid) + + mismatch = self.payload() + mismatch["evidence"]["file"] = "service/other.py" + with self.assertRaises(IdentityValidationError): + build_contract_identity(mismatch) + secret_scope = self.payload() + secret_scope["scope"]["repo"] = "owner/token=secret-value" + with self.assertRaises(IdentityValidationError): + build_contract_identity(secret_scope) + secret_ref = self.payload() + secret_ref["evidence"]["location_or_hunk_digest"] = "token=secret-value" + with self.assertRaises(IdentityValidationError): + build_contract_identity(secret_ref) + + def test_digest_tamper_order_and_evidence_binding_are_explicit(self) -> None: + identity = build_contract_identity(self.payload()) + tampered = identity.as_record() + tampered["behavior_digest"] = "0" * 64 + with self.assertRaises(IdentityValidationError): + verify_persisted_identity(tampered) + + reversed_payload = self.payload() + reversed_payload["ordering_constraints"] = [["dispatch", "before", "validate"]] + reversed_identity = build_contract_identity(reversed_payload) + self.assertEqual(identity.contract_key, reversed_identity.contract_key) + self.assertNotEqual(identity.behavior_digest, reversed_identity.behavior_digest) + + evidence_change = self.payload() + evidence_change["evidence"]["head_sha"] = "feedface" + evidence_change["evidence"]["diff_digest"] = "b" * 64 + rebound = build_contract_identity(evidence_change) + self.assertEqual(identity.fingerprint_v2, rebound.fingerprint_v2) + self.assertNotEqual( + json.dumps(identity.evidence.as_dict(), sort_keys=True), + json.dumps(rebound.evidence.as_dict(), sort_keys=True), + ) + + +if __name__ == "__main__": + unittest.main() From 7ff41a324e9d79b96554f9e75b5a493782246cb0 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 13 Jul 2026 03:47:43 +0800 Subject: [PATCH 2/4] fix: harden canonical identity input boundaries Co-Authored-By: Codex --- scripts/contract_identity.py | 31 ++++++++++++++++++++++++++----- tests/test_contract_identity.py | 31 ++++++++++++++++++++++++++++++- 2 files changed, 56 insertions(+), 6 deletions(-) diff --git a/scripts/contract_identity.py b/scripts/contract_identity.py index a75d53c..8121975 100644 --- a/scripts/contract_identity.py +++ b/scripts/contract_identity.py @@ -22,6 +22,8 @@ ) MAX_REPO_BYTES = 200 +MAX_OWNER_LENGTH = 39 +MAX_REPOSITORY_NAME_LENGTH = 100 MAX_FILE_BYTES = 512 MAX_ANCHOR_BYTES = 256 MAX_TOKEN_BYTES = 256 @@ -50,6 +52,9 @@ re.UNICODE, ) _PLACEHOLDER_RE = re.compile(r"") +_SAFE_CREDENTIAL_STATES = frozenset( + {"absent", "disabled", "enabled", "forbidden", "invalid", "missing", "none", "optional", "present", "redacted", "required", "valid"} +) _SECRET_PATTERNS = ( ("PRIVATE_KEY", re.compile(r"-----BEGIN [^-]+-----.*?-----END [^-]+-----", re.I | re.S)), ("AWS", re.compile(r"\b(?:AKIA|ASIA|AIDA|AROA)[A-Z0-9]{16}\b")), @@ -59,9 +64,9 @@ ("DSN", re.compile(r"\b[a-z][a-z0-9+.-]*://[^\s/@:]+:[^\s/@]+@[^\s]+", re.I)), ("BEARER", re.compile(r"(?i)\bbearer\s+\S+")), ("CREDENTIAL", re.compile( - r"(?i)(?token|secret|password|api[ _-]?key|authorization|cookie|aws[_ -]?secret[_ -]?access[_ -]?key)\b\s*[:=]\s*(?P\S+)" )), - ("API_KEY", re.compile(r"\b(?:gh[pousr]_|sk-)[A-Za-z0-9_-]{20,}\b")), + ("API_KEY", re.compile(r"\b(?:gh[pousr]_|github_pat_|sk-)[A-Za-z0-9_-]{20,}\b")), ) @@ -151,12 +156,14 @@ def redact(self, value: str, field: str) -> str: text = value for secret_type, pattern in _SECRET_PATTERNS: def replacement(_match: re.Match[str], kind: str = secret_type) -> str: + if kind == "CREDENTIAL" and _match.groupdict().get("value", "").lower() in _SAFE_CREDENTIAL_STATES: + return _match.group(0) self.counts[kind] += 1 return f"" text = pattern.sub(replacement, text) - malformed = re.search(r"]*>", text) - if malformed and not _PLACEHOLDER_RE.fullmatch(malformed.group(0)): + remainder = _PLACEHOLDER_RE.sub("", text) + if " normalized = unicodedata.normalize("NFC", value) if not normalized or any(unicodedata.category(char).startswith("C") for char in normalized): raise IdentityValidationError(f"{field} is empty or contains control characters") + if len(normalized.encode("utf-8")) > max_bytes: + raise IdentityValidationError(f"{field} exceeds {max_bytes} bytes") normalized = redactor.redact(normalized, field) if len(normalized.encode("utf-8")) > max_bytes: raise IdentityValidationError(f"{field} exceeds {max_bytes} bytes") @@ -181,7 +190,19 @@ def _text(value: Any, field: str, max_bytes: int, redactor: _SecretRedactor) -> def _repo(value: Any, field: str, redactor: _SecretRedactor) -> str: repo = _text(value, field, MAX_REPO_BYTES, redactor) - if not re.fullmatch(r"[A-Za-z0-9._-]+/[A-Za-z0-9._-]+", repo): + parts = repo.split("/") + if len(parts) != 2: + raise IdentityValidationError(f"{field} must be owner/name") + owner, name = parts + owner_valid = re.fullmatch(r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,37}[A-Za-z0-9])?", owner) + name_valid = re.fullmatch(r"[A-Za-z0-9._-]+", name) + if ( + not owner_valid + or not name_valid + or len(owner) > MAX_OWNER_LENGTH + or len(name) > MAX_REPOSITORY_NAME_LENGTH + or name in {".", ".."} + ): raise IdentityValidationError(f"{field} must be owner/name") return repo diff --git a/tests/test_contract_identity.py b/tests/test_contract_identity.py index aa528ec..ed7745b 100644 --- a/tests/test_contract_identity.py +++ b/tests/test_contract_identity.py @@ -130,6 +130,32 @@ def test_secret_literals_become_typed_placeholders_before_hashing(self) -> None: record["required_behavior"] = [["token=secret-value"]] with self.assertRaises(IdentityValidationError): verify_persisted_identity(record) + for malformed in (" ", " None: + required = self.payload() + required["required_behavior"] = [["authorization=required"]] + forbidden = copy.deepcopy(required) + forbidden["required_behavior"] = [["authorization=forbidden"]] + self.assertNotEqual( + build_contract_identity(required).behavior_digest, + build_contract_identity(forbidden).behavior_digest, + ) + secrets = self.payload() + secrets["required_behavior"] = [["github_pat_" + "x" * 30], ["aws_secret_access_key=secret-value"]] + identity = build_contract_identity(secrets) + serialized = canonical_json(identity) + self.assertNotIn("github_pat_", serialized) + self.assertNotIn("secret-value", serialized) def test_schema_fields_category_unicode_and_controls_are_strict(self) -> None: for mutate in ( @@ -154,7 +180,10 @@ def test_repo_and_relative_path_boundaries(self) -> None: payload = self.payload() payload["scope"]["repo"] = repo self.assertEqual(build_contract_identity(payload).scope.repo, repo) - for repo in ("owner", "/owner/repo", "owner/", "owner//repo", "owner/repo/extra"): + for repo in ( + "owner", "/owner/repo", "owner/", "owner//repo", "owner/repo/extra", + "owner_name/repo", "-owner/repo", "owner-/repo", "./repo", "owner/.", "owner/..", + ): payload = self.payload() payload["scope"]["repo"] = repo with self.assertRaises(IdentityValidationError): From fb4d2384cdf1382b6b23bb1301a4c9b617d3cdc0 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 13 Jul 2026 03:49:44 +0800 Subject: [PATCH 3/4] fix: preserve safe credential state replay Co-Authored-By: Codex --- scripts/contract_identity.py | 17 ++++++++++++++--- tests/test_contract_identity.py | 10 +++++++++- 2 files changed, 23 insertions(+), 4 deletions(-) diff --git a/scripts/contract_identity.py b/scripts/contract_identity.py index 8121975..0dc2f9c 100644 --- a/scripts/contract_identity.py +++ b/scripts/contract_identity.py @@ -148,15 +148,26 @@ def __init__(self, *, allow_placeholders: bool) -> None: self.allow_placeholders = allow_placeholders self.counts: dict[str, int] = defaultdict(int) + @staticmethod + def _safe_credential(match: re.Match[str]) -> bool: + return match.groupdict().get("value", "").lower() in _SAFE_CREDENTIAL_STATES + + def _has_unredacted_secret(self, value: str) -> bool: + for secret_type, pattern in _SECRET_PATTERNS: + for match in pattern.finditer(value): + if secret_type != "CREDENTIAL" or not self._safe_credential(match): + return True + return False + def redact(self, value: str, field: str) -> str: if " str: - if kind == "CREDENTIAL" and _match.groupdict().get("value", "").lower() in _SAFE_CREDENTIAL_STATES: + if kind == "CREDENTIAL" and self._safe_credential(_match): return _match.group(0) self.counts[kind] += 1 return f"" @@ -194,7 +205,7 @@ def _repo(value: Any, field: str, redactor: _SecretRedactor) -> str: if len(parts) != 2: raise IdentityValidationError(f"{field} must be owner/name") owner, name = parts - owner_valid = re.fullmatch(r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,37}[A-Za-z0-9])?", owner) + owner_valid = re.fullmatch(r"(?!.*--)[A-Za-z0-9](?:[A-Za-z0-9-]{0,37}[A-Za-z0-9])?", owner) name_valid = re.fullmatch(r"[A-Za-z0-9._-]+", name) if ( not owner_valid diff --git a/tests/test_contract_identity.py b/tests/test_contract_identity.py index ed7745b..37507a3 100644 --- a/tests/test_contract_identity.py +++ b/tests/test_contract_identity.py @@ -146,6 +146,14 @@ def test_credential_states_are_not_collapsed_and_common_literals_are_redacted(se required["required_behavior"] = [["authorization=required"]] forbidden = copy.deepcopy(required) forbidden["required_behavior"] = [["authorization=forbidden"]] + self.assertEqual( + verify_persisted_identity(build_contract_identity(required).as_record()), + build_contract_identity(required), + ) + self.assertEqual( + verify_persisted_identity(build_contract_identity(forbidden).as_record()), + build_contract_identity(forbidden), + ) self.assertNotEqual( build_contract_identity(required).behavior_digest, build_contract_identity(forbidden).behavior_digest, @@ -182,7 +190,7 @@ def test_repo_and_relative_path_boundaries(self) -> None: self.assertEqual(build_contract_identity(payload).scope.repo, repo) for repo in ( "owner", "/owner/repo", "owner/", "owner//repo", "owner/repo/extra", - "owner_name/repo", "-owner/repo", "owner-/repo", "./repo", "owner/.", "owner/..", + "owner_name/repo", "-owner/repo", "owner-/repo", "owner--name/repo", "./repo", "owner/.", "owner/..", ): payload = self.payload() payload["scope"]["repo"] = repo From 950274a61a4b6d3dea89e1d19e016c71408fa2ad Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 13 Jul 2026 04:13:39 +0800 Subject: [PATCH 4/4] fix: bound clause and secret input processing Co-Authored-By: Codex --- scripts/contract_identity.py | 46 ++++++++++++++++++++++----------- tests/test_contract_identity.py | 45 ++++++++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 15 deletions(-) diff --git a/scripts/contract_identity.py b/scripts/contract_identity.py index 0dc2f9c..e93f039 100644 --- a/scripts/contract_identity.py +++ b/scripts/contract_identity.py @@ -30,6 +30,7 @@ MAX_EVIDENCE_BYTES = 256 MAX_ITEMS = 32 MAX_TOKENS_PER_CLAUSE = 128 +MAX_CLAUSE_BYTES = MAX_TOKEN_BYTES * MAX_TOKENS_PER_CLAUSE MAX_CANONICAL_BYTES = 16_384 _TOP_REQUIRED = frozenset( @@ -180,7 +181,10 @@ def replacement(_match: re.Match[str], kind: str = secret_type) -> str: def _exact_fields(value: Any, required: frozenset[str], optional: frozenset[str], field: str) -> dict[str, Any]: - if not isinstance(value, dict) or set(value) != required | (set(value) & optional): + if not isinstance(value, dict) or len(value) > len(required) + len(optional): + raise IdentityValidationError(f"{field} has missing or extra fields") + keys = set(value) + if not required.issubset(keys) or not keys.difference(required).issubset(optional): raise IdentityValidationError(f"{field} has missing or extra fields") return value @@ -188,6 +192,11 @@ def _exact_fields(value: Any, required: frozenset[str], optional: frozenset[str] def _text(value: Any, field: str, max_bytes: int, redactor: _SecretRedactor) -> str: if not isinstance(value, str): raise IdentityValidationError(f"{field} must be a string") + if len(value) > max_bytes: + raise IdentityValidationError(f"{field} exceeds {max_bytes} bytes") + raw_bytes = value.encode("utf-8") + if len(raw_bytes) > max_bytes: + raise IdentityValidationError(f"{field} exceeds {max_bytes} bytes") normalized = unicodedata.normalize("NFC", value) if not normalized or any(unicodedata.category(char).startswith("C") for char in normalized): raise IdentityValidationError(f"{field} is empty or contains control characters") @@ -238,6 +247,15 @@ def _reference(value: Any, field: str, redactor: _SecretRedactor) -> str: return reference +def _enum(value: Any, field: str, allowed: frozenset[str]) -> str: + if not isinstance(value, str) or len(value) > 64: + raise IdentityValidationError(f"{field} is not a bounded string") + normalized = value.lower() + if normalized not in allowed: + raise IdentityValidationError(f"{field} is not controlled") + return normalized + + def _clauses( value: Any, field: str, redactor: _SecretRedactor, *, allow_empty: bool ) -> tuple[tuple[str, ...], ...]: @@ -245,12 +263,16 @@ def _clauses( raise IdentityValidationError(f"{field} must be a bounded list") clauses: list[tuple[str, ...]] = [] for index, clause in enumerate(value): - if not isinstance(clause, list) or not clause: + if not isinstance(clause, list) or not clause or len(clause) > MAX_TOKENS_PER_CLAUSE: raise IdentityValidationError(f"{field}[{index}] must be a non-empty token list") - tokens: list[str] = [] - for raw_token in clause: - token = _text(raw_token, f"{field}[{index}]", MAX_TOKEN_BYTES, redactor) - tokens.extend(_TOKEN_RE.findall(token)) + if any(not isinstance(raw_token, str) or len(raw_token) > MAX_TOKEN_BYTES for raw_token in clause): + raise IdentityValidationError(f"{field}[{index}] contains an oversized token") + if sum(len(raw_token) for raw_token in clause) > MAX_CLAUSE_BYTES: + raise IdentityValidationError(f"{field}[{index}] exceeds aggregate size") + joined = _text(" ".join(clause), f"{field}[{index}]", MAX_CLAUSE_BYTES, redactor) + tokens = _TOKEN_RE.findall(joined) + if any(len(token.encode("utf-8")) > MAX_TOKEN_BYTES for token in tokens): + raise IdentityValidationError(f"{field}[{index}] contains an oversized token") if not tokens or len(tokens) > MAX_TOKENS_PER_CLAUSE: raise IdentityValidationError(f"{field}[{index}] has invalid token count") clauses.append(tuple(tokens)) @@ -272,9 +294,7 @@ def build_contract_identity(payload: dict[str, Any], *, _allow_placeholders: boo redactor = _SecretRedactor(allow_placeholders=_allow_placeholders) raw_scope = _exact_fields(top["scope"], frozenset({"repo", "file", "category"}), frozenset(), "scope") - category = str(raw_scope["category"]).lower() if isinstance(raw_scope["category"], str) else "" - if category not in ALLOWED_CATEGORIES: - raise IdentityValidationError("scope.category is not controlled") + category = _enum(raw_scope["category"], "scope.category", ALLOWED_CATEGORIES) scope = Scope( repo=_repo(raw_scope["repo"], "scope.repo", redactor), file=_relative_path(raw_scope["file"], "scope.file", redactor), @@ -287,9 +307,7 @@ def build_contract_identity(payload: dict[str, Any], *, _allow_placeholders: boo anchors: list[Anchor] = [] for index, raw_anchor in enumerate(raw_anchors): anchor = _exact_fields(raw_anchor, frozenset({"kind", "value"}), frozenset(), f"anchors[{index}]") - kind = str(anchor["kind"]).lower() if isinstance(anchor["kind"], str) else "" - if kind not in ALLOWED_ANCHOR_KINDS: - raise IdentityValidationError(f"anchors[{index}].kind is not controlled") + kind = _enum(anchor["kind"], f"anchors[{index}].kind", ALLOWED_ANCHOR_KINDS) anchors.append(Anchor(kind, _text(anchor["value"], f"anchors[{index}].value", MAX_ANCHOR_BYTES, redactor))) predicates = _clauses(top["predicates"], "predicates", redactor, allow_empty=False) @@ -327,9 +345,7 @@ def build_contract_identity(payload: dict[str, Any], *, _allow_placeholders: boo raise IdentityValidationError("evidence.file must match scope.file") severity = top.get("severity") if severity is not None: - severity = str(severity).lower() if isinstance(severity, str) else "" - if severity not in ALLOWED_SEVERITIES: - raise IdentityValidationError("severity is not controlled") + severity = _enum(severity, "severity", ALLOWED_SEVERITIES) contract_payload = { "schema": SCHEMA, diff --git a/tests/test_contract_identity.py b/tests/test_contract_identity.py index 37507a3..b37700c 100644 --- a/tests/test_contract_identity.py +++ b/tests/test_contract_identity.py @@ -3,11 +3,17 @@ import copy import json import unittest +from unittest.mock import patch from scripts.contract_identity import ( MAX_ANCHOR_BYTES, + MAX_CLAUSE_BYTES, + MAX_ITEMS, MAX_TOKEN_BYTES, + MAX_TOKENS_PER_CLAUSE, IdentityValidationError, + _SecretRedactor, + _text, build_contract_identity, canonical_json, operators, @@ -165,6 +171,22 @@ def test_credential_states_are_not_collapsed_and_common_literals_are_redacted(se self.assertNotIn("github_pat_", serialized) self.assertNotIn("secret-value", serialized) + for state in ("required", "forbidden"): + split = self.payload() + split["required_behavior"] = [["authorization", "=", state]] + split_identity = build_contract_identity(split) + self.assertEqual(verify_persisted_identity(split_identity.as_record()), split_identity) + for clauses in ( + [["token", "=", "supersecret"]], + [["github_pat_" + "x" * 30], ["aws_secret_access_key", "=", "secret-value"]], + ): + split = self.payload() + split["required_behavior"] = clauses + serialized = canonical_json(build_contract_identity(split)) + self.assertNotIn("supersecret", serialized) + self.assertNotIn("github_pat_", serialized) + self.assertNotIn("secret-value", serialized) + def test_schema_fields_category_unicode_and_controls_are_strict(self) -> None: for mutate in ( lambda value: value.pop("schema"), @@ -221,6 +243,29 @@ def test_repo_and_relative_path_boundaries(self) -> None: with self.assertRaises(IdentityValidationError): build_contract_identity(secret_ref) + def test_untrusted_shape_and_normalize_bounds_fail_closed(self) -> None: + anchors = self.payload() + anchors["anchors"] = anchors["anchors"] * (MAX_ITEMS + 1) + with self.assertRaises(IdentityValidationError): + build_contract_identity(anchors) + clauses = self.payload() + clauses["predicates"] = [["x"] * (MAX_TOKENS_PER_CLAUSE + 1)] + with self.assertRaises(IdentityValidationError): + build_contract_identity(clauses) + aggregate = self.payload() + aggregate["predicates"] = [["x" * MAX_TOKEN_BYTES] * MAX_TOKENS_PER_CLAUSE] + self.assertGreater(sum(map(len, aggregate["predicates"][0])), MAX_CLAUSE_BYTES - 1) + with self.assertRaises(IdentityValidationError): + build_contract_identity(aggregate) + oversized_dict = self.payload() + oversized_dict.update({f"extra_{index}": index for index in range(1000)}) + with self.assertRaises(IdentityValidationError): + build_contract_identity(oversized_dict) + with patch("scripts.contract_identity.unicodedata.normalize") as normalize: + with self.assertRaises(IdentityValidationError): + _text("x" * 2_000_000, "oversized", MAX_TOKEN_BYTES, _SecretRedactor(allow_placeholders=False)) + normalize.assert_not_called() + def test_digest_tamper_order_and_evidence_binding_are_explicit(self) -> None: identity = build_contract_identity(self.payload()) tampered = identity.as_record()