fix(evidence): close canonical contract gaps

This commit is contained in:
2026-07-12 03:15:17 +02:00
parent 4424fd3d90
commit 293d96e1a6
5 changed files with 176 additions and 15 deletions
+36 -6
View File
@@ -4,8 +4,8 @@ import re
from collections.abc import Iterable, Mapping, Sequence
from datetime import UTC, datetime
from enum import Enum
from typing import Protocol, runtime_checkable
from urllib.parse import parse_qsl, urlsplit
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
@@ -97,6 +97,20 @@ def validate_canonical_uri(value: str) -> str:
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
@@ -121,6 +135,13 @@ class _EvidenceValue(BaseModel):
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)
@@ -159,14 +180,23 @@ class EvidenceSourceError(Exception):
def __init__(
self,
message: str,
_message: str,
*,
category: EvidenceSourceErrorCategory,
details: dict[str, JsonValue] | None = None,
) -> None:
super().__init__(message)
self.category = EvidenceSourceErrorCategory(category)
self.details = validate_safe_metadata(_JSON_METADATA.validate_python(details or {}))
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: