DataOps-platform/app/core/system/security_governance.py

"""Cross-domain data security governance and security engineering controls."""

from __future__ import annotations

import copy
import hashlib
import json
import re
import uuid
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from typing import Any
from urllib.parse import urlsplit

from app.core.common.identifiers import new_governance_uid
from app.core.common.timezone_utils import now_china

CLASSIFICATIONS = ("public", "internal", "sensitive", "highly_sensitive")
CLASSIFICATION_RANK = {value: index for index, value in enumerate(CLASSIFICATIONS)}
FINDING_STATUSES = {"pending_review", "confirmed", "dismissed"}
ACCESS_ACTIONS = {"read", "use"}
ENVIRONMENTS = {"development", "test", "production"}
RETENTION_EVIDENCE_TYPES = {
    "classification_evidence", "access_decision", "egress_request",
    "audit_event", "siem_delivery", "sbom", "vulnerability",
}
ARCHIVE_MODES = {"hot", "immutable_external"}
DISPOSITION_ACTIONS = {"review", "archive"}
SIEM_CATEGORIES = {
    "authentication", "ingestion", "entity_resolution", "publication",
    "remediation", "knowledge_query", "authorization", "workflow_task",
    "data_product", "agent", "security_governance",
}
SEVERITIES = {"unknown", "low", "medium", "high", "critical"}
RESOLUTION_TYPES = {"patched", "not_affected", "accepted_risk"}
ROLE_PATTERN = re.compile(r"^[a-z][a-z0-9:_-]{1,79}$")
CODE_PATTERN = re.compile(r"^[A-Z][A-Z0-9_]{2,119}$")
FIELD_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]{0,199}$")
HASH_PATTERN = re.compile(r"^[0-9a-f]{64}$")
PHONE_PATTERN = re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)")
EMAIL_PATTERN = re.compile(r"(?i)\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b")
BANK_CARD_PATTERN = re.compile(r"(?<!\d)\d{15,19}(?!\d)")
PRC_ID_PATTERN = re.compile(r"(?<!\d)\d{17}[0-9Xx](?!\d)")


def _closed(value: Any, allowed: set[str], label: str) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ValueError(f"{label} must be an object")
    unknown = sorted(set(value) - allowed)
    if unknown:
        raise ValueError(f"{label} contains unsupported fields: {', '.join(unknown)}")
    return copy.deepcopy(value)


def _text(value: Any, label: str, maximum: int = 1000) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"{label} is required")
    result = value.strip()
    if len(result) > maximum:
        raise ValueError(f"{label} exceeds {maximum} characters")
    return result


def _optional_text(value: Any, label: str, maximum: int = 1000) -> str | None:
    if value in (None, ""):
        return None
    return _text(value, label, maximum)


def _uid(value: Any, label: str) -> str:
    try:
        return str(uuid.UUID(str(value)))
    except (TypeError, ValueError, AttributeError) as error:
        raise ValueError(f"{label} must be a UUID") from error


def _list(value: Any, label: str, minimum: int = 0, maximum: int = 1000) -> list[Any]:
    if not isinstance(value, list) or len(value) < minimum or len(value) > maximum:
        raise ValueError(f"{label} must contain between {minimum} and {maximum} items")
    return copy.deepcopy(value)


def _time(value: Any, label: str) -> datetime:
    if isinstance(value, datetime):
        result = value
    else:
        try:
            result = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
        except (TypeError, ValueError) as error:
            raise ValueError(f"{label} must be ISO-8601") from error
    if result.tzinfo is None:
        raise ValueError(f"{label} must include a timezone")
    return result.astimezone(UTC)


def _classification(value: Any, label: str = "classification") -> str:
    result = _text(value, label, 40)
    if result not in CLASSIFICATION_RANK:
        raise ValueError(f"unsupported {label}")
    return result


def _canonical(value: Any) -> bytes:
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()


def _digest(value: Any) -> str:
    return hashlib.sha256(_canonical(value)).hexdigest()


def _mask(value: str, detector: str) -> str:
    if detector == "email" and "@" in value:
        local, domain = value.split("@", 1)
        return f"{local[:1]}***@{domain}"
    if detector == "phone" and len(value) >= 7:
        return f"{value[:3]}****{value[-4:]}"
    if len(value) >= 6:
        return f"{value[:2]}****{value[-2:]}"
    return "***"


def _normalize_evidence(value: Any) -> list[dict[str, str]]:
    result = []
    for item in _list(value, "evidence_refs", 1, 50):
        body = _closed(item, {"type", "ref", "digest"}, "evidence reference")
        digest = _text(body.get("digest"), "evidence digest", 64).lower()
        if not HASH_PATTERN.fullmatch(digest):
            raise ValueError("evidence digest must be SHA-256")
        result.append({
            "type": _text(body.get("type"), "evidence type", 60),
            "ref": _text(body.get("ref"), "evidence ref", 300),
            "digest": digest,
        })
    return result


