DataOps-platform/tests/integration/test_security_governance_postgres.py

from __future__ import annotations

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

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

from app.core.system.governance_audit_repository import (
    SqlAlchemyGovernanceAuditRepository,
)
from app.core.system.security_governance import SecurityGovernanceService
from app.core.system.security_governance_repository import (
    SqlAlchemySecurityGovernanceRepository,
)

pytestmark = pytest.mark.integration


def _uid():
    return str(uuid.uuid4())


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

    def create_egress_task(self, request_record, workflow_uid, actor_uid):
        task = {"uid": _uid(), "status": "pending"}
        self.tasks[task["uid"]] = task
        return deepcopy(task)

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


class NoopTransport:
    def deliver(self, sink, envelope):
        return {"status": "delivered", "remote_ref": "integration"}


def test_second_domain_security_governance_and_vulnerability_closure_in_postgres():
    database_url = os.environ.get("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")
    engine = create_engine(database_url)
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)
    owner, reviewer, user, domain = _uid(), _uid(), _uid(), _uid()
    now = datetime.now(UTC)
    approvals = ApprovalGateway()
    try:
        for uid, name in ((owner, "owner"), (reviewer, "reviewer"), (user, "operator")):
            session.execute(text(
                "INSERT INTO public.users (id,username,display_name,password_hash,status) VALUES "
                "(CAST(:uid AS uuid),:username,:username,'integration-only','active')"
            ), {"uid": uid, "username": f"wp10-{name}-{uid[:8]}"})
        repository = SqlAlchemySecurityGovernanceRepository(session)
        service = SecurityGovernanceService(
            repository,
            approval_gateway=approvals,
            siem_transport=NoopTransport(),
            siem_host_allowlist={"siem.example.internal"},
            now_factory=lambda: now,
        )
        profile = service.create_classification_profile({
            "code": f"MATERIAL_{domain[:8].upper()}", "name": "备品备件分类规则",
            "business_domain_uid": domain, "default_classification": "internal",
            "rules": [
                {"field_tokens": ["phone"], "category": "personal_contact", "classification": "sensitive"},
                {"field_tokens": ["bank"], "category": "financial_account", "classification": "highly_sensitive"},
            ],
        }, actor_uid=owner)
        scan = service.scan_sensitive_sample({
            "profile_uid": profile["uid"], "resource_type": "material_master",
            "resource_uid": _uid(), "business_domain_uid": domain,
            "fields": [
                {"name": "material_code", "sample_values": ["MAT-0001"]},
                {"name": "supplier_phone", "sample_values": ["13800138000"]},
                {"name": "supplier_bank_account", "sample_values": ["6222021234567890"]},
            ],
        }, actor_uid=owner)
        assert scan["sample_retained"] is False
        assert {item["proposed_classification"] for item in scan["findings"]} == {"sensitive", "highly_sensitive"}
        reviewed = service.review_classification_finding(
            scan["findings"][0]["uid"],
            {"decision": "confirm", "final_classification": "sensitive", "reason": "供应商联系方式属于个人联系信息"},
            expected_version=1, actor_uid=reviewer,
        )
        assert reviewed["current_version"] == 2

        service.create_access_policy({
            "code": f"MATERIAL_READ_{domain[:8].upper()}", "name": "物料质量用途受控读取",
            "business_domain_uid": domain, "subject_user_uids": [user], "subject_roles": [],
            "purposes": ["material_quality"], "environments": ["test"], "actions": ["read"],
            "max_classification": "sensitive", "allowed_fields": ["material_code", "supplier_phone"],
            "review_due_at": (now + timedelta(days=7)).isoformat(),
            "expires_at": (now + timedelta(days=14)).isoformat(),
        }, actor_uid=owner)
        assert service.evaluate_access({
            "user_uid": user, "roles": ["editor"], "business_domain_uid": domain,
            "purpose": "material_quality", "environment": "test", "action": "read",
            "resource_uid": _uid(), "classification": "sensitive", "requested_fields": ["material_code"],
        })["decision"] == "authorized"
        assert service.submit_egress_request({
            "business_domain_uid": domain, "resource_uid": _uid(), "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)["status"] == "denied"

        sbom = service.register_sbom({
            "artifact_name": "dataops-backend", "artifact_version": "wp10", "artifact_type": "application",
            "source_ref": "integration:wp10", "document": {
                "bomFormat": "CycloneDX", "specVersion": "1.5", "version": 1,
                "components": [{"type": "library", "name": "example-lib", "version": "1.0.0"}],
            },
        }, actor_uid=owner)
        finding = service.ingest_vulnerabilities(sbom["uid"], {
            "scanner": "trivy", "scan_ref": "integration:scan", "findings": [{
                "external_id": "CVE-2026-1000", "severity": "high", "component_name": "example-lib",
                "installed_version": "1.0.0", "fixed_version": "1.0.1", "title": "integration finding",
            }],
        }, actor_uid=owner)[0]
        finding = service.assign_vulnerability(finding["uid"], {
            "assignee_uid": user, "due_at": (now + timedelta(days=7)).isoformat()
        }, expected_version=1, actor_uid=owner)
        finding = service.resolve_vulnerability(finding["uid"], {
            "resolution_type": "patched", "resolved_version": "1.0.1", "resolution": "已升级并复扫",
            "evidence_refs": [{"type": "scan", "ref": "integration:fixed", "digest": "a" * 64}],
        }, expected_version=2, actor_uid=user)
        finding = service.close_vulnerability(finding["uid"], {"reason": "复扫无对应发现"}, expected_version=3, actor_uid=reviewer)
        assert finding["status"] == "closed"

        events = SqlAlchemyGovernanceAuditRepository(session).fetch_events(
            categories=["security_governance"], period_start=now - timedelta(minutes=1), period_end=now + timedelta(minutes=1)
        )
        assert any(item["action"] == "egress_denied" for item in events)
        serialized = repr(repository.list_findings())
        assert "13800138000" not in serialized
        assert "6222021234567890" not in serialized
    finally:
        session.close()
        transaction.rollback()
        connection.close()
        engine.dispose()