326 lines
11 KiB
Python
326 lines
11 KiB
Python
# [DEF:backend.src.models.clean_release:Module]
|
|
# @TIER: CRITICAL
|
|
# @SEMANTICS: clean-release, models, lifecycle, policy, manifest, compliance
|
|
# @PURPOSE: Define clean release domain entities and validation contracts for enterprise compliance flow.
|
|
# @LAYER: Domain
|
|
# @RELATION: BINDS_TO -> specs/023-clean-repo-enterprise/data-model.md
|
|
# @INVARIANT: Enterprise-clean policy always forbids external sources.
|
|
# @TEST_CONTRACT: CleanReleaseModelPayload -> ValidatedCleanReleaseModel | ValidationError
|
|
# @TEST_SCENARIO: valid_enterprise_models -> CRITICAL entities validate and preserve lifecycle/compliance invariants.
|
|
# @TEST_FIXTURE: clean_release_models_baseline -> backend/tests/fixtures/clean_release/fixtures_clean_release.json
|
|
# @TEST_EDGE: empty_required_identifiers -> Empty candidate_id/source_snapshot_ref/internal_source_registry_ref fails validation.
|
|
# @TEST_EDGE: compliant_run_missing_mandatory_stage -> COMPLIANT run without all mandatory PASS stages fails validation.
|
|
# @TEST_EDGE: blocked_report_without_blocking_violations -> BLOCKED report with zero blocking violations fails validation.
|
|
# @TEST_INVARIANT: external_source_must_block -> VERIFIED_BY: [valid_enterprise_models, blocked_report_without_blocking_violations]
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
from typing import List, Optional
|
|
|
|
from pydantic import BaseModel, Field, model_validator
|
|
|
|
|
|
# [DEF:ReleaseCandidateStatus:Class]
|
|
# @PURPOSE: Lifecycle states for release candidate.
|
|
class ReleaseCandidateStatus(str, Enum):
|
|
DRAFT = "draft"
|
|
PREPARED = "prepared"
|
|
COMPLIANT = "compliant"
|
|
BLOCKED = "blocked"
|
|
RELEASED = "released"
|
|
# [/DEF:ReleaseCandidateStatus:Class]
|
|
|
|
|
|
# [DEF:ProfileType:Class]
|
|
# @PURPOSE: Supported profile identifiers.
|
|
class ProfileType(str, Enum):
|
|
ENTERPRISE_CLEAN = "enterprise-clean"
|
|
DEVELOPMENT = "development"
|
|
# [/DEF:ProfileType:Class]
|
|
|
|
|
|
# [DEF:ClassificationType:Class]
|
|
# @PURPOSE: Manifest classification outcomes for artifacts.
|
|
class ClassificationType(str, Enum):
|
|
REQUIRED_SYSTEM = "required-system"
|
|
ALLOWED = "allowed"
|
|
EXCLUDED_PROHIBITED = "excluded-prohibited"
|
|
# [/DEF:ClassificationType:Class]
|
|
|
|
|
|
# [DEF:RegistryStatus:Class]
|
|
# @PURPOSE: Registry lifecycle status.
|
|
class RegistryStatus(str, Enum):
|
|
ACTIVE = "active"
|
|
INACTIVE = "inactive"
|
|
# [/DEF:RegistryStatus:Class]
|
|
|
|
|
|
# [DEF:CheckFinalStatus:Class]
|
|
# @PURPOSE: Final status for compliance check run.
|
|
class CheckFinalStatus(str, Enum):
|
|
RUNNING = "running"
|
|
COMPLIANT = "compliant"
|
|
BLOCKED = "blocked"
|
|
FAILED = "failed"
|
|
# [/DEF:CheckFinalStatus:Class]
|
|
|
|
|
|
# [DEF:ExecutionMode:Class]
|
|
# @PURPOSE: Execution channel for compliance checks.
|
|
class ExecutionMode(str, Enum):
|
|
TUI = "tui"
|
|
CI = "ci"
|
|
# [/DEF:ExecutionMode:Class]
|
|
|
|
|
|
# [DEF:CheckStageName:Class]
|
|
# @PURPOSE: Mandatory check stages.
|
|
class CheckStageName(str, Enum):
|
|
DATA_PURITY = "data_purity"
|
|
INTERNAL_SOURCES_ONLY = "internal_sources_only"
|
|
NO_EXTERNAL_ENDPOINTS = "no_external_endpoints"
|
|
MANIFEST_CONSISTENCY = "manifest_consistency"
|
|
# [/DEF:CheckStageName:Class]
|
|
|
|
|
|
# [DEF:CheckStageStatus:Class]
|
|
# @PURPOSE: Stage-level execution status.
|
|
class CheckStageStatus(str, Enum):
|
|
PASS = "pass"
|
|
FAIL = "fail"
|
|
SKIPPED = "skipped"
|
|
# [/DEF:CheckStageStatus:Class]
|
|
|
|
|
|
# [DEF:ViolationCategory:Class]
|
|
# @PURPOSE: Normalized compliance violation categories.
|
|
class ViolationCategory(str, Enum):
|
|
DATA_PURITY = "data-purity"
|
|
EXTERNAL_SOURCE = "external-source"
|
|
MANIFEST_INTEGRITY = "manifest-integrity"
|
|
POLICY_CONFLICT = "policy-conflict"
|
|
OPERATIONAL_RISK = "operational-risk"
|
|
# [/DEF:ViolationCategory:Class]
|
|
|
|
|
|
# [DEF:ViolationSeverity:Class]
|
|
# @PURPOSE: Severity levels for violation triage.
|
|
class ViolationSeverity(str, Enum):
|
|
CRITICAL = "critical"
|
|
HIGH = "high"
|
|
MEDIUM = "medium"
|
|
LOW = "low"
|
|
# [/DEF:ViolationSeverity:Class]
|
|
|
|
|
|
# [DEF:ReleaseCandidate:Class]
|
|
# @PURPOSE: Candidate metadata for clean-release workflow.
|
|
# @PRE: candidate_id, source_snapshot_ref are non-empty.
|
|
# @POST: Model instance is valid for lifecycle transitions.
|
|
class ReleaseCandidate(BaseModel):
|
|
candidate_id: str
|
|
version: str
|
|
profile: ProfileType
|
|
created_at: datetime
|
|
created_by: str
|
|
source_snapshot_ref: str
|
|
status: ReleaseCandidateStatus = ReleaseCandidateStatus.DRAFT
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_non_empty(self):
|
|
if not self.candidate_id.strip():
|
|
raise ValueError("candidate_id must be non-empty")
|
|
if not self.source_snapshot_ref.strip():
|
|
raise ValueError("source_snapshot_ref must be non-empty")
|
|
return self
|
|
# [/DEF:ReleaseCandidate:Class]
|
|
|
|
|
|
# [DEF:CleanProfilePolicy:Class]
|
|
# @PURPOSE: Policy contract for artifact/source decisions.
|
|
class CleanProfilePolicy(BaseModel):
|
|
policy_id: str
|
|
policy_version: str
|
|
active: bool
|
|
prohibited_artifact_categories: List[str] = Field(default_factory=list)
|
|
required_system_categories: List[str] = Field(default_factory=list)
|
|
external_source_forbidden: bool = True
|
|
internal_source_registry_ref: str
|
|
effective_from: datetime
|
|
effective_to: Optional[datetime] = None
|
|
profile: ProfileType = ProfileType.ENTERPRISE_CLEAN
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_policy(self):
|
|
if self.profile == ProfileType.ENTERPRISE_CLEAN:
|
|
if not self.external_source_forbidden:
|
|
raise ValueError("enterprise-clean policy requires external_source_forbidden=true")
|
|
if not self.prohibited_artifact_categories:
|
|
raise ValueError("enterprise-clean policy requires prohibited_artifact_categories")
|
|
if not self.internal_source_registry_ref.strip():
|
|
raise ValueError("internal_source_registry_ref must be non-empty")
|
|
return self
|
|
# [/DEF:CleanProfilePolicy:Class]
|
|
|
|
|
|
# [DEF:ResourceSourceEntry:Class]
|
|
# @PURPOSE: One internal source definition.
|
|
class ResourceSourceEntry(BaseModel):
|
|
source_id: str
|
|
host: str
|
|
protocol: str
|
|
purpose: str
|
|
allowed_paths: List[str] = Field(default_factory=list)
|
|
enabled: bool = True
|
|
# [/DEF:ResourceSourceEntry:Class]
|
|
|
|
|
|
# [DEF:ResourceSourceRegistry:Class]
|
|
# @PURPOSE: Allowlist of internal sources.
|
|
class ResourceSourceRegistry(BaseModel):
|
|
registry_id: str
|
|
name: str
|
|
entries: List[ResourceSourceEntry]
|
|
updated_at: datetime
|
|
updated_by: str
|
|
status: RegistryStatus = RegistryStatus.ACTIVE
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_registry(self):
|
|
if not self.entries:
|
|
raise ValueError("registry entries cannot be empty")
|
|
if self.status == RegistryStatus.ACTIVE and not any(e.enabled for e in self.entries):
|
|
raise ValueError("active registry must include at least one enabled entry")
|
|
return self
|
|
# [/DEF:ResourceSourceRegistry:Class]
|
|
|
|
|
|
# [DEF:ManifestItem:Class]
|
|
# @PURPOSE: One artifact entry in manifest.
|
|
class ManifestItem(BaseModel):
|
|
path: str
|
|
category: str
|
|
classification: ClassificationType
|
|
reason: str
|
|
checksum: Optional[str] = None
|
|
# [/DEF:ManifestItem:Class]
|
|
|
|
|
|
# [DEF:ManifestSummary:Class]
|
|
# @PURPOSE: Aggregate counters for manifest decisions.
|
|
class ManifestSummary(BaseModel):
|
|
included_count: int = Field(ge=0)
|
|
excluded_count: int = Field(ge=0)
|
|
prohibited_detected_count: int = Field(ge=0)
|
|
# [/DEF:ManifestSummary:Class]
|
|
|
|
|
|
# [DEF:DistributionManifest:Class]
|
|
# @PURPOSE: Deterministic release composition for audit.
|
|
class DistributionManifest(BaseModel):
|
|
manifest_id: str
|
|
candidate_id: str
|
|
policy_id: str
|
|
generated_at: datetime
|
|
generated_by: str
|
|
items: List[ManifestItem]
|
|
summary: ManifestSummary
|
|
deterministic_hash: str
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_counts(self):
|
|
if self.summary.included_count + self.summary.excluded_count != len(self.items):
|
|
raise ValueError("manifest summary counts must match items size")
|
|
return self
|
|
# [/DEF:DistributionManifest:Class]
|
|
|
|
|
|
# [DEF:CheckStageResult:Class]
|
|
# @PURPOSE: Per-stage compliance result.
|
|
class CheckStageResult(BaseModel):
|
|
stage: CheckStageName
|
|
status: CheckStageStatus
|
|
details: Optional[str] = None
|
|
duration_ms: Optional[int] = Field(default=None, ge=0)
|
|
# [/DEF:CheckStageResult:Class]
|
|
|
|
|
|
# [DEF:ComplianceCheckRun:Class]
|
|
# @PURPOSE: One execution run of compliance pipeline.
|
|
class ComplianceCheckRun(BaseModel):
|
|
check_run_id: str
|
|
candidate_id: str
|
|
policy_id: str
|
|
started_at: datetime
|
|
finished_at: Optional[datetime] = None
|
|
final_status: CheckFinalStatus = CheckFinalStatus.RUNNING
|
|
triggered_by: str
|
|
execution_mode: ExecutionMode
|
|
checks: List[CheckStageResult] = Field(default_factory=list)
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_terminal_integrity(self):
|
|
if self.final_status == CheckFinalStatus.COMPLIANT:
|
|
mandatory = {c.stage: c.status for c in self.checks}
|
|
required = {
|
|
CheckStageName.DATA_PURITY,
|
|
CheckStageName.INTERNAL_SOURCES_ONLY,
|
|
CheckStageName.NO_EXTERNAL_ENDPOINTS,
|
|
CheckStageName.MANIFEST_CONSISTENCY,
|
|
}
|
|
if not required.issubset(mandatory.keys()):
|
|
raise ValueError("compliant run requires all mandatory stages")
|
|
if any(mandatory[s] != CheckStageStatus.PASS for s in required):
|
|
raise ValueError("compliant run requires PASS on all mandatory stages")
|
|
return self
|
|
# [/DEF:ComplianceCheckRun:Class]
|
|
|
|
|
|
# [DEF:ComplianceViolation:Class]
|
|
# @PURPOSE: Normalized violation row for triage and blocking decisions.
|
|
class ComplianceViolation(BaseModel):
|
|
violation_id: str
|
|
check_run_id: str
|
|
category: ViolationCategory
|
|
severity: ViolationSeverity
|
|
location: str
|
|
evidence: Optional[str] = None
|
|
remediation: str
|
|
blocked_release: bool
|
|
detected_at: datetime
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_violation(self):
|
|
if self.category == ViolationCategory.EXTERNAL_SOURCE and not self.blocked_release:
|
|
raise ValueError("external-source violation must block release")
|
|
if self.severity == ViolationSeverity.CRITICAL and not self.remediation.strip():
|
|
raise ValueError("critical violation requires remediation")
|
|
return self
|
|
# [/DEF:ComplianceViolation:Class]
|
|
|
|
|
|
# [DEF:ComplianceReport:Class]
|
|
# @PURPOSE: Final report payload for operator and audit systems.
|
|
class ComplianceReport(BaseModel):
|
|
report_id: str
|
|
check_run_id: str
|
|
candidate_id: str
|
|
generated_at: datetime
|
|
final_status: CheckFinalStatus
|
|
operator_summary: str
|
|
structured_payload_ref: str
|
|
violations_count: int = Field(ge=0)
|
|
blocking_violations_count: int = Field(ge=0)
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_report_counts(self):
|
|
if self.blocking_violations_count > self.violations_count:
|
|
raise ValueError("blocking_violations_count cannot exceed violations_count")
|
|
if self.final_status == CheckFinalStatus.BLOCKED and self.blocking_violations_count <= 0:
|
|
raise ValueError("blocked report requires blocking violations")
|
|
return self
|
|
# [/DEF:ComplianceReport:Class]
|
|
# [/DEF:backend.src.models.clean_release:Module] |