75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
"""Port for discovering and acquiring Evidence source objects.
|
|
|
|
Source adapters own transport details and credentials. The values crossing this
|
|
boundary are deliberately credential-free so they can safely become provenance.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from typing import Iterable, Protocol, runtime_checkable
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, JsonValue, field_validator
|
|
|
|
|
|
_SECRET_KEYS = {
|
|
"api_key",
|
|
"apikey",
|
|
"authorization",
|
|
"credential",
|
|
"credentials",
|
|
"password",
|
|
"secret",
|
|
"token",
|
|
}
|
|
|
|
|
|
def _reject_credentials(value: JsonValue, path: str = "metadata") -> JsonValue:
|
|
if isinstance(value, dict):
|
|
for key, child in value.items():
|
|
normalized = key.lower().replace("-", "_")
|
|
if normalized in _SECRET_KEYS or normalized.endswith(("_password", "_secret", "_token")):
|
|
raise ValueError(f"credential-like metadata key is not allowed: {path}.{key}")
|
|
_reject_credentials(child, f"{path}.{key}")
|
|
elif isinstance(value, list):
|
|
for index, child in enumerate(value):
|
|
_reject_credentials(child, f"{path}[{index}]")
|
|
return value
|
|
|
|
|
|
class _EvidenceValue(BaseModel):
|
|
model_config = ConfigDict(frozen=True, extra="forbid")
|
|
|
|
|
|
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)
|
|
|
|
@field_validator("metadata")
|
|
@classmethod
|
|
def metadata_has_no_credentials(cls, value: dict[str, JsonValue]) -> dict[str, JsonValue]:
|
|
_reject_credentials(value)
|
|
return value
|
|
|
|
|
|
class AcquiredDocument(_EvidenceValue):
|
|
source: SourceObject
|
|
content: bytes
|
|
media_type: str | None = None
|
|
acquired_at: datetime | None = None
|
|
metadata: dict[str, JsonValue] = Field(default_factory=dict)
|
|
|
|
@field_validator("metadata")
|
|
@classmethod
|
|
def metadata_has_no_credentials(cls, value: dict[str, JsonValue]) -> dict[str, JsonValue]:
|
|
_reject_credentials(value)
|
|
return value
|
|
|
|
|
|
@runtime_checkable
|
|
class EvidenceSource(Protocol):
|
|
def discover(self) -> Iterable[SourceObject]: ...
|
|
|
|
def acquire(self, item: SourceObject) -> AcquiredDocument: ...
|