"""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"(? 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()