DataOps-platform/tests/security/test_security_governance.py

from __future__ import annotations

from copy import deepcopy
from datetime import UTC, datetime, timedelta

import pytest

from app.core.system.security_governance import SecurityGovernanceService

OWNER = "01900000-0000-7000-8000-000000010001"
REVIEWER = "01900000-0000-7000-8000-000000010002"
USER = "01900000-0000-7000-8000-000000010003"
DOMAIN = "01900000-0000-7000-8000-000000010101"
OTHER_DOMAIN = "01900000-0000-7000-8000-000000010102"
WORKFLOW = "01900000-0000-7000-8000-000000010201"


class MemorySecurityRepository:
    def __init__(self):
        self.users = {OWNER, REVIEWER, USER}
        self.profiles = {}
        self.scans = {}
        self.findings = {}
        self.policies = {}
        self.decisions = {}
        self.egress = {}
        self.retention = {}
        self.sinks = {}
        self.deliveries = {}
        self.sboms = {}
        self.vulnerabilities = {}
        self.events = []

    def users_available(self, values):
        return set(values) & self.users

    def create_profile(self, record):
        self.profiles[record["uid"]] = deepcopy(record)
        return deepcopy(record)

    def get_profile(self, uid):
        return deepcopy(self.profiles.get(uid))

    def create_scan(self, scan, findings):
        self.scans[scan["uid"]] = deepcopy(scan)
        for finding in findings:
            self.findings[finding["uid"]] = deepcopy(finding)
        return {**deepcopy(scan), "findings": deepcopy(findings)}

    def get_finding(self, uid):
        return deepcopy(self.findings.get(uid))

    def update_finding(self, record, expected_version):
        current = self.findings[record["uid"]]
        if current["current_version"] != expected_version:
            raise RuntimeError("classification finding version conflict")
        saved = deepcopy(record)
        saved["current_version"] += 1
        self.findings[record["uid"]] = saved
        return deepcopy(saved)

    def create_access_policy(self, record):
        self.policies[record["uid"]] = deepcopy(record)
        return deepcopy(record)

    def matching_access_policies(self, **context):
        return [deepcopy(item) for item in self.policies.values() if item["status"] == "active"]

    def create_access_decision(self, record):
        self.decisions[record["uid"]] = deepcopy(record)
        return deepcopy(record)

    def create_egress_request(self, record):
        self.egress[record["uid"]] = deepcopy(record)
        return deepcopy(record)

    def get_egress_request(self, uid):
        return deepcopy(self.egress.get(uid))

    def update_egress_request(self, record, expected_version):
        if self.egress[record["uid"]]["current_version"] != expected_version:
            raise RuntimeError("egress request version conflict")
        saved = deepcopy(record)
        saved["current_version"] += 1
        self.egress[record["uid"]] = saved
        return deepcopy(saved)

    def create_retention_policy(self, record):
        self.retention[record["uid"]] = deepcopy(record)
        return deepcopy(record)

    def retention_candidates(self, as_of, limit):
        return [{"evidence_type": "access_decision", "candidate_count": 2, "automatic_deletion": False}]

    def create_siem_sink(self, record):
        self.sinks[record["uid"]] = deepcopy(record)
        return deepcopy(record)

    def get_siem_sink(self, uid):
        return deepcopy(self.sinks.get(uid))

    def fetch_siem_events(self, **filters):
        return [{"event_uid": "event-1", "category": "authorization", "action": "denied", "safe_detail": {}}]

    def create_siem_delivery(self, record):
        self.deliveries[record["uid"]] = deepcopy(record)
        return deepcopy(record)

    def create_sbom(self, record):
        self.sboms[record["uid"]] = deepcopy(record)
        return deepcopy(record)

    def get_sbom(self, uid):
        return deepcopy(self.sboms.get(uid))

    def upsert_vulnerabilities(self, sbom_uid, records):
        for record in records:
            self.vulnerabilities[record["uid"]] = deepcopy(record)
        return deepcopy(records)

    def get_vulnerability(self, uid):
        return deepcopy(self.vulnerabilities.get(uid))

    def update_vulnerability(self, record, expected_version, action, actor_uid):
        current = self.vulnerabilities[record["uid"]]
        if current["current_version"] != expected_version:
            raise RuntimeError("vulnerability version conflict")
        saved = deepcopy(record)
        saved["current_version"] += 1
        self.vulnerabilities[record["uid"]] = saved
        return deepcopy(saved)

    def add_event(self, resource_type, resource_uid, action, actor_uid, safe_detail):
        self.events.append({
            "resource_type": resource_type, "resource_uid": resource_uid,
            "action": action, "actor_uid": actor_uid, "safe_detail": deepcopy(safe_detail),
        })

    def dashboard(self):
        return {
            "pending_classification_reviews": sum(item["status"] == "pending_review" for item in self.findings.values()),
            "denied_access_count": sum(item["decision"] == "denied" for item in self.decisions.values()),
            "open_vulnerability_count": sum(item["status"] != "closed" for item in self.vulnerabilities.values()),
        }


