DataOps-platform/deployment/app/api/system/governance_audit.py

"""Admin-only governance audit and tamper-evident runtime evidence APIs."""

from __future__ import annotations

import hashlib
import logging
from datetime import UTC, datetime, timedelta

from flask import current_app, g, jsonify, request

from app import db
from app.api.system import bp
from app.config.config import is_placeholder_env_value
from app.core.data_source.redaction import sanitize_exception
from app.core.system.governance_audit import (
    GovernanceAuditInvalid,
    GovernanceAuditNotFound,
    GovernanceAuditService,
)
from app.core.system.governance_audit_repository import (
    SqlAlchemyGovernanceAuditRepository,
)
from app.models.result import failed, success

logger = logging.getLogger(__name__)


class GovernanceAuditUnavailable(RuntimeError):
    """Raised when the server cannot safely perform a sealing operation."""


def _effective_evidence_secret() -> tuple[str, bool]:
    dedicated = str(
        current_app.config.get("AUDIT_EVIDENCE_SECRET") or ""
    ).strip()
    dedicated_ready = (
        len(dedicated.encode("utf-8")) >= 32
        and not is_placeholder_env_value(dedicated)
    )
    if dedicated_ready:
        return dedicated, True
    fallback = hashlib.sha256(
        (
            "dataops-wp12-local-audit:"
            + str(current_app.config.get("SECRET_KEY") or "")
        ).encode("utf-8")
    ).hexdigest()
    return fallback, False


def _sealing_allowed(dedicated_ready: bool) -> bool:
    if dedicated_ready:
        return True
    if current_app.config.get("TESTING"):
        return True
    return (
        str(current_app.config.get("FLASK_ENV") or "").lower()
        != "production"
    )


def _require_sealing_key() -> None:
    _secret, dedicated_ready = _effective_evidence_secret()
    if not _sealing_allowed(dedicated_ready):
        raise GovernanceAuditUnavailable(
            "dedicated audit evidence key is required"
        )


def get_governance_audit_service():
    secret, _dedicated = _effective_evidence_secret()
    return GovernanceAuditService(
        SqlAlchemyGovernanceAuditRepository(db.session),
        evidence_secret=secret,
        key_version=str(
            current_app.config.get("AUDIT_EVIDENCE_KEY_VERSION")
            or "local-fallback-v1"
        ),
    )


def get_security_checks():
    repository = SqlAlchemyGovernanceAuditRepository(db.session)
    snapshot = repository.security_snapshot()
    _secret, dedicated_ready = _effective_evidence_secret()
    credential_count = snapshot["credential_count"]
    encrypted_count = snapshot["encrypted_credential_count"]
    plaintext_count = snapshot["plaintext_source_config_count"]
    return {
        "seal_ready": _sealing_allowed(dedicated_ready),
        "production_key_ready": dedicated_ready,
        "assurance": (
            "签名封存用于检测篡改，不等于阻止数据库管理员修改；"
            "生产环境必须使用独立密钥并纳入企业密钥保管与轮换。"
        ),
        "checks": [
            {
                "code": "datasource_credential_encryption",
                "name": "数据源凭据密文存储",
                "status": (
                    "passed"
                    if credential_count == encrypted_count
                    else "failed"
                ),
                "observed_count": encrypted_count,
                "expected_count": credential_count,
            },
            {
                "code": "legacy_plaintext_cleanup",
                "name": "采集源配置无明文秘密字段",
                "status": "passed" if plaintext_count == 0 else "failed",
                "observed_count": plaintext_count,
                "expected_count": 0,
            },
            {
                "code": "dedicated_evidence_key",
                "name": "独立审计封存密钥",
                "status": "passed" if dedicated_ready else "warning",
                "reason_code": (
                    "dedicated_key_configured"
                    if dedicated_ready
                    else "local_fallback_key_in_use"
                ),
            },
            {
                "code": "safe_error_boundary",
                "name": "异常与日志脱敏边界",
                "status": "passed",
            },
            {
                "code": "security_response_headers",
                "name": "API 安全响应头",
                "status": "passed",
            },
            {
                "code": "audit_source_coverage",
                "name": "十一类关键操作审计源",
                "status": "passed",
                "expected_count": 11,
            },
        ],
    }


