138 lines
5.1 KiB
Python
138 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
|
|
|
|
|
RETENTION_DAY_KEYS = (
|
|
"raw_campaign_json_retention_days",
|
|
"generated_eml_retention_days",
|
|
"stored_report_detail_retention_days",
|
|
"mock_mailbox_retention_days",
|
|
"audit_detail_retention_days",
|
|
)
|
|
|
|
|
|
RETENTION_POLICY_FIELD_KEYS = (
|
|
"store_raw_campaign_json",
|
|
*RETENTION_DAY_KEYS,
|
|
"audit_detail_level",
|
|
)
|
|
|
|
|
|
def default_allow_lower_level_limits() -> dict[str, bool]:
|
|
return {key: True for key in RETENTION_POLICY_FIELD_KEYS}
|
|
|
|
|
|
def normalize_allow_lower_level_limits(value: Any, *, fill_defaults: bool) -> dict[str, bool] | None:
|
|
if value in (None, ""):
|
|
return default_allow_lower_level_limits() if fill_defaults else None
|
|
if not isinstance(value, dict):
|
|
raise ValueError("allow_lower_level_limits must be an object")
|
|
normalized = default_allow_lower_level_limits() if fill_defaults else {}
|
|
for key, allowed in value.items():
|
|
clean_key = str(key)
|
|
if clean_key not in RETENTION_POLICY_FIELD_KEYS:
|
|
raise ValueError(f"Unknown retention policy field: {clean_key}")
|
|
normalized[clean_key] = bool(allowed)
|
|
return normalized
|
|
|
|
|
|
class PolicySourceStepItem(BaseModel):
|
|
scope_type: str
|
|
scope_id: str | None = None
|
|
path: str
|
|
label: str
|
|
applied_fields: list[str] = Field(default_factory=list)
|
|
policy: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class PrivacyRetentionPolicyItem(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
store_raw_campaign_json: bool = True
|
|
raw_campaign_json_retention_days: int | None = Field(default=None, ge=0)
|
|
generated_eml_retention_days: int | None = Field(default=None, ge=0)
|
|
stored_report_detail_retention_days: int | None = Field(default=None, ge=0)
|
|
mock_mailbox_retention_days: int | None = Field(default=None, ge=0)
|
|
audit_detail_retention_days: int | None = Field(default=None, ge=0)
|
|
audit_detail_level: Literal["full", "redacted", "minimal"] = "full"
|
|
allow_lower_level_limits: dict[str, bool] = Field(default_factory=default_allow_lower_level_limits)
|
|
|
|
@field_validator("allow_lower_level_limits", mode="before")
|
|
@classmethod
|
|
def _normalize_allow_lower_level_limits(cls, value: Any) -> Any:
|
|
return normalize_allow_lower_level_limits(value, fill_defaults=True)
|
|
|
|
|
|
class PrivacyRetentionPolicyPatchItem(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
store_raw_campaign_json: bool | None = None
|
|
raw_campaign_json_retention_days: int | None = Field(default=None, ge=0)
|
|
generated_eml_retention_days: int | None = Field(default=None, ge=0)
|
|
stored_report_detail_retention_days: int | None = Field(default=None, ge=0)
|
|
mock_mailbox_retention_days: int | None = Field(default=None, ge=0)
|
|
audit_detail_retention_days: int | None = Field(default=None, ge=0)
|
|
audit_detail_level: Literal["full", "redacted", "minimal"] | None = None
|
|
allow_lower_level_limits: dict[str, bool] | None = None
|
|
|
|
@field_validator("allow_lower_level_limits", mode="before")
|
|
@classmethod
|
|
def _normalize_allow_lower_level_limits(cls, value: Any) -> Any:
|
|
return normalize_allow_lower_level_limits(value, fill_defaults=False)
|
|
|
|
|
|
class PrivacyRetentionPolicyScopeRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
policy: PrivacyRetentionPolicyPatchItem = Field(default_factory=PrivacyRetentionPolicyPatchItem)
|
|
change_request_id: str | None = None
|
|
|
|
|
|
class PrivacyRetentionPolicyScopeResponse(BaseModel):
|
|
scope_type: Literal["system", "tenant", "user", "group", "campaign"]
|
|
scope_id: str | None = None
|
|
policy: dict[str, Any]
|
|
effective_policy: PrivacyRetentionPolicyItem
|
|
parent_policy: PrivacyRetentionPolicyItem | None = None
|
|
effective_policy_sources: list[PolicySourceStepItem] = Field(default_factory=list)
|
|
parent_policy_sources: list[PolicySourceStepItem] = Field(default_factory=list)
|
|
|
|
|
|
class PolicyDecisionItem(BaseModel):
|
|
allowed: bool
|
|
reason: str | None = None
|
|
source_path: list[PolicySourceStepItem] = Field(default_factory=list)
|
|
requirements: list[str] = Field(default_factory=list)
|
|
details: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class PrivacyRetentionPolicyExplainResponse(BaseModel):
|
|
scope_type: Literal["system", "tenant", "user", "group", "campaign"]
|
|
scope_id: str | None = None
|
|
decision: PolicyDecisionItem
|
|
effective_policy: PrivacyRetentionPolicyItem
|
|
parent_policy: PrivacyRetentionPolicyItem | None = None
|
|
effective_policy_sources: list[PolicySourceStepItem] = Field(default_factory=list)
|
|
parent_policy_sources: list[PolicySourceStepItem] = Field(default_factory=list)
|
|
blocked_fields: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class PrivacyRetentionPolicySimulationResponse(BaseModel):
|
|
scope_type: Literal["system", "tenant", "user", "group", "campaign"]
|
|
scope_id: str | None = None
|
|
simulation: dict[str, Any]
|
|
|
|
|
|
class RetentionRunRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
dry_run: bool = True
|
|
|
|
|
|
class RetentionRunResponse(BaseModel):
|
|
result: dict[str, Any]
|