class FakeApprovalGateway:
    def __init__(self):
        self.tasks = {}

    def create_egress_task(self, request_record, workflow_uid, actor_uid):
        task = {"uid": f"01900000-0000-7000-8000-{len(self.tasks) + 1:012d}", "status": "pending"}
        self.tasks[task["uid"]] = task
        return deepcopy(task)

    def get_task(self, uid):
        return deepcopy(self.tasks.get(uid))


class FakeSiemTransport:
    def __init__(self):
        self.sent = []

    def deliver(self, sink, envelope):
        self.sent.append((deepcopy(sink), deepcopy(envelope)))
        return {"status": "delivered", "remote_ref": "siem-accepted-1"}


@pytest.fixture()
def security():
    repository = MemorySecurityRepository()
    approvals = FakeApprovalGateway()
    transport = FakeSiemTransport()
    ids = iter(f"01900000-0000-7000-8000-{value:012d}" for value in range(300, 900))
    now = datetime(2026, 8, 2, 8, 0, tzinfo=UTC)
    service = SecurityGovernanceService(
        repository,
        approval_gateway=approvals,
        siem_transport=transport,
        siem_host_allowlist={"siem.example.internal"},
        uid_factory=lambda: next(ids),
        now_factory=lambda: now,
    )
    return service, repository, approvals, transport, now


def profile_payload():
    return {
        "code": "MATERIAL_SECURITY",
        "name": "备品备件敏感识别规则",
        "business_domain_uid": DOMAIN,
        "default_classification": "internal",
        "rules": [
            {"field_tokens": ["phone", "mobile"], "category": "personal_contact", "classification": "sensitive"},
            {"field_tokens": ["bank", "account"], "category": "financial_account", "classification": "highly_sensitive"},
        ],
    }


def test_second_domain_sensitive_sample_is_identified_without_persisting_raw_values(security):
    service, repository, _approvals, _transport, _now = security
    profile = service.create_classification_profile(profile_payload(), actor_uid=OWNER)
    scan = service.scan_sensitive_sample({
        "profile_uid": profile["uid"],
        "resource_type": "active_metadata_asset",
        "resource_uid": "01900000-0000-7000-8000-000000010301",
        "business_domain_uid": DOMAIN,
        "fields": [
            {"name": "material_code", "sample_values": ["MAT-0001"]},
            {"name": "supplier_contact_phone", "sample_values": ["13800138000"]},
            {"name": "supplier_bank_account", "sample_values": ["6222021234567890"]},
        ],
    }, actor_uid=OWNER)
    assert {item["proposed_classification"] for item in scan["findings"]} == {"sensitive", "highly_sensitive"}
    serialized = repr({"scan": repository.scans, "findings": repository.findings})
    assert "13800138000" not in serialized
    assert "6222021234567890" not in serialized
    assert "MAT-0001" not in serialized
    assert all(item["status"] == "pending_review" for item in scan["findings"])