def _timestamp(value, *, default=None):
    if value in (None, ""):
        if default is None:
            raise GovernanceAuditInvalid("timestamp is required")
        return default
    try:
        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except (TypeError, ValueError) as exc:
        raise GovernanceAuditInvalid("timestamp must be ISO-8601") from exc
    if parsed.tzinfo is None:
        raise GovernanceAuditInvalid("timestamp must include a timezone")
    return parsed.astimezone(UTC)


def _window(values):
    now = datetime.now(UTC)
    return (
        _timestamp(
            values.get("period_start"),
            default=now - timedelta(days=30),
        ),
        _timestamp(values.get("period_end"), default=now),
    )


def _categories(values):
    if hasattr(values, "getlist"):
        result = values.getlist("category")
        if result:
            return result
    return values.get("categories") or values.get("category")


def _safe_error(error, status=503):
    logger.error(
        "governance audit request failed: %s",
        sanitize_exception(error, limit=300),
    )
    return (
        jsonify(
            failed(
                "审计与运行证据暂不可用",
                code=status,
                error={"code": "GOVERNANCE_AUDIT_UNAVAILABLE"},
            )
        ),
        status,
    )


def _execute(operation):
    try:
        return jsonify(success(operation()))
    except GovernanceAuditInvalid as exc:
        return (
            jsonify(
                failed(
                    str(exc),
                    code=400,
                    error={"code": "GOVERNANCE_AUDIT_INVALID"},
                )
            ),
            400,
        )
    except GovernanceAuditNotFound:
        return (
            jsonify(
                failed(
                    "审计证据封存不存在",
                    code=404,
                    error={"code": "GOVERNANCE_AUDIT_NOT_FOUND"},
                )
            ),
            404,
        )
    except GovernanceAuditUnavailable:
        return _safe_error(
            GovernanceAuditUnavailable("audit sealing is not configured")
        )
    except Exception as exc:
        db.session.rollback()
        return _safe_error(exc)


@bp.route("/governance-audit/security-checks", methods=["GET"])
def governance_audit_security_checks():
    return _execute(get_security_checks)


@bp.route("/governance-audit/coverage", methods=["GET"])
def governance_audit_coverage():
    def operation():
        start, end = _window(request.args)
        return get_governance_audit_service().coverage(
            period_start=start,
            period_end=end,
        )

    return _execute(operation)


@bp.route("/governance-audit/events", methods=["GET"])
def governance_audit_events():
    def operation():
        start, end = _window(request.args)
        return get_governance_audit_service().list_events(
            period_start=start,
            period_end=end,
            categories=_categories(request.args),
            action=request.args.get("action"),
            status=request.args.get("status"),
            page=request.args.get("page", 1),
            page_size=request.args.get("page_size", 20),
        )

    return _execute(operation)


@bp.route("/governance-audit/seals", methods=["GET"])
def governance_audit_seals():
    return _execute(
        lambda: get_governance_audit_service().list_seals(
            limit=request.args.get("limit", 50)
        )
    )


@bp.route("/governance-audit/seals", methods=["POST"])
def create_governance_audit_seal():
    def operation():
        _require_sealing_key()
        payload = request.get_json(silent=True) or {}
        start, end = _window(payload)
        seal = get_governance_audit_service().create_seal(
            period_start=start,
            period_end=end,
            categories=_categories(payload),
            actor_uid=g.current_user["id"],
        )
        db.session.commit()
        return seal

    return _execute(operation)


@bp.route(
    "/governance-audit/seals/<seal_uid>/verify",
    methods=["POST"],
)
def verify_governance_audit_seal(seal_uid):
    def operation():
        _require_sealing_key()
        return get_governance_audit_service().verify_seal(seal_uid)

    return _execute(operation)