class SecurityGovernanceService:
    """Own security decisions while leaving data and external security tools authoritative."""

    def __init__(
        self,
        repository,
        *,
        approval_gateway,
        siem_transport,
        siem_host_allowlist: set[str] | frozenset[str],
        uid_factory: Callable[[], str] = new_governance_uid,
        now_factory: Callable[[], datetime] = now_china,
        commit: Callable[[], None] = lambda: None,
        rollback: Callable[[], None] = lambda: None,
    ):
        self.repository = repository
        self.approval_gateway = approval_gateway
        self.siem_transport = siem_transport
        self.siem_host_allowlist = {
            str(value).strip().lower() for value in siem_host_allowlist if str(value).strip()
        }
        self.uid_factory = uid_factory
        self.now_factory = now_factory
        self.commit = commit
        self.rollback = rollback

    def _actor(self, actor_uid: Any) -> str:
        actor = _uid(actor_uid, "actor_uid")
        if self.repository.users_available({actor}) != {actor}:
            raise ValueError("security actor is unavailable")
        return actor

    def _save(self, operation):
        try:
            result = operation()
            self.commit()
            return result
        except Exception:
            self.rollback()
            raise

    def _event(self, resource_type, resource_uid, action, actor_uid, detail=None):
        self.repository.add_event(
            resource_type, resource_uid, action, actor_uid, copy.deepcopy(detail or {})
        )

    def create_classification_profile(self, payload: Any, *, actor_uid: str):
        body = _closed(
            payload,
            {"code", "name", "business_domain_uid", "default_classification", "rules"},
            "classification profile",
        )
        actor = self._actor(actor_uid)
        code = _text(body.get("code"), "profile code", 120).upper()
        if not CODE_PATTERN.fullmatch(code):
            raise ValueError("classification profile code is invalid")
        rules = []
        for raw in _list(body.get("rules"), "classification rules", 1, 100):
            rule = _closed(raw, {"field_tokens", "category", "classification"}, "classification rule")
            tokens = sorted({
                _text(item, "field token", 60).casefold()
                for item in _list(rule.get("field_tokens"), "field_tokens", 1, 20)
            })
            if any(not re.fullmatch(r"[a-z0-9_-]+", item) for item in tokens):
                raise ValueError("field tokens must be simple identifiers")
            rules.append({
                "field_tokens": tokens,
                "category": _text(rule.get("category"), "category", 80),
                "classification": _classification(rule.get("classification")),
            })
        now = self.now_factory().isoformat()
        record = {
            "uid": self.uid_factory(), "code": code,
            "name": _text(body.get("name"), "profile name", 300),
            "business_domain_uid": _uid(body.get("business_domain_uid"), "business_domain_uid"),
            "default_classification": _classification(body.get("default_classification")),
            "rules": rules, "status": "active", "current_version": 1,
            "created_by": actor, "created_at": now, "updated_at": now,
        }

        def operation():
            result = self.repository.create_profile(record)
            self._event("classification_profile", record["uid"], "profile_created", actor, {"code": code})
            return result

        return self._save(operation)

    @staticmethod
    def _detect_field(profile: dict[str, Any], field: dict[str, Any]):
        field_name = _text(field.get("name"), "field name", 200)
        if not FIELD_PATTERN.fullmatch(field_name):
            raise ValueError("field name is invalid")
        values = [str(value)[:500] for value in _list(field.get("sample_values", []), "sample_values", 0, 20)]
        normalized = field_name.casefold().replace("-", "_").replace(".", "_")
        matches = []
        for rule in profile["rules"]:
            if any(token in normalized.split("_") or token in normalized for token in rule["field_tokens"]):
                matches.append((rule["classification"], rule["category"], "field_rule", None))
        detectors = (
            ("prc_id", PRC_ID_PATTERN, "personal_identifier", "highly_sensitive"),
            ("bank_card", BANK_CARD_PATTERN, "financial_account", "highly_sensitive"),
            ("phone", PHONE_PATTERN, "personal_contact", "sensitive"),
            ("email", EMAIL_PATTERN, "personal_contact", "sensitive"),
        )
        for value in values:
            for detector, pattern, category, level in detectors:
                found = pattern.search(value)
                if found:
                    sample = found.group(0)
                    matches.append((level, category, detector, sample))
        if not matches:
            return None
        level = max(matches, key=lambda item: CLASSIFICATION_RANK[item[0]])[0]
        categories = sorted({item[1] for item in matches})
        detector_codes = sorted({item[2] for item in matches})
        raw_matches = [item for item in matches if item[3] is not None]
        return {
            "field_name": field_name,
            "categories": categories,
            "proposed_classification": level,
            "detector_codes": detector_codes,
            "sample_fingerprints": sorted({_digest(item[3]) for item in raw_matches}),
            "masked_examples": sorted({_mask(item[3], item[2]) for item in raw_matches})[:3],
        }

    def scan_sensitive_sample(self, payload: Any, *, actor_uid: str):
        body = _closed(
            payload,
            {"profile_uid", "resource_type", "resource_uid", "business_domain_uid", "fields"},
            "sensitive sample scan",
        )
        actor = self._actor(actor_uid)
        profile = self.repository.get_profile(_uid(body.get("profile_uid"), "profile_uid"))
        if not profile or profile["status"] != "active":
            raise LookupError("active classification profile was not found")
        domain_uid = _uid(body.get("business_domain_uid"), "business_domain_uid")
        if profile["business_domain_uid"] != domain_uid:
            raise PermissionError("classification profile domain does not match")
        fields = _list(body.get("fields"), "fields", 1, 100)
        now = self.now_factory().isoformat()
        scan_uid = self.uid_factory()
        findings = []
        for field in fields:
            normalized = self._detect_field(profile, field)
            if not normalized:
                continue
            findings.append({
                "uid": self.uid_factory(), "scan_uid": scan_uid,
                **normalized, "status": "pending_review", "final_classification": None,
                "review_reason": None, "reviewed_by": None, "reviewed_at": None,
                "current_version": 1, "created_at": now,
            })
        scan = {
            "uid": scan_uid, "profile_uid": profile["uid"],
            "resource_type": _text(body.get("resource_type"), "resource_type", 80),
            "resource_uid": _text(body.get("resource_uid"), "resource_uid", 200),
            "business_domain_uid": domain_uid, "field_count": len(fields),
            "finding_count": len(findings), "sample_retained": False,
            "created_by": actor, "created_at": now,
        }

        def operation():
            result = self.repository.create_scan(scan, findings)
            self._event(
                "classification_scan", scan_uid, "sample_scanned", actor,
                {"finding_count": len(findings), "sample_retained": False},
            )
            return result

        return self._save(operation)

    def review_classification_finding(
        self, finding_uid: str, payload: Any, *, expected_version: int, actor_uid: str
    ):
        body = _closed(payload, {"decision", "final_classification", "reason"}, "classification review")
        actor = self._actor(actor_uid)
        finding = self.repository.get_finding(_uid(finding_uid, "finding_uid"))
        if not finding:
            raise LookupError("classification finding was not found")
        if finding["status"] != "pending_review":
            raise RuntimeError("classification finding is not pending review")
        scan = self.repository.get_scan(finding["scan_uid"]) if hasattr(self.repository, "get_scan") else None
        creator = scan.get("created_by") if scan else None
        if actor == creator or (creator is None and actor == finding.get("created_by")):
            raise PermissionError("classification requires an independent reviewer")
        # Memory repositories keep the creator on the scan rather than the finding.
        if (
            creator is None
            and hasattr(self.repository, "scans")
            and actor == self.repository.scans[finding["scan_uid"]]["created_by"]
        ):
            raise PermissionError("classification requires an independent reviewer")
        decision = _text(body.get("decision"), "decision", 20)
        if decision not in {"confirm", "dismiss"}:
            raise ValueError("unsupported classification review decision")
        updated = {
            **finding,
            "status": "confirmed" if decision == "confirm" else "dismissed",
            "final_classification": (
                _classification(body.get("final_classification")) if decision == "confirm" else None
            ),
            "review_reason": _text(body.get("reason"), "review reason", 1000),
            "reviewed_by": actor, "reviewed_at": self.now_factory().isoformat(),
        }

        def operation():
            result = self.repository.update_finding(updated, int(expected_version))
            self._event("classification_finding", finding["uid"], f"finding_{updated['status']}", actor, {"classification": updated["final_classification"]})
            return result

        return self._save(operation)

    def create_access_policy(self, payload: Any, *, actor_uid: str):
        body = _closed(
            payload,
            {
                "code", "name", "business_domain_uid", "subject_user_uids",
                "subject_roles", "purposes", "environments", "actions",
                "max_classification", "allowed_fields", "expires_at", "review_due_at",
            },
            "access policy",
        )
        actor = self._actor(actor_uid)
        code = _text(body.get("code"), "policy code", 120).upper()
        if not CODE_PATTERN.fullmatch(code):
            raise ValueError("access policy code is invalid")
        users = sorted({_uid(value, "subject_user_uid") for value in _list(body.get("subject_user_uids", []), "subject_user_uids", 0, 100)})
        roles = sorted({_text(value, "subject role", 80) for value in _list(body.get("subject_roles", []), "subject_roles", 0, 50)})
        if not users and not roles:
            raise ValueError("access policy requires a user or role subject")
        if users and self.repository.users_available(users) != set(users):
            raise ValueError("access policy contains unavailable users")
        if any(not ROLE_PATTERN.fullmatch(role) for role in roles):
            raise ValueError("access policy role is invalid")
        purposes = sorted({_text(value, "purpose", 100) for value in _list(body.get("purposes"), "purposes", 1, 50)})
        environments = sorted({_text(value, "environment", 30) for value in _list(body.get("environments"), "environments", 1, 10)})
        actions = sorted({_text(value, "action", 20) for value in _list(body.get("actions"), "actions", 1, 10)})
        if not set(environments) <= ENVIRONMENTS or not set(actions) <= ACCESS_ACTIONS:
            raise ValueError("unsupported access environment or action")
        fields = sorted({_text(value, "allowed field", 200) for value in _list(body.get("allowed_fields"), "allowed_fields", 1, 500)})
        if any(value != "*" and not FIELD_PATTERN.fullmatch(value) for value in fields):
            raise ValueError("allowed field is invalid")
        now = self.now_factory().astimezone(UTC)
        expires = _time(body.get("expires_at"), "expires_at")
        review = _time(body.get("review_due_at"), "review_due_at")
        if review <= now or expires <= review:
            raise ValueError("access policy review and expiry dates are invalid")
        record = {
            "uid": self.uid_factory(), "code": code,
            "name": _text(body.get("name"), "policy name", 300),
            "business_domain_uid": _uid(body.get("business_domain_uid"), "business_domain_uid"),
            "subject_user_uids": users, "subject_roles": roles, "purposes": purposes,
            "environments": environments, "actions": actions,
            "max_classification": _classification(body.get("max_classification"), "max_classification"),
            "allowed_fields": fields, "expires_at": expires.isoformat(),
            "review_due_at": review.isoformat(), "status": "active", "current_version": 1,
            "created_by": actor, "created_at": now.isoformat(), "updated_at": now.isoformat(),
        }

        def operation():
            result = self.repository.create_access_policy(record)
            self._event("access_policy", record["uid"], "access_policy_created", actor, {"code": code})
            return result

        return self._save(operation)

    @staticmethod
    def _policy_matches(policy: dict[str, Any], context: dict[str, Any], now: datetime, *, check_fields=True):
        subject = (
            context["user_uid"] in policy["subject_user_uids"]
            or bool(set(context["roles"]) & set(policy["subject_roles"]))
        )
        if not subject:
            return False
        if policy["business_domain_uid"] != context["business_domain_uid"]:
            return False
        if context["purpose"] not in policy["purposes"]:
            return False
        if context["environment"] not in policy["environments"]:
            return False
        if context["action"] not in policy["actions"]:
            return False
        if CLASSIFICATION_RANK[context["classification"]] > CLASSIFICATION_RANK[policy["max_classification"]]:
            return False
        if _time(policy["expires_at"], "expires_at") <= now or _time(policy["review_due_at"], "review_due_at") <= now:
            return False
        if check_fields and "*" not in policy["allowed_fields"]:
            return set(context["requested_fields"]) <= set(policy["allowed_fields"])
        return True

    def evaluate_access(self, payload: Any):
        body = _closed(
            payload,
            {
                "user_uid", "roles", "business_domain_uid", "purpose", "environment",
                "action", "resource_uid", "classification", "requested_fields",
            },
            "access decision",
        )
        user_uid = _uid(body.get("user_uid"), "user_uid")
        if self.repository.users_available({user_uid}) != {user_uid}:
            raise ValueError("access user is unavailable")
        roles = sorted({_text(value, "role", 80) for value in _list(body.get("roles"), "roles", 1, 50)})
        fields = sorted({_text(value, "requested field", 200) for value in _list(body.get("requested_fields"), "requested_fields", 1, 500)})
        context = {
            "user_uid": user_uid, "roles": roles,
            "business_domain_uid": _uid(body.get("business_domain_uid"), "business_domain_uid"),
            "purpose": _text(body.get("purpose"), "purpose", 100),
            "environment": _text(body.get("environment"), "environment", 30),
            "action": _text(body.get("action"), "action", 20),
            "resource_uid": _text(body.get("resource_uid"), "resource_uid", 200),
            "classification": _classification(body.get("classification")),
            "requested_fields": fields,
        }
        if context["environment"] not in ENVIRONMENTS or context["action"] not in ACCESS_ACTIONS:
            raise ValueError("unsupported access environment or action")
        now = self.now_factory().astimezone(UTC)
        policies = self.repository.matching_access_policies(**context)
        policy = next((item for item in policies if self._policy_matches(item, context, now)), None)
        scope_policy = next((item for item in policies if self._policy_matches(item, context, now, check_fields=False)), None)
        if policy:
            decision, reason = "authorized", "policy_allowed"
        elif scope_policy:
            decision, reason = "denied", "field_minimization_denied"
        else:
            decision, reason = "denied", "default_deny"
        record = {
            "uid": self.uid_factory(), **context,
            "policy_uid": policy["uid"] if policy else None,
            "decision": decision, "reason_code": reason,
            "decided_at": self.now_factory().isoformat(),
        }

        def operation():
            result = self.repository.create_access_decision(record)
            self._event("access_decision", record["uid"], f"access_{decision}", user_uid, {"reason_code": reason, "policy_uid": record["policy_uid"]})
            return result

        return self._save(operation)

    def submit_egress_request(self, payload: Any, *, actor_uid: str):
        body = _closed(
            payload,
            {
                "business_domain_uid", "resource_uid", "classification", "purpose",
                "environment", "requested_fields", "minimized_fields", "masking_applied",
                "destination_zone", "expires_at", "workflow_uid",
            },
            "egress request",
        )
        actor = self._actor(actor_uid)
        requested = sorted({_text(value, "requested field", 200) for value in _list(body.get("requested_fields"), "requested_fields", 1, 500)})
        minimized = sorted({_text(value, "minimized field", 200) for value in _list(body.get("minimized_fields"), "minimized_fields", 1, 500)})
        if not set(minimized) <= set(requested):
            raise ValueError("minimized fields must be a subset of requested fields")
        masking = body.get("masking_applied")
        if not isinstance(masking, bool):
            raise ValueError("masking_applied must be boolean")
        classification = _classification(body.get("classification"))
        if classification in {"sensitive", "highly_sensitive"} and not masking:
            raise ValueError("sensitive egress requires masking")
        environment = _text(body.get("environment"), "environment", 30)
        if environment not in ENVIRONMENTS:
            raise ValueError("unsupported egress environment")
        now = self.now_factory().astimezone(UTC)
        expires = _time(body.get("expires_at"), "expires_at")
        if expires <= now or expires > now + timedelta(days=30):
            raise ValueError("egress expiry must be within 30 days")
        high = classification == "highly_sensitive"
        record = {
            "uid": self.uid_factory(),
            "business_domain_uid": _uid(body.get("business_domain_uid"), "business_domain_uid"),
            "resource_uid": _text(body.get("resource_uid"), "resource_uid", 200),
            "classification": classification,
            "purpose": _text(body.get("purpose"), "purpose", 300),
            "environment": environment, "requested_fields": requested,
            "minimized_fields": minimized, "approved_fields": [],
            "masking_applied": masking,
            "destination_zone": _text(body.get("destination_zone"), "destination_zone", 100),
            "expires_at": expires.isoformat(),
            "status": "denied" if high else "pending_approval",
            "reason_code": "highly_sensitive_egress_disabled" if high else "approval_required",
            "approval_task_uid": None, "current_version": 1,
            "created_by": actor, "created_at": now.isoformat(), "updated_at": now.isoformat(),
        }
        if not high:
            workflow_uid = _uid(body.get("workflow_uid"), "workflow_uid")
            task = self.approval_gateway.create_egress_task(record, workflow_uid, actor)
            record["approval_task_uid"] = task["uid"]

        def operation():
            result = self.repository.create_egress_request(record)
            self._event("egress_request", record["uid"], f"egress_{record['status']}", actor, {"classification": classification, "reason_code": record["reason_code"]})
            return result

        return self._save(operation)

    def reconcile_egress_request(self, request_uid: str, *, expected_version: int, actor_uid: str):
        actor = self._actor(actor_uid)
        record = self.repository.get_egress_request(_uid(request_uid, "request_uid"))
        if not record:
            raise LookupError("egress request was not found")
        if record["status"] != "pending_approval":
            raise RuntimeError("egress request is not pending approval")
        task = self.approval_gateway.get_task(record["approval_task_uid"])
        if not task or task["status"] not in {"approved", "rejected"}:
            raise RuntimeError("egress approval has no final decision")
        approved = task["status"] == "approved"
        updated = {
            **record,
            "status": "authorized_until_expiry" if approved else "denied",
            "reason_code": "approval_granted" if approved else "approval_rejected",
            "approved_fields": record["minimized_fields"] if approved else [],
            "updated_at": self.now_factory().isoformat(),
        }

        def operation():
            result = self.repository.update_egress_request(updated, int(expected_version))
            self._event("egress_request", record["uid"], f"egress_{updated['status']}", actor, {"approved_field_count": len(updated["approved_fields"])})
            return result

        return self._save(operation)

    def create_retention_policy(self, payload: Any, *, actor_uid: str):
        body = _closed(
            payload,
            {"code", "name", "evidence_type", "retention_days", "archive_mode", "disposition_action"},
            "retention policy",
        )
        actor = self._actor(actor_uid)
        code = _text(body.get("code"), "retention code", 120).upper()
        if not CODE_PATTERN.fullmatch(code):
            raise ValueError("retention code is invalid")
        evidence_type = _text(body.get("evidence_type"), "evidence_type", 60)
        archive_mode = _text(body.get("archive_mode"), "archive_mode", 40)
        disposition = _text(body.get("disposition_action"), "disposition_action", 40)
        if evidence_type not in RETENTION_EVIDENCE_TYPES or archive_mode not in ARCHIVE_MODES or disposition not in DISPOSITION_ACTIONS:
            raise ValueError("unsupported retention policy option")
        try:
            days = int(body.get("retention_days"))
        except (TypeError, ValueError) as error:
            raise ValueError("retention_days must be an integer") from error
        if days < 30 or days > 36500:
            raise ValueError("retention_days must be between 30 and 36500")
        now = self.now_factory().isoformat()
        record = {
            "uid": self.uid_factory(), "code": code,
            "name": _text(body.get("name"), "retention name", 300),
            "evidence_type": evidence_type, "retention_days": days,
            "archive_mode": archive_mode, "disposition_action": disposition,
            "automatic_deletion": False, "status": "active", "current_version": 1,
            "created_by": actor, "created_at": now, "updated_at": now,
        }

        def operation():
            result = self.repository.create_retention_policy(record)
            self._event("retention_policy", record["uid"], "retention_policy_created", actor, {"evidence_type": evidence_type, "automatic_deletion": False})
            return result

        return self._save(operation)

    def retention_candidates(self, *, as_of: datetime, limit: int):
        instant = _time(as_of, "as_of")
        size = int(limit)
        if size < 1 or size > 1000:
            raise ValueError("retention candidate limit must be between 1 and 1000")
        return self.repository.retention_candidates(instant, size)

    def create_siem_sink(self, payload: Any, *, actor_uid: str):
        body = _closed(payload, {"name", "sink_type", "endpoint", "categories"}, "SIEM sink")
        actor = self._actor(actor_uid)
        sink_type = _text(body.get("sink_type"), "sink_type", 30)
        endpoint = _text(body.get("endpoint"), "endpoint", 500)
        parsed = urlsplit(endpoint)
        required_scheme = "https" if sink_type == "webhook" else "tls"
        if sink_type not in {"webhook", "syslog_tls"} or parsed.scheme != required_scheme:
            raise ValueError("SIEM sink requires HTTPS webhook or TLS syslog")
        if parsed.username or parsed.password or parsed.query or parsed.fragment:
            raise ValueError("SIEM endpoint must not contain credentials, query or fragment")
        host = str(parsed.hostname or "").lower()
        if not host or host not in self.siem_host_allowlist:
            raise ValueError("SIEM endpoint host is outside the allowlist")
        categories = sorted({_text(value, "SIEM category", 40) for value in _list(body.get("categories"), "categories", 1, 20)})
        if not set(categories) <= SIEM_CATEGORIES:
            raise ValueError("unsupported SIEM audit category")
        now = self.now_factory().isoformat()
        record = {
            "uid": self.uid_factory(), "name": _text(body.get("name"), "sink name", 300),
            "sink_type": sink_type, "endpoint": endpoint, "endpoint_host": host,
            "categories": categories, "status": "active", "current_version": 1,
            "created_by": actor, "created_at": now, "updated_at": now,
        }

        def operation():
            result = self.repository.create_siem_sink(record)
            self._event("siem_sink", record["uid"], "siem_sink_created", actor, {"sink_type": sink_type, "endpoint_host": host})
            return result

        return self._save(operation)

    def dispatch_siem_events(self, sink_uid: str, payload: Any, *, actor_uid: str):
        body = _closed(payload, {"period_start", "period_end", "limit"}, "SIEM delivery")
        actor = self._actor(actor_uid)
        sink = self.repository.get_siem_sink(_uid(sink_uid, "sink_uid"))
        if not sink or sink["status"] != "active":
            raise LookupError("active SIEM sink was not found")
        start = _time(body.get("period_start"), "period_start")
        end = _time(body.get("period_end"), "period_end")
        limit = int(body.get("limit", 100))
        if end <= start or limit < 1 or limit > 1000:
            raise ValueError("SIEM delivery window or limit is invalid")
        events = self.repository.fetch_siem_events(
            categories=sink["categories"], period_start=start, period_end=end, limit=limit
        )
        envelope = {
            "schema": "dataops.security.audit.v1",
            "period_start": start.isoformat(), "period_end": end.isoformat(),
            "events": events,
        }
        result = self.siem_transport.deliver(sink, envelope)
        status = result.get("status")
        if status not in {"delivered", "failed"}:
            raise RuntimeError("SIEM transport returned an invalid status")
        now = self.now_factory().isoformat()
        record = {
            "uid": self.uid_factory(), "sink_uid": sink["uid"],
            "period_start": start.isoformat(), "period_end": end.isoformat(),
            "event_count": len(events), "payload_digest": _digest(envelope),
            "status": status, "remote_ref": _optional_text(result.get("remote_ref"), "remote_ref", 300),
            "error_code": _optional_text(result.get("error_code"), "error_code", 80),
            "created_by": actor, "created_at": now,
        }

        def operation():
            saved = self.repository.create_siem_delivery(record)
            self._event("siem_delivery", saved["uid"], f"siem_{status}", actor, {"event_count": len(events), "payload_digest": record["payload_digest"]})
            return saved

        return self._save(operation)

    def register_sbom(self, payload: Any, *, actor_uid: str):
        body = _closed(
            payload,
            {"artifact_name", "artifact_version", "artifact_type", "source_ref", "document"},
            "SBOM registration",
        )
        actor = self._actor(actor_uid)
        document = body.get("document")
        if not isinstance(document, dict):
            raise ValueError("SBOM document must be an object")
        if document.get("bomFormat") != "CycloneDX" or document.get("specVersion") != "1.5":
            raise ValueError("SBOM must use CycloneDX 1.5")
        components = []
        for raw in _list(document.get("components", []), "SBOM components", 0, 10000):
            component = _closed(
                raw,
                {
                    "type", "name", "version", "purl", "bom-ref", "licenses",
                    "externalReferences", "properties", "group", "supplier", "publisher",
                    "author", "description", "hashes", "scope", "copyright",
                },
                "SBOM component",
            )
            components.append({
                "type": _text(component.get("type"), "component type", 50),
                "name": _text(component.get("name"), "component name", 300),
                "version": _optional_text(component.get("version"), "component version", 200),
                "purl": _optional_text(component.get("purl"), "component purl", 500),
            })
        now = self.now_factory().isoformat()
        record = {
            "uid": self.uid_factory(),
            "artifact_name": _text(body.get("artifact_name"), "artifact name", 300),
            "artifact_version": _text(body.get("artifact_version"), "artifact version", 120),
            "artifact_type": _text(body.get("artifact_type"), "artifact type", 50),
            "source_ref": _text(body.get("source_ref"), "source_ref", 500),
            "format": "CycloneDX", "spec_version": "1.5",
            "document_digest": _digest(document), "component_count": len(components),
            "components": sorted(components, key=lambda item: (item["name"], item.get("version") or "")),
            "created_by": actor, "created_at": now,
        }

        def operation():
            result = self.repository.create_sbom(record)
            self._event("sbom", record["uid"], "sbom_registered", actor, {"artifact_name": record["artifact_name"], "component_count": len(components), "document_digest": record["document_digest"]})
            return result

        return self._save(operation)

    def ingest_vulnerabilities(self, sbom_uid: str, payload: Any, *, actor_uid: str):
        body = _closed(payload, {"scanner", "scan_ref", "findings"}, "vulnerability import")
        actor = self._actor(actor_uid)
        sbom = self.repository.get_sbom(_uid(sbom_uid, "sbom_uid"))
        if not sbom:
            raise LookupError("SBOM was not found")
        scanner = _text(body.get("scanner"), "scanner", 100)
        scan_ref = _text(body.get("scan_ref"), "scan_ref", 500)
        now = self.now_factory().isoformat()
        records = []
        for raw in _list(body.get("findings"), "vulnerability findings", 1, 5000):
            finding = _closed(
                raw,
                {"external_id", "severity", "component_name", "installed_version", "fixed_version", "title"},
                "vulnerability finding",
            )
            severity = _text(finding.get("severity"), "severity", 20).lower()
            if severity not in SEVERITIES:
                raise ValueError("unsupported vulnerability severity")
            records.append({
                "uid": self.uid_factory(), "sbom_uid": sbom["uid"],
                "scanner": scanner, "scan_ref": scan_ref,
                "external_id": _text(finding.get("external_id"), "external_id", 120),
                "severity": severity,
                "component_name": _text(finding.get("component_name"), "component_name", 300),
                "installed_version": _text(finding.get("installed_version"), "installed_version", 200),
                "fixed_version": _optional_text(finding.get("fixed_version"), "fixed_version", 200),
                "title": _text(finding.get("title"), "title", 500),
                "status": "open", "assignee_uid": None, "due_at": None,
                "resolution_type": None, "resolved_version": None, "resolution": None,
                "evidence_refs": [], "resolved_by": None, "resolved_at": None,
                "closed_by": None, "closed_at": None, "close_reason": None,
                "current_version": 1, "created_by": actor, "created_at": now, "updated_at": now,
            })

        def operation():
            result = self.repository.upsert_vulnerabilities(sbom["uid"], records)
            self._event("sbom", sbom["uid"], "vulnerabilities_imported", actor, {"scanner": scanner, "finding_count": len(result)})
            return result

        return self._save(operation)

    def assign_vulnerability(
        self, finding_uid: str, payload: Any, *, expected_version: int, actor_uid: str
    ):
        body = _closed(payload, {"assignee_uid", "due_at"}, "vulnerability assignment")
        actor = self._actor(actor_uid)
        finding = self.repository.get_vulnerability(_uid(finding_uid, "finding_uid"))
        if not finding:
            raise LookupError("vulnerability was not found")
        if finding["status"] not in {"open", "triaged", "in_progress"}:
            raise RuntimeError("vulnerability cannot be assigned")
        assignee = _uid(body.get("assignee_uid"), "assignee_uid")
        if self.repository.users_available({assignee}) != {assignee}:
            raise ValueError("vulnerability assignee is unavailable")
        due_at = _time(body.get("due_at"), "due_at")
        if due_at <= self.now_factory().astimezone(UTC):
            raise ValueError("vulnerability due date must be in the future")
        updated = {
            **finding, "status": "in_progress", "assignee_uid": assignee,
            "due_at": due_at.isoformat(), "updated_at": self.now_factory().isoformat(),
        }
        return self._update_vulnerability(updated, expected_version, "vulnerability_assigned", actor, {"assignee_uid": assignee})

    def resolve_vulnerability(
        self, finding_uid: str, payload: Any, *, expected_version: int, actor_uid: str
    ):
        body = _closed(
            payload,
            {"resolution_type", "resolved_version", "resolution", "evidence_refs"},
            "vulnerability resolution",
        )
        actor = self._actor(actor_uid)
        finding = self.repository.get_vulnerability(_uid(finding_uid, "finding_uid"))
        if not finding:
            raise LookupError("vulnerability was not found")
        if finding["status"] != "in_progress" or actor != finding["assignee_uid"]:
            raise PermissionError("only the assigned owner can resolve an in-progress vulnerability")
        resolution_type = _text(body.get("resolution_type"), "resolution_type", 30)
        if resolution_type not in RESOLUTION_TYPES:
            raise ValueError("unsupported vulnerability resolution type")
        resolved_version = _optional_text(body.get("resolved_version"), "resolved_version", 200)
        if resolution_type == "patched" and not resolved_version:
            raise ValueError("patched vulnerabilities require a resolved version")
        updated = {
            **finding, "status": "resolved", "resolution_type": resolution_type,
            "resolved_version": resolved_version,
            "resolution": _text(body.get("resolution"), "resolution", 2000),
            "evidence_refs": _normalize_evidence(body.get("evidence_refs")),
            "resolved_by": actor, "resolved_at": self.now_factory().isoformat(),
            "updated_at": self.now_factory().isoformat(),
        }
        return self._update_vulnerability(updated, expected_version, "vulnerability_resolved", actor, {"resolution_type": resolution_type})

    def close_vulnerability(
        self, finding_uid: str, payload: Any, *, expected_version: int, actor_uid: str
    ):
        body = _closed(payload, {"reason"}, "vulnerability closure")
        actor = self._actor(actor_uid)
        finding = self.repository.get_vulnerability(_uid(finding_uid, "finding_uid"))
        if not finding:
            raise LookupError("vulnerability was not found")
        if finding["status"] != "resolved":
            raise RuntimeError("only resolved vulnerabilities can be closed")
        if actor in {finding.get("assignee_uid"), finding.get("resolved_by")}:
            raise PermissionError("vulnerability closure requires an independent reviewer")
        updated = {
            **finding, "status": "closed", "closed_by": actor,
            "closed_at": self.now_factory().isoformat(),
            "close_reason": _text(body.get("reason"), "close reason", 1000),
            "updated_at": self.now_factory().isoformat(),
        }
        return self._update_vulnerability(updated, expected_version, "vulnerability_closed", actor, {"resolution_type": finding["resolution_type"]})

    def _update_vulnerability(self, record, expected_version, action, actor, detail):
        def operation():
            result = self.repository.update_vulnerability(record, int(expected_version), action, actor)
            self._event("vulnerability", record["uid"], action, actor, detail)
            return result

        return self._save(operation)

    def list_profiles(self, **filters):
        return self.repository.list_profiles(**filters)

    def list_classification_scans(self, **filters):
        return self.repository.list_scans(**filters)

    def list_classification_findings(self, **filters):
        return self.repository.list_findings(**filters)

    def list_access_policies(self, **filters):
        return self.repository.list_access_policies(**filters)

    def list_access_decisions(self, **filters):
        return self.repository.list_access_decisions(**filters)

    def list_egress_requests(self, **filters):
        return self.repository.list_egress_requests(**filters)

    def list_retention_policies(self):
        return self.repository.list_retention_policies()

    def list_siem_sinks(self):
        return self.repository.list_siem_sinks()

    def list_siem_deliveries(self, **filters):
        return self.repository.list_siem_deliveries(**filters)

    def list_sboms(self):
        return self.repository.list_sboms()

    def list_vulnerabilities(self, **filters):
        return self.repository.list_vulnerabilities(**filters)

    def dashboard(self):
        return self.repository.dashboard()