Christian Nennemann
884d2dc836
- Rename `par` to `pred` (predecessor) in types, serialization, tests - Remove `pol`, `pol_decision` from core payload; move to `ect_ext` - Remove `sub` from payload (not part of ECT spec) - Update `typ` from `wimse-exec+jwt` to `exec+jwt` (accept both) - Rename MaxParLength to MaxPredLength everywhere - Update testdata, demos, READMEs with migration table - All Go tests pass, all 56 Python tests pass (90% coverage)
66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
"""Validation helpers: ext size/depth, UUID, inp_hash/out_hash format."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
|
|
EXT_MAX_SIZE = 4096
|
|
EXT_MAX_DEPTH = 5
|
|
DEFAULT_MAX_PRED_LENGTH = 100
|
|
|
|
_UUID_RE = re.compile(
|
|
r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$"
|
|
)
|
|
_ALLOWED_HASH_ALGS = frozenset(("sha-256", "sha-384", "sha-512"))
|
|
|
|
|
|
def _json_depth(obj: Any, depth: int = 0) -> int:
|
|
if depth > EXT_MAX_DEPTH:
|
|
return depth
|
|
if isinstance(obj, dict):
|
|
return max((_json_depth(v, depth + 1) for v in obj.values()), default=depth + 1)
|
|
if isinstance(obj, list):
|
|
return max((_json_depth(x, depth + 1) for x in obj), default=depth + 1)
|
|
return depth
|
|
|
|
|
|
def validate_ext(ext: dict[str, Any] | None) -> None:
|
|
"""Raise ValueError if ext exceeds EXT_MAX_SIZE or nesting depth EXT_MAX_DEPTH."""
|
|
if not ext:
|
|
return
|
|
raw = json.dumps(ext)
|
|
if len(raw.encode("utf-8")) > EXT_MAX_SIZE:
|
|
raise ValueError("ect: ext exceeds max size (4096 bytes)")
|
|
if _json_depth(ext) > EXT_MAX_DEPTH:
|
|
raise ValueError("ect: ext exceeds max nesting depth (5)")
|
|
|
|
|
|
def valid_uuid(s: str) -> bool:
|
|
"""Return True if s is a UUID string (RFC 9562)."""
|
|
return bool(_UUID_RE.match(s))
|
|
|
|
|
|
def validate_hash_format(s: str) -> None:
|
|
"""Raise ValueError if s is non-empty and not algorithm:base64url (sha-256, sha-384, sha-512)."""
|
|
if not s:
|
|
return
|
|
idx = s.find(":")
|
|
if idx <= 0:
|
|
raise ValueError("ect: inp_hash/out_hash must be algorithm:base64url (e.g. sha-256:...)")
|
|
alg = s[:idx].lower()
|
|
if alg not in _ALLOWED_HASH_ALGS:
|
|
raise ValueError("ect: inp_hash/out_hash must be algorithm:base64url (e.g. sha-256:...)")
|
|
encoded = s[idx + 1:]
|
|
if not encoded:
|
|
raise ValueError("ect: inp_hash/out_hash must be algorithm:base64url (e.g. sha-256:...)")
|
|
pad = 4 - len(encoded) % 4
|
|
if pad != 4:
|
|
encoded += "=" * pad
|
|
try:
|
|
base64.urlsafe_b64decode(encoded)
|
|
except Exception:
|
|
raise ValueError("ect: inp_hash/out_hash must be algorithm:base64url (e.g. sha-256:...)") from None
|