feat: add verifiable audit evidence bundles
This commit is contained in:
@@ -3,7 +3,7 @@ from __future__ import annotations
|
||||
from datetime import datetime
|
||||
from typing import Any, Literal
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
||||
|
||||
from govoplan_core.api.v1.schemas import DeltaDeletedItem
|
||||
|
||||
@@ -79,3 +79,66 @@ class EventDeliveryReplayResponse(BaseModel):
|
||||
last_replayed_by: str | None = None
|
||||
last_replay_reason: str | None = None
|
||||
last_error: str | None = None
|
||||
|
||||
|
||||
class EvidenceBundleReferenceRequest(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
reference_id: str = Field(min_length=1, max_length=200)
|
||||
kind: str = Field(min_length=1, max_length=80)
|
||||
owner_module: str = Field(min_length=1, max_length=100)
|
||||
locator: str = Field(min_length=1, max_length=2048)
|
||||
content_sha256: str | None = Field(default=None, pattern=r"^[0-9A-Fa-f]{64}$")
|
||||
required: bool = True
|
||||
|
||||
|
||||
class EvidenceBundleExportRequest(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
scope: Literal["tenant", "system", "all"] = "tenant"
|
||||
tenant_id: str | None = Field(default=None, max_length=36)
|
||||
since: datetime | None = None
|
||||
until: datetime | None = None
|
||||
record_ids: list[str] = Field(default_factory=list, max_length=500)
|
||||
action: str | None = Field(default=None, max_length=100)
|
||||
object_type: str | None = Field(default=None, max_length=100)
|
||||
object_id: str | None = Field(default=None, max_length=100)
|
||||
max_records: int = Field(default=500, ge=1, le=500)
|
||||
references: list[EvidenceBundleReferenceRequest] = Field(
|
||||
default_factory=list,
|
||||
max_length=200,
|
||||
)
|
||||
sign: bool = False
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_window_and_selection(self):
|
||||
for label, value in (("since", self.since), ("until", self.until)):
|
||||
if value is not None and (value.tzinfo is None or value.utcoffset() is None):
|
||||
raise ValueError(f"Evidence bundle {label} must include a timezone.")
|
||||
if self.since and self.until and self.until < self.since:
|
||||
raise ValueError("Evidence bundle until must not precede since.")
|
||||
if len(self.record_ids) != len(set(self.record_ids)):
|
||||
raise ValueError("Evidence bundle record ids must be unique.")
|
||||
reference_ids = [item.reference_id for item in self.references]
|
||||
if len(reference_ids) != len(set(reference_ids)):
|
||||
raise ValueError("Evidence bundle reference ids must be unique.")
|
||||
return self
|
||||
|
||||
|
||||
class EvidenceBundleResponse(BaseModel):
|
||||
id: str
|
||||
scope: Literal["tenant", "system", "all"]
|
||||
tenant_id: str | None = None
|
||||
status: Literal["pending", "ready", "failed"]
|
||||
bundle_sha256: str | None = None
|
||||
record_count: int
|
||||
reference_count: int
|
||||
generated_at: datetime | None = None
|
||||
downloaded_at: datetime | None = None
|
||||
error_code: str | None = None
|
||||
created_at: datetime
|
||||
download_url: str | None = None
|
||||
|
||||
|
||||
class EvidenceBundleDownloadResponse(BaseModel):
|
||||
bundle: dict[str, Any]
|
||||
|
||||
Reference in New Issue
Block a user