ThothII/harness/tht/ports/evidence.py

"""Credential-free port 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, runtime_checkable
from urllib.parse import parse_qsl, urlsplit

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


_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 tuple(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 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",
    )


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__(message)
        self.category = EvidenceSourceErrorCategory(category)
        self.details = validate_safe_metadata(_JSON_METADATA.validate_python(details or {}))

    @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: ...