From 7c86e69525ce98aecc24de535e50d9ba113781c6 Mon Sep 17 00:00:00 2001 From: Philip Z Date: Thu, 1 Oct 2026 03:18:13 +0800 Subject: [PATCH] feat: add portable opaque entity references (#37) --- README.md | 38 ++- examples/opaque-entity-reference/README.md | 6 + examples/opaque-entity-reference/main.py | 26 ++ pyproject.toml | 1 + scripts/verify-examples.sh | 3 +- src/teaql/runtime/__init__.py | 11 +- src/teaql/runtime/context.py | 58 ++++- src/teaql/runtime/entity_reference.py | 262 +++++++++++++++++++++ tests/runtime/test_entity_reference.py | 70 ++++++ 9 files changed, 464 insertions(+), 11 deletions(-) create mode 100644 examples/opaque-entity-reference/README.md create mode 100644 examples/opaque-entity-reference/main.py create mode 100644 src/teaql/runtime/entity_reference.py create mode 100644 tests/runtime/test_entity_reference.py diff --git a/README.md b/README.md index e85c102..c53fa11 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,7 @@ and evidence-based verification as the generator and runtimes evolve. * **Python**: 3.10+ (Recommended 3.12+) * **Testing**: `pytest` 7.4+ -* **Dependencies**: `pydantic` >= 2.0, `aiosqlite`, `aiomysql`, `asyncpg` +* **Dependencies**: `pydantic` >= 2.0, `aiosqlite`, `aiomysql`, `asyncpg`, `cryptography` >= 42 ## 2. Tests Performed @@ -96,19 +96,41 @@ non-empty comment/purpose and audited mutations retain their audit reason. Runtime logging should keep parameterized SQL and intent separate from any restricted value-bearing diagnostic output. -Portable `UserContext` opaque entity references are not implemented in the -Python runtime yet. Until that capability is added, applications must not -invent a Python-specific token format or serialize internal ID/version pairs as -if they were the TeaQL portable contract. A Python TFP client may carry an -opaque token issued by a trusted Java, Rust, Go, or .NET backend, but it must -not decode, rewrite, or mint that token. +Python implements the portable `tqr1` tuple codec shared with Go and .NET. It +encrypts and authenticates entity type, internal ID, optimistic version, +issued/expiry time, and purpose with AES-256-GCM. Key rings permit rotation; +encoding always uses the active key while decoding can accept retained keys. -The planned wire format, fail-closed behavior, shared golden vector, and exact +```python +import os +from datetime import timedelta +from teaql.runtime import AeadEntityReferenceCodec, UserContext + +codec = AeadEntityReferenceCodec(2, { + 1: bytes.fromhex(os.environ["TEAQL_ENTITY_REFERENCE_KEY_V1_HEX"]), + 2: bytes.fromhex(os.environ["TEAQL_ENTITY_REFERENCE_KEY_V2_HEX"]), +}) +context = UserContext().with_entity_reference_codec(codec) +token = context.encode_entity_reference( + "OrderItem", 42, 7, "edit-order", timedelta(minutes=30) +) +claims = context.decode_entity_reference(token, "OrderItem", "edit-order") +``` + +Without a configured codec the runtime fails closed. Local debugging can use +the canonical long `TEAQL_UNSAFE_RAW_ENTITY_REFERENCES` acknowledgement, which +emits visibly distinct `tqr0` references. It must not be enabled in production. + +The wire format, fail-closed behavior, shared golden vector, and exact development-only acknowledgement are maintained in the canonical [opaque entity reference contract](https://github.com/teaql/teaql-conformance/blob/main/design/opaque-entity-references.md). Opaque tokens never replace the backend's authorization, tenant, ownership, role, or optimistic-version checks. +The repeatable [`examples/opaque-entity-reference`](examples/opaque-entity-reference) +example proves the exact cross-language golden vector and purpose-substitution +rejection. + ### Mutation Policy installation Policy implementations are installed from trusted application startup through diff --git a/examples/opaque-entity-reference/README.md b/examples/opaque-entity-reference/README.md new file mode 100644 index 0000000..a966311 --- /dev/null +++ b/examples/opaque-entity-reference/README.md @@ -0,0 +1,6 @@ +# Opaque entity reference + +This example proves that Python emits the same portable `tqr1` golden vector as +Go and .NET, decodes it through `UserContext`, and rejects purpose +substitution. Decoding authenticates identity claims; it does not replace the +application's ownership, permission, or optimistic-version checks. diff --git a/examples/opaque-entity-reference/main.py b/examples/opaque-entity-reference/main.py new file mode 100644 index 0000000..69e6296 --- /dev/null +++ b/examples/opaque-entity-reference/main.py @@ -0,0 +1,26 @@ +from datetime import datetime, timedelta, timezone + +from teaql.runtime import AeadEntityReferenceCodec, EntityReferenceTokenError, UserContext + + +GOLDEN = "tqr1.AAAAAjMzMzMzMzMzMzMzM3bKiZgRSQQhfIj2cBXRDZIloUGHWLBp8QrXL_aejwIXPFtvV_E71O7wbOXy3cvYo_SwxvuS-89x572T9CO_pDAY4tbjWCNv" +now = datetime(2026, 9, 26, 12, 0, tzinfo=timezone.utc) +codec = AeadEntityReferenceCodec( + 2, {1: bytes([0x11]) * 32, 2: bytes([0x22]) * 32} +).with_clock(lambda: now).with_nonce_source(lambda: bytes([0x33]) * 12) +context = UserContext().with_entity_reference_codec(codec) + +token = context.encode_entity_reference( + "OrderItem", 42, 7, "edit-order", timedelta(hours=1) +) +assert token == GOLDEN +claims = context.decode_entity_reference(token, "OrderItem", "edit-order") +assert (claims.id, claims.version, claims.key_version) == (42, 7, 2) + +try: + context.decode_entity_reference(token, "OrderItem", "view-order") + raise AssertionError("purpose substitution must fail") +except EntityReferenceTokenError as error: + assert error.code == "ENTITY_REFERENCE_INVALID" + +print("PASS: Python opaque entity reference") diff --git a/pyproject.toml b/pyproject.toml index ebb4fe5..a2c6ebb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,7 @@ dependencies = [ "aiomysql>=0.2.0", "asyncpg>=0.29.0", "httpx>=0.24.0", + "cryptography>=42.0.0", ] [project.optional-dependencies] diff --git a/scripts/verify-examples.sh b/scripts/verify-examples.sh index 103f0fc..0f0e578 100755 --- a/scripts/verify-examples.sh +++ b/scripts/verify-examples.sh @@ -2,7 +2,7 @@ set -euo pipefail repo="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -expected=(business-clock conformance mutation-policy order-management query-policy school-management task_board) +expected=(business-clock conformance mutation-policy opaque-entity-reference order-management query-policy school-management task_board) mapfile -t actual < <(find "$repo/examples" -mindepth 1 -maxdepth 1 -type d -printf '%f\n' | sort) if [[ "${actual[*]}" != "${expected[*]}" ]]; then echo "example inventory changed; update scripts/verify-examples.sh: ${actual[*]}" >&2 @@ -27,6 +27,7 @@ PYTHONPATH="$repo/src" python -m unittest discover -s "$repo/examples/school-man PYTHONPATH="$repo/src" python "$repo/examples/mutation-policy/main.py" PYTHONPATH="$repo/src" python "$repo/examples/business-clock/main.py" PYTHONPATH="$repo/src" python "$repo/examples/query-policy/main.py" +PYTHONPATH="$repo/src" python "$repo/examples/opaque-entity-reference/main.py" order_management_tmp="$(mktemp -d)" task_board_tmp="$(mktemp -d)" trap 'rm -rf "$order_management_tmp" "$task_board_tmp"' EXIT diff --git a/src/teaql/runtime/__init__.py b/src/teaql/runtime/__init__.py index a3a9a1c..11f61dc 100644 --- a/src/teaql/runtime/__init__.py +++ b/src/teaql/runtime/__init__.py @@ -11,6 +11,15 @@ from .store import DataStore from .audit import RawAuditEvent, SafeAuditEvent, MutationAuditKind from .business_clock import BusinessClock, FixedBusinessClock, SystemBusinessClock +from .entity_reference import ( + AeadEntityReferenceCodec, + EntityReferenceClaims, + EntityReferenceCodec, + EntityReferenceTokenError, + ENTITY_REFERENCE_AAD, + UNSAFE_RAW_ENTITY_REFERENCES_ACKNOWLEDGEMENT, + UNSAFE_RAW_ENTITY_REFERENCES_ENVIRONMENT, +) from .mutation_policy import ( MISSING_APPROVAL, MISSING_POLICY, @@ -35,7 +44,7 @@ from .wire_fields import NormalizedWireInput, WireEntityMetadata, WireFieldMetadata, WireInputError, create_wire_entity_metadata, encode_wire_output, normalize_wire_input, retain_submitted_paths from teaql.core.entity import EntityKey, EntityChangeSet, EntityRoot -__all__ = ["WireFieldMetadata", "WireEntityMetadata", "NormalizedWireInput", "WireInputError", "create_wire_entity_metadata", "normalize_wire_input", "encode_wire_output", "retain_submitted_paths", "EntityKey", "EntityChangeSet", "EntityRoot", "ContextEntityRef", "ContextRootError", "CheckException", "CheckResult", "I18nCatalog", "JsonFieldNamingProfile", "Locale", "ObjectLocation", "UnsupportedLocaleError", "UserContext", "TeaqlRuntime", "SqlLogEntry", "SqlLogOperation", "DiagnosticSqlLogSink", "TextDiagnosticSqlLogSink", "ServiceRuntimeFromEnv", "RuntimeModule", "DataStore", "RawAuditEvent", "SafeAuditEvent", "MutationAuditKind", "BusinessClock", "FixedBusinessClock", "SystemBusinessClock", "ContextTools", "ExecutableHttpTool", "HTTP_TOOL", "HttpIntentPhase", "HttpTool", "HttpToolProvider", "ToolDeniedError", "ToolError", "ToolPolicy", "ToolRisk", "Tools", "ToolToken", "ToolUnavailableError"] +__all__ = ["WireFieldMetadata", "WireEntityMetadata", "NormalizedWireInput", "WireInputError", "create_wire_entity_metadata", "normalize_wire_input", "encode_wire_output", "retain_submitted_paths", "EntityKey", "EntityChangeSet", "EntityRoot", "ContextEntityRef", "ContextRootError", "CheckException", "CheckResult", "I18nCatalog", "JsonFieldNamingProfile", "Locale", "ObjectLocation", "UnsupportedLocaleError", "UserContext", "TeaqlRuntime", "SqlLogEntry", "SqlLogOperation", "DiagnosticSqlLogSink", "TextDiagnosticSqlLogSink", "ServiceRuntimeFromEnv", "RuntimeModule", "DataStore", "RawAuditEvent", "SafeAuditEvent", "MutationAuditKind", "BusinessClock", "FixedBusinessClock", "SystemBusinessClock", "AeadEntityReferenceCodec", "EntityReferenceClaims", "EntityReferenceCodec", "EntityReferenceTokenError", "ENTITY_REFERENCE_AAD", "UNSAFE_RAW_ENTITY_REFERENCES_ACKNOWLEDGEMENT", "UNSAFE_RAW_ENTITY_REFERENCES_ENVIRONMENT", "ContextTools", "ExecutableHttpTool", "HTTP_TOOL", "HttpIntentPhase", "HttpTool", "HttpToolProvider", "ToolDeniedError", "ToolError", "ToolPolicy", "ToolRisk", "Tools", "ToolToken", "ToolUnavailableError"] __all__ += [ "MISSING_APPROVAL", "MISSING_POLICY", diff --git a/src/teaql/runtime/context.py b/src/teaql/runtime/context.py index 32ae04c..b5573f1 100644 --- a/src/teaql/runtime/context.py +++ b/src/teaql/runtime/context.py @@ -4,9 +4,19 @@ from copy import deepcopy from dataclasses import dataclass from array import array -from datetime import date, datetime +from datetime import date, datetime, timedelta, timezone from .business_clock import BusinessClock, SystemBusinessClock +from .entity_reference import ( + EntityReferenceClaims, + EntityReferenceCodec, + EntityReferenceTokenError, + decode_raw_entity_reference, + encode_raw_entity_reference, + raw_entity_references_enabled, + validate_decoded_reference, + validate_reference_request, +) TEntity = TypeVar("TEntity") @@ -110,6 +120,7 @@ def __init__(self): self._fix_evidence_last: List[FixEvidence] = [] self._checked_mutations = set() self._business_clock: BusinessClock = SystemBusinessClock() + self._entity_reference_codec: Optional[EntityReferenceCodec] = None from .mutation_policy import MutationPolicyRuntimeState self._mutation_policy = MutationPolicyRuntimeState() @@ -151,6 +162,51 @@ def require_active_root(self, expected_type: str) -> ContextEntityRef: raise ContextRootError("type_mismatch", expected_type, root) return root + def with_entity_reference_codec( + self, codec: EntityReferenceCodec + ) -> 'UserContext': + if codec is None: + raise TypeError("entity reference codec is required") + self._entity_reference_codec = codec + return self + + def encode_entity_reference( + self, + entity_type: str, + entity_id: int, + version: int, + purpose: str, + lifetime: timedelta, + ) -> str: + if self._entity_reference_codec is not None: + return self._entity_reference_codec.encode_entity_reference( + entity_type, entity_id, version, purpose, lifetime + ) + if not raw_entity_references_enabled(): + raise EntityReferenceTokenError("ENTITY_REFERENCE_CODEC_REQUIRED") + validate_reference_request(entity_type, entity_id, version, purpose, lifetime) + now = datetime.now(timezone.utc) + return encode_raw_entity_reference( + EntityReferenceClaims( + entity_type, entity_id, version, now, now + lifetime, purpose + ) + ) + + def decode_entity_reference( + self, token: str, expected_entity_type: str, purpose: str + ) -> EntityReferenceClaims: + if self._entity_reference_codec is not None: + return self._entity_reference_codec.decode_entity_reference( + token, expected_entity_type, purpose + ) + if not raw_entity_references_enabled(): + raise EntityReferenceTokenError("ENTITY_REFERENCE_CODEC_REQUIRED") + claims = decode_raw_entity_reference(token) + validate_decoded_reference( + claims, datetime.now(timezone.utc), expected_entity_type, purpose + ) + return claims + async def execute_graph_save(self, work): """Run one generated entity graph in one provider transaction.""" if self._graph_save_owner.get() is not None: diff --git a/src/teaql/runtime/entity_reference.py b/src/teaql/runtime/entity_reference.py new file mode 100644 index 0000000..ace5182 --- /dev/null +++ b/src/teaql/runtime/entity_reference.py @@ -0,0 +1,262 @@ +"""Opaque boundary references for internal entity identity tuples.""" + +from __future__ import annotations + +import base64 +import os +import struct +from dataclasses import dataclass, replace +from datetime import datetime, timedelta, timezone +from typing import Callable, Mapping, Protocol + +from cryptography.exceptions import InvalidTag +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + + +ENTITY_REFERENCE_AAD = b"teaql.entity-reference.v1" +ENTITY_REFERENCE_PREFIX = "tqr1." +UNSAFE_RAW_REFERENCE_PREFIX = "tqr0." +UNSAFE_RAW_ENTITY_REFERENCES_ENVIRONMENT = "TEAQL_UNSAFE_RAW_ENTITY_REFERENCES" +UNSAFE_RAW_ENTITY_REFERENCES_ACKNOWLEDGEMENT = ( + "I_UNDERSTAND_RAW_ENTITY_IDS_ARE_VISIBLE_FOR_LOCAL_DEVELOPMENT_ONLY" +) + + +class EntityReferenceTokenError(ValueError): + """Stable public failure without cryptographic-oracle details.""" + + def __init__(self, code: str): + self.code = code + super().__init__(code) + + +@dataclass(frozen=True) +class EntityReferenceClaims: + entity_type: str + id: int + version: int + issued_at: datetime + expires_at: datetime + purpose: str + key_version: int = 0 + + +class EntityReferenceCodec(Protocol): + def encode_entity_reference( + self, + entity_type: str, + entity_id: int, + version: int, + purpose: str, + lifetime: timedelta, + ) -> str: ... + + def decode_entity_reference( + self, token: str, expected_entity_type: str, purpose: str + ) -> EntityReferenceClaims: ... + + +def _invalid() -> EntityReferenceTokenError: + return EntityReferenceTokenError("ENTITY_REFERENCE_INVALID") + + +def _utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _write_text(value: str) -> bytes: + encoded = value.encode("utf-8") + if len(encoded) > 0xFFFF: + raise ValueError("entity reference text exceeds 65535 bytes") + return struct.pack(">H", len(encoded)) + encoded + + +def _read_text(payload: bytes, offset: int) -> tuple[str, int]: + if offset + 2 > len(payload): + raise _invalid() + length = struct.unpack_from(">H", payload, offset)[0] + offset += 2 + if offset + length > len(payload): + raise _invalid() + try: + return payload[offset : offset + length].decode("utf-8"), offset + length + except UnicodeDecodeError as error: + raise _invalid() from error + + +def encode_claims(claims: EntityReferenceClaims) -> bytes: + return b"".join( + ( + _write_text(claims.entity_type), + struct.pack( + ">Qqqq", + claims.id, + claims.version, + int(claims.issued_at.timestamp()), + int(claims.expires_at.timestamp()), + ), + _write_text(claims.purpose), + ) + ) + + +def decode_claims(payload: bytes) -> EntityReferenceClaims: + entity_type, offset = _read_text(payload, 0) + if offset + 32 > len(payload): + raise _invalid() + entity_id, version, issued_at, expires_at = struct.unpack_from(">Qqqq", payload, offset) + purpose, offset = _read_text(payload, offset + 32) + if offset != len(payload) or not entity_type or entity_id == 0: + raise _invalid() + try: + return EntityReferenceClaims( + entity_type=entity_type, + id=entity_id, + version=version, + issued_at=datetime.fromtimestamp(issued_at, timezone.utc), + expires_at=datetime.fromtimestamp(expires_at, timezone.utc), + purpose=purpose, + ) + except (OverflowError, OSError, ValueError) as error: + raise _invalid() from error + + +def _base64url_encode(value: bytes) -> str: + return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=") + + +def _base64url_decode(value: str) -> bytes: + try: + return base64.b64decode( + value + "=" * (-len(value) % 4), altchars=b"-_", validate=True + ) + except (ValueError, base64.binascii.Error) as error: + raise _invalid() from error + + +def validate_reference_request( + entity_type: str, entity_id: int, version: int, purpose: str, lifetime: timedelta +) -> None: + if ( + not isinstance(entity_type, str) + or not entity_type.strip() + or not isinstance(entity_id, int) + or isinstance(entity_id, bool) + or entity_id <= 0 + or entity_id > 0xFFFFFFFFFFFFFFFF + or not isinstance(version, int) + or isinstance(version, bool) + or version < -(1 << 63) + or version >= (1 << 63) + or not isinstance(purpose, str) + or not isinstance(lifetime, timedelta) + or lifetime <= timedelta(0) + ): + raise _invalid() + + +def validate_decoded_reference( + claims: EntityReferenceClaims, + now: datetime, + expected_entity_type: str, + purpose: str, +) -> None: + if ( + claims.expires_at <= now + or claims.issued_at > now + timedelta(minutes=1) + or claims.entity_type != expected_entity_type + or claims.purpose != purpose + ): + raise _invalid() + + +class AeadEntityReferenceCodec: + """Portable tuple-codec v1 using AES-256-GCM and a rotating key ring.""" + + def __init__(self, active_key_version: int, keys: Mapping[int, bytes]): + copied = {version: bytes(key) for version, key in keys.items()} + if active_key_version not in copied or len(copied[active_key_version]) != 32: + raise ValueError("active entity reference key must contain 32 bytes") + for version, key in copied.items(): + if version < 0 or version > 0xFFFFFFFF or len(key) != 32: + raise ValueError(f"entity reference key {version} must contain 32 bytes") + self._active_key_version = active_key_version + self._keys = copied + self._clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc) + self._nonce_source: Callable[[], bytes] = lambda: os.urandom(12) + + def with_clock(self, clock: Callable[[], datetime]) -> "AeadEntityReferenceCodec": + self._clock = clock + return self + + def with_nonce_source( + self, nonce_source: Callable[[], bytes] + ) -> "AeadEntityReferenceCodec": + self._nonce_source = nonce_source + return self + + def encode_entity_reference( + self, + entity_type: str, + entity_id: int, + version: int, + purpose: str, + lifetime: timedelta, + ) -> str: + validate_reference_request(entity_type, entity_id, version, purpose, lifetime) + now = _utc(self._clock()) + nonce = bytes(self._nonce_source()) + if len(nonce) != 12: + raise ValueError("entity reference nonce must contain 12 bytes") + claims = EntityReferenceClaims( + entity_type, entity_id, version, now, now + lifetime, purpose + ) + encrypted = AESGCM(self._keys[self._active_key_version]).encrypt( + nonce, encode_claims(claims), ENTITY_REFERENCE_AAD + ) + envelope = struct.pack(">I", self._active_key_version) + nonce + encrypted + return ENTITY_REFERENCE_PREFIX + _base64url_encode(envelope) + + def decode_entity_reference( + self, token: str, expected_entity_type: str, purpose: str + ) -> EntityReferenceClaims: + try: + if not isinstance(token, str) or not token.startswith(ENTITY_REFERENCE_PREFIX): + raise _invalid() + envelope = _base64url_decode(token[len(ENTITY_REFERENCE_PREFIX) :]) + if len(envelope) < 4 + 12 + 16: + raise _invalid() + key_version = struct.unpack_from(">I", envelope)[0] + key = self._keys.get(key_version) + if key is None: + raise _invalid() + plaintext = AESGCM(key).decrypt( + envelope[4:16], envelope[16:], ENTITY_REFERENCE_AAD + ) + claims = replace(decode_claims(plaintext), key_version=key_version) + validate_decoded_reference( + claims, _utc(self._clock()), expected_entity_type, purpose + ) + return claims + except EntityReferenceTokenError: + raise + except (InvalidTag, ValueError, TypeError, struct.error) as error: + raise _invalid() from error + + +def raw_entity_references_enabled() -> bool: + return os.environ.get(UNSAFE_RAW_ENTITY_REFERENCES_ENVIRONMENT) == ( + UNSAFE_RAW_ENTITY_REFERENCES_ACKNOWLEDGEMENT + ) + + +def encode_raw_entity_reference(claims: EntityReferenceClaims) -> str: + return UNSAFE_RAW_REFERENCE_PREFIX + _base64url_encode(encode_claims(claims)) + + +def decode_raw_entity_reference(token: str) -> EntityReferenceClaims: + if not token.startswith(UNSAFE_RAW_REFERENCE_PREFIX): + raise EntityReferenceTokenError("ENTITY_REFERENCE_CODEC_REQUIRED") + return decode_claims(_base64url_decode(token[len(UNSAFE_RAW_REFERENCE_PREFIX) :])) diff --git a/tests/runtime/test_entity_reference.py b/tests/runtime/test_entity_reference.py new file mode 100644 index 0000000..a5b0c5b --- /dev/null +++ b/tests/runtime/test_entity_reference.py @@ -0,0 +1,70 @@ +from datetime import datetime, timedelta, timezone + +import pytest + +from teaql.runtime import ( + AeadEntityReferenceCodec, + EntityReferenceTokenError, + UNSAFE_RAW_ENTITY_REFERENCES_ACKNOWLEDGEMENT, + UNSAFE_RAW_ENTITY_REFERENCES_ENVIRONMENT, + UserContext, +) + + +GOLDEN = "tqr1.AAAAAjMzMzMzMzMzMzMzM3bKiZgRSQQhfIj2cBXRDZIloUGHWLBp8QrXL_aejwIXPFtvV_E71O7wbOXy3cvYo_SwxvuS-89x572T9CO_pDAY4tbjWCNv" + + +def test_portable_codec_is_opaque_bound_rotatable_and_expiring(): + now = datetime(2026, 9, 26, 12, 0, tzinfo=timezone.utc) + codec = AeadEntityReferenceCodec( + 2, {1: bytes([0x11]) * 32, 2: bytes([0x22]) * 32} + ).with_clock(lambda: now).with_nonce_source(lambda: bytes([0x33]) * 12) + + token = codec.encode_entity_reference( + "OrderItem", 42, 7, "edit-order", timedelta(hours=1) + ) + assert token == GOLDEN + assert "OrderItem" not in token + claims = codec.decode_entity_reference(token, "OrderItem", "edit-order") + assert (claims.id, claims.version, claims.key_version) == (42, 7, 2) + + old_codec = AeadEntityReferenceCodec( + 1, {1: bytes([0x11]) * 32} + ).with_clock(lambda: now).with_nonce_source(lambda: bytes([0x44]) * 12) + old_token = old_codec.encode_entity_reference( + "OrderItem", 42, 7, "edit-order", timedelta(hours=1) + ) + assert codec.decode_entity_reference( + old_token, "OrderItem", "edit-order" + ).key_version == 1 + + for entity_type, purpose, candidate in ( + ("InvoiceItem", "edit-order", token), + ("OrderItem", "other-purpose", token), + ("OrderItem", "edit-order", token[:-1] + "A"), + ): + with pytest.raises(EntityReferenceTokenError, match="ENTITY_REFERENCE_INVALID"): + codec.decode_entity_reference(candidate, entity_type, purpose) + + codec.with_clock(lambda: now + timedelta(hours=2)) + with pytest.raises(EntityReferenceTokenError, match="ENTITY_REFERENCE_INVALID"): + codec.decode_entity_reference(token, "OrderItem", "edit-order") + + +def test_raw_references_require_exact_development_acknowledgement(monkeypatch): + context = UserContext() + monkeypatch.delenv(UNSAFE_RAW_ENTITY_REFERENCES_ENVIRONMENT, raising=False) + with pytest.raises(EntityReferenceTokenError, match="ENTITY_REFERENCE_CODEC_REQUIRED"): + context.encode_entity_reference("Order", 1, 1, "edit", timedelta(minutes=1)) + + monkeypatch.setenv( + UNSAFE_RAW_ENTITY_REFERENCES_ENVIRONMENT, + UNSAFE_RAW_ENTITY_REFERENCES_ACKNOWLEDGEMENT, + ) + token = context.encode_entity_reference( + "Order", 1, 1, "edit", timedelta(minutes=1) + ) + assert token.startswith("tqr0.") + assert context.decode_entity_reference(token, "Order", "edit").id == 1 + with pytest.raises(EntityReferenceTokenError, match="ENTITY_REFERENCE_INVALID"): + context.decode_entity_reference(token, "Order", "other-purpose")