def test_classification_requires_independent_human_review(security):
    service, _repository, _approvals, _transport, _now = security
    profile = service.create_classification_profile(profile_payload(), actor_uid=OWNER)
    scan = service.scan_sensitive_sample({
        "profile_uid": profile["uid"], "resource_type": "file_field",
        "resource_uid": "01900000-0000-7000-8000-000000010302",
        "business_domain_uid": DOMAIN,
        "fields": [{"name": "supplier_contact_phone", "sample_values": ["13800138000"]}],
    }, actor_uid=OWNER)
    finding = scan["findings"][0]
    with pytest.raises(PermissionError, match="independent reviewer"):
        service.review_classification_finding(
            finding["uid"], {"decision": "confirm", "final_classification": "sensitive", "reason": "rule confirmed"},
            expected_version=1, actor_uid=OWNER,
        )
    reviewed = service.review_classification_finding(
        finding["uid"], {"decision": "confirm", "final_classification": "sensitive", "reason": "supplier contact is personal information"},
        expected_version=1, actor_uid=REVIEWER,
    )
    assert reviewed["status"] == "confirmed"
    assert reviewed["reviewed_by"] == REVIEWER


def test_access_policy_binds_identity_role_domain_purpose_environment_fields_and_expiry(security):
    service, _repository, _approvals, _transport, now = security
    service.create_access_policy({
        "code": "MATERIAL_STEWARD_READ",
        "name": "物料管理员受控读取",
        "business_domain_uid": DOMAIN,
        "subject_user_uids": [USER],
        "subject_roles": ["editor"],
        "purposes": ["material_quality"],
        "environments": ["test"],
        "actions": ["read", "use"],
        "max_classification": "sensitive",
        "allowed_fields": ["material_code", "supplier_contact_phone"],
        "expires_at": (now + timedelta(days=30)).isoformat(),
        "review_due_at": (now + timedelta(days=15)).isoformat(),
    }, actor_uid=OWNER)
    allowed = service.evaluate_access({
        "user_uid": USER, "roles": ["editor"], "business_domain_uid": DOMAIN,
        "purpose": "material_quality", "environment": "test", "action": "read",
        "resource_uid": "01900000-0000-7000-8000-000000010401",
        "classification": "sensitive", "requested_fields": ["material_code"],
    })
    denied = service.evaluate_access({
        "user_uid": USER, "roles": ["editor"], "business_domain_uid": OTHER_DOMAIN,
        "purpose": "material_quality", "environment": "test", "action": "read",
        "resource_uid": "01900000-0000-7000-8000-000000010401",
        "classification": "sensitive", "requested_fields": ["material_code"],
    })
    excessive = service.evaluate_access({
        "user_uid": USER, "roles": ["editor"], "business_domain_uid": DOMAIN,
        "purpose": "material_quality", "environment": "test", "action": "read",
        "resource_uid": "01900000-0000-7000-8000-000000010401",
        "classification": "sensitive", "requested_fields": ["supplier_bank_account"],
    })
    assert allowed["decision"] == "authorized"
    assert denied["reason_code"] == "default_deny"
    assert excessive["reason_code"] == "field_minimization_denied"


def test_highly_sensitive_egress_is_denied_and_sensitive_egress_requires_approval(security):
    service, _repository, approvals, _transport, now = security
    high = service.submit_egress_request({
        "business_domain_uid": DOMAIN, "resource_uid": "01900000-0000-7000-8000-000000010501",
        "classification": "highly_sensitive", "purpose": "supplier_settlement",
        "environment": "production", "requested_fields": ["supplier_bank_account"],
        "minimized_fields": ["supplier_bank_account"], "masking_applied": True,
        "destination_zone": "partner", "expires_at": (now + timedelta(days=1)).isoformat(),
    }, actor_uid=USER)
    assert high["status"] == "denied"
    assert high["reason_code"] == "highly_sensitive_egress_disabled"

    pending = service.submit_egress_request({
        "business_domain_uid": DOMAIN, "resource_uid": "01900000-0000-7000-8000-000000010502",
        "classification": "sensitive", "purpose": "supplier_contact_validation",
        "environment": "test", "requested_fields": ["supplier_contact_phone", "material_code"],
        "minimized_fields": ["supplier_contact_phone"], "masking_applied": True,
        "destination_zone": "controlled_partner", "expires_at": (now + timedelta(days=2)).isoformat(),
        "workflow_uid": WORKFLOW,
    }, actor_uid=USER)
    assert pending["status"] == "pending_approval"
    approvals.tasks[pending["approval_task_uid"]]["status"] = "approved"
    approved = service.reconcile_egress_request(
        pending["uid"], expected_version=1, actor_uid=REVIEWER
    )
    assert approved["status"] == "authorized_until_expiry"
    assert approved["approved_fields"] == ["supplier_contact_phone"]


