231 lines
7.6 KiB
Python
231 lines
7.6 KiB
Python
"""Credential-free contracts for discovering and acquiring Evidence objects."""
|
|
|
|
import re
|
|
from collections.abc import Iterable, Mapping, Sequence
|
|
from datetime import UTC, datetime
|
|
from enum import Enum
|
|
from typing import Protocol, Self, runtime_checkable
|
|
from urllib.parse import parse_qsl, urlsplit, urlunsplit
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter, field_validator
|
|
|
|
|
|
class FrozenDict(dict):
|
|
"""A JSON-serializable dict whose mutation operations are disabled."""
|
|
|
|
def _immutable(self, *args, **kwargs):
|
|
raise TypeError("frozen JSON metadata cannot be mutated")
|
|
|
|
__delitem__ = _immutable
|
|
__ior__ = _immutable
|
|
__setitem__ = _immutable
|
|
clear = _immutable
|
|
pop = _immutable
|
|
popitem = _immutable
|
|
setdefault = _immutable
|
|
update = _immutable
|
|
|
|
|
|
class FrozenList(list):
|
|
"""A JSON-serializable list whose mutation operations are disabled."""
|
|
|
|
def _immutable(self, *args, **kwargs):
|
|
raise TypeError("frozen JSON metadata cannot be mutated")
|
|
|
|
__delitem__ = _immutable
|
|
__iadd__ = _immutable
|
|
__imul__ = _immutable
|
|
__setitem__ = _immutable
|
|
append = _immutable
|
|
clear = _immutable
|
|
extend = _immutable
|
|
insert = _immutable
|
|
pop = _immutable
|
|
remove = _immutable
|
|
reverse = _immutable
|
|
sort = _immutable
|
|
|
|
|
|
_CAMEL_BOUNDARY = re.compile(r"(?<=[a-z0-9])(?=[A-Z])")
|
|
_SEPARATORS = re.compile(r"[^a-z0-9]+")
|
|
_NAMESPACED_VALUE = re.compile(r"^[a-z][a-z0-9_-]*:[A-Za-z0-9._:-]+$")
|
|
_CREDENTIAL_KEYS = {
|
|
"apikey",
|
|
"authorization",
|
|
"authtoken",
|
|
"bearertoken",
|
|
"clientsecret",
|
|
"credential",
|
|
"credentials",
|
|
"password",
|
|
"passwd",
|
|
"privatekey",
|
|
"refreshtoken",
|
|
"sessioncookie",
|
|
"xapikey",
|
|
"accesstoken",
|
|
}
|
|
_JSON_METADATA = TypeAdapter(dict[str, JsonValue])
|
|
|
|
|
|
def _normalize_key(key: str) -> str:
|
|
return _SEPARATORS.sub("", _CAMEL_BOUNDARY.sub("_", key).lower())
|
|
|
|
|
|
def _is_credential_key(key: str) -> bool:
|
|
return _normalize_key(key) in _CREDENTIAL_KEYS
|
|
|
|
|
|
def _reject_credentials(value, path: str = "metadata") -> None:
|
|
if isinstance(value, Mapping):
|
|
for key, child in value.items():
|
|
if _is_credential_key(str(key)):
|
|
raise ValueError(f"credential-like metadata key is not allowed: {path}.{key}")
|
|
_reject_credentials(child, f"{path}.{key}")
|
|
elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
|
|
for index, child in enumerate(value):
|
|
_reject_credentials(child, f"{path}[{index}]")
|
|
|
|
|
|
def freeze_json(value):
|
|
"""Recursively freeze a Pydantic-validated JSON value without changing its JSON shape."""
|
|
if isinstance(value, Mapping):
|
|
return FrozenDict({str(key): freeze_json(child) for key, child in value.items()})
|
|
if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
|
|
return FrozenList(freeze_json(child) for child in value)
|
|
return value
|
|
|
|
|
|
def validate_safe_metadata(value: dict[str, JsonValue]) -> FrozenDict:
|
|
_reject_credentials(value)
|
|
return freeze_json(value)
|
|
|
|
|
|
def validate_canonical_uri(value: str) -> str:
|
|
try:
|
|
parsed = urlsplit(value)
|
|
_ = parsed.port
|
|
except ValueError as error:
|
|
raise ValueError("invalid canonical URI") from error
|
|
if not parsed.scheme:
|
|
raise ValueError("canonical URI must include a scheme")
|
|
if parsed.username is not None or parsed.password is not None:
|
|
raise ValueError("canonical URI must not contain credentials in userinfo")
|
|
for key, _ in parse_qsl(parsed.query, keep_blank_values=True):
|
|
if _is_credential_key(key):
|
|
raise ValueError("canonical URI must not contain credentials in query parameters")
|
|
return value
|
|
|
|
|
|
def canonical_provenance_uri(value: str) -> str:
|
|
"""Return only stable URI identity; transport query/fragment data is never provenance."""
|
|
try:
|
|
parsed = urlsplit(value)
|
|
_ = parsed.port
|
|
except ValueError as error:
|
|
raise ValueError("invalid canonical URI") from error
|
|
if not parsed.scheme:
|
|
raise ValueError("canonical URI must include a scheme")
|
|
if parsed.username is not None or parsed.password is not None:
|
|
raise ValueError("canonical URI must not contain credentials in userinfo")
|
|
return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))
|
|
|
|
|
|
def normalize_aware_datetime(value: datetime | None) -> datetime | None:
|
|
if value is None:
|
|
return None
|
|
if value.tzinfo is None or value.utcoffset() is None:
|
|
raise ValueError("datetime must be timezone-aware")
|
|
return value.astimezone(UTC)
|
|
|
|
|
|
def validate_namespaced_value(value: str) -> str:
|
|
if not _NAMESPACED_VALUE.fullmatch(value):
|
|
raise ValueError("value must be namespaced as '<kind>:<stable-value>'")
|
|
return value
|
|
|
|
|
|
class _EvidenceValue(BaseModel):
|
|
model_config = ConfigDict(
|
|
frozen=True,
|
|
extra="forbid",
|
|
revalidate_instances="always",
|
|
validate_default=True,
|
|
ser_json_bytes="base64",
|
|
val_json_bytes="base64",
|
|
)
|
|
|
|
def model_copy(self, *, update: Mapping[str, object] | None = None, deep: bool = False) -> Self:
|
|
"""Copy through validation; Pydantic's unchecked update-copy is unsafe for contracts."""
|
|
data = self.model_dump(round_trip=True)
|
|
if update:
|
|
data.update(update)
|
|
return type(self).model_validate(data)
|
|
|
|
|
|
class SourceObject(_EvidenceValue):
|
|
source_id: str = Field(min_length=1)
|
|
uri: str = Field(min_length=1)
|
|
fingerprint: str = Field(min_length=1)
|
|
modified_at: datetime | None = None
|
|
metadata: dict[str, JsonValue] = Field(default_factory=dict)
|
|
|
|
_source_id = field_validator("source_id")(validate_namespaced_value)
|
|
_fingerprint = field_validator("fingerprint")(validate_namespaced_value)
|
|
_safe_uri = field_validator("uri")(validate_canonical_uri)
|
|
_aware_modified_at = field_validator("modified_at")(normalize_aware_datetime)
|
|
_frozen_metadata = field_validator("metadata")(validate_safe_metadata)
|
|
|
|
|
|
class AcquiredDocument(_EvidenceValue):
|
|
"""Transport result; bytes use explicit base64 encoding in JSON mode."""
|
|
|
|
source: SourceObject
|
|
content: bytes
|
|
media_type: str | None = None
|
|
acquired_at: datetime | None = None
|
|
metadata: dict[str, JsonValue] = Field(default_factory=dict)
|
|
|
|
_aware_acquired_at = field_validator("acquired_at")(normalize_aware_datetime)
|
|
_frozen_metadata = field_validator("metadata")(validate_safe_metadata)
|
|
|
|
|
|
class EvidenceSourceErrorCategory(str, Enum):
|
|
TRANSIENT = "transient"
|
|
PERMANENT = "permanent"
|
|
|
|
|
|
class EvidenceSourceError(Exception):
|
|
"""Classified source failure with credential-free structured diagnostics."""
|
|
|
|
def __init__(
|
|
self,
|
|
_message: str,
|
|
*,
|
|
category: EvidenceSourceErrorCategory,
|
|
details: dict[str, JsonValue] | None = None,
|
|
) -> None:
|
|
super().__init__("evidence source operation failed")
|
|
object.__setattr__(self, "category", EvidenceSourceErrorCategory(category))
|
|
object.__setattr__(
|
|
self,
|
|
"details",
|
|
validate_safe_metadata(_JSON_METADATA.validate_python(details or {})),
|
|
)
|
|
|
|
def __setattr__(self, name: str, value) -> None:
|
|
if name in {"args", "category", "details"} and hasattr(self, name):
|
|
raise AttributeError(f"{name} is immutable")
|
|
super().__setattr__(name, value)
|
|
|
|
@property
|
|
def retryable(self) -> bool:
|
|
return self.category is EvidenceSourceErrorCategory.TRANSIENT
|
|
|
|
|
|
@runtime_checkable
|
|
class EvidenceSource(Protocol):
|
|
def discover(self) -> Iterable[SourceObject]: ...
|
|
|
|
def acquire(self, item: SourceObject) -> AcquiredDocument: ...
|