def test_retention_is_policy_driven_and_never_automatically_deletes_evidence(security):
    service, _repository, _approvals, _transport, now = security
    policy = service.create_retention_policy({
        "code": "ACCESS_DECISION_5Y", "name": "访问决策五年保留",
        "evidence_type": "access_decision", "retention_days": 1825,
        "archive_mode": "immutable_external", "disposition_action": "review",
    }, actor_uid=OWNER)
    candidates = service.retention_candidates(as_of=now, limit=100)
    assert policy["automatic_deletion"] is False
    assert candidates[0]["automatic_deletion"] is False


def test_siem_sink_is_allowlisted_and_delivery_keeps_only_digest(security):
    service, repository, _approvals, transport, now = security
    with pytest.raises(ValueError, match="allowlist"):
        service.create_siem_sink({
            "name": "untrusted", "sink_type": "webhook",
            "endpoint": "https://evil.example/collect", "categories": ["authorization"],
        }, actor_uid=OWNER)
    sink = service.create_siem_sink({
        "name": "enterprise siem", "sink_type": "webhook",
        "endpoint": "https://siem.example.internal/dataops/events",
        "categories": ["authorization", "agent"],
    }, actor_uid=OWNER)
    delivery = service.dispatch_siem_events(sink["uid"], {
        "period_start": (now - timedelta(hours=1)).isoformat(),
        "period_end": now.isoformat(), "limit": 100,
    }, actor_uid=OWNER)
    assert delivery["status"] == "delivered"
    assert len(delivery["payload_digest"]) == 64
    assert "events" not in repository.deliveries[delivery["uid"]]
    assert transport.sent[0][1]["events"][0]["category"] == "authorization"


def test_sbom_and_vulnerability_lifecycle_requires_assignment_evidence_and_independent_close(security):
    service, _repository, _approvals, _transport, now = security
    sbom = service.register_sbom({
        "artifact_name": "dataops-backend", "artifact_version": "0.3.0",
        "artifact_type": "application", "source_ref": "build:wp10",
        "document": {
            "bomFormat": "CycloneDX", "specVersion": "1.5", "version": 1,
            "components": [{"type": "library", "name": "example-lib", "version": "1.0.0", "purl": "pkg:pypi/example-lib@1.0.0"}],
        },
    }, actor_uid=OWNER)
    finding = service.ingest_vulnerabilities(sbom["uid"], {
        "scanner": "trivy", "scan_ref": "scan:wp10",
        "findings": [{
            "external_id": "CVE-2026-1000", "severity": "high", "component_name": "example-lib",
            "installed_version": "1.0.0", "fixed_version": "1.0.1", "title": "example vulnerability",
        }],
    }, actor_uid=OWNER)[0]
    assigned = service.assign_vulnerability(
        finding["uid"], {"assignee_uid": USER, "due_at": (now + timedelta(days=7)).isoformat()},
        expected_version=1, actor_uid=OWNER,
    )
    resolved = service.resolve_vulnerability(
        finding["uid"], {
            "resolution_type": "patched", "resolved_version": "1.0.1",
            "resolution": "dependency upgraded and rescanned",
            "evidence_refs": [{"type": "scan", "ref": "scan:wp10-fixed", "digest": "a" * 64}],
        }, expected_version=2, actor_uid=USER,
    )
    with pytest.raises(PermissionError, match="independent reviewer"):
        service.close_vulnerability(
            finding["uid"], {"reason": "verified"}, expected_version=3, actor_uid=USER
        )
    closed = service.close_vulnerability(
        finding["uid"], {"reason": "fixed version and rescan evidence verified"},
        expected_version=3, actor_uid=REVIEWER,
    )
    assert assigned["status"] == "in_progress"
    assert resolved["status"] == "resolved"
    assert closed["status"] == "closed"
    assert service.dashboard()["open_vulnerability_count"] == 0