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

"""Cross-domain data security and security engineering APIs."""

from __future__ import annotations

from datetime import UTC, datetime

from flask import current_app, g, jsonify, request

from app import db
from app.api.system import bp
from app.core.system.permissions import (
    SECURITY_GOVERNANCE_MANAGE,
    SECURITY_GOVERNANCE_OPERATE,
    SECURITY_GOVERNANCE_READ,
    require_permissions,
)
from app.core.system.security_delivery import HttpsWebhookSyslogTransport
from app.core.system.security_governance import SecurityGovernanceService
from app.core.system.security_governance_repository import (
    SqlAlchemySecurityGovernanceRepository,
    WorkCenterSecurityApprovalGateway,
)
from app.models.result import failed, success


def _service():
    allowlist = {
        item.strip().lower()
        for item in str(current_app.config.get("SECURITY_SIEM_HOST_ALLOWLIST") or "").split(",")
        if item.strip()
    }
    return SecurityGovernanceService(
        SqlAlchemySecurityGovernanceRepository(db.session),
        approval_gateway=WorkCenterSecurityApprovalGateway(db.session),
        siem_transport=HttpsWebhookSyslogTransport(),
        siem_host_allowlist=allowlist,
        commit=db.session.commit,
        rollback=db.session.rollback,
    )


def _version():
    value = str(request.headers.get("If-Match") or "").strip().removeprefix("W/").strip('"')
    if not value.isdigit():
        raise ValueError("missing valid If-Match version")
    return int(value)


def _etag(response, record):
    if record.get("current_version"):
        response.headers["ETag"] = f'"{record["current_version"]}"'
    return response


def _execute(operation, *, created=False, versioned=False):
    try:
        result = operation()
        response = jsonify(success(result, code=201 if created else 200))
        if versioned and isinstance(result, dict):
            response = _etag(response, result)
        return (response, 201) if created else response
    except Exception as exc:
        db.session.rollback()
        if isinstance(exc, LookupError):
            status = 404
        elif isinstance(exc, PermissionError):
            status = 403
        elif isinstance(exc, RuntimeError):
            status = 409
        else:
            status = 400
        return jsonify(failed(str(exc), code=status)), status


@bp.route("/security-governance/dashboard", methods=["GET"])
@require_permissions(SECURITY_GOVERNANCE_READ)
def security_dashboard():
    return _execute(_service().dashboard)


@bp.route("/security-governance/profiles", methods=["GET"])
@require_permissions(SECURITY_GOVERNANCE_READ)
def security_list_profiles():
    return _execute(lambda: _service().list_profiles(business_domain_uid=request.args.get("business_domain_uid")))


@bp.route("/security-governance/profiles", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def security_create_profile():
    return _execute(
        lambda: _service().create_classification_profile(request.get_json(silent=True) or {}, actor_uid=g.current_user["id"]),
        created=True,
    )


@bp.route("/security-governance/scans", methods=["GET"])
@require_permissions(SECURITY_GOVERNANCE_READ)
def security_list_scans():
    return _execute(lambda: _service().list_classification_scans(business_domain_uid=request.args.get("business_domain_uid")))


@bp.route("/security-governance/scans", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def security_create_scan():
    return _execute(
        lambda: _service().scan_sensitive_sample(request.get_json(silent=True) or {}, actor_uid=g.current_user["id"]),
        created=True,
    )


@bp.route("/security-governance/findings", methods=["GET"])
@require_permissions(SECURITY_GOVERNANCE_READ)
def security_findings():
    return _execute(lambda: _service().list_classification_findings(status=request.args.get("status")))


@bp.route("/security-governance/findings/<finding_uid>/review", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def security_review_finding(finding_uid):
    return _execute(
        lambda: _service().review_classification_finding(
            finding_uid, request.get_json(silent=True) or {}, expected_version=_version(), actor_uid=g.current_user["id"]
        ),
        versioned=True,
    )


@bp.route("/security-governance/access-policies", methods=["GET"])
@require_permissions(SECURITY_GOVERNANCE_READ)
def security_list_access_policies():
    return _execute(lambda: _service().list_access_policies(business_domain_uid=request.args.get("business_domain_uid"), status=request.args.get("status")))


@bp.route("/security-governance/access-policies", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def security_create_access_policy():
    return _execute(
        lambda: _service().create_access_policy(request.get_json(silent=True) or {}, actor_uid=g.current_user["id"]),
        created=True,
    )


@bp.route("/security-governance/access-decisions", methods=["GET"])
@require_permissions(SECURITY_GOVERNANCE_READ)
def security_access_decisions():
    return _execute(lambda: _service().list_access_decisions(decision=request.args.get("decision")))


@bp.route("/security-governance/access/evaluate", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def security_evaluate_access():
    body = request.get_json(silent=True) or {}
    body["user_uid"] = g.current_user["id"]
    body["roles"] = list(g.current_user["roles"])
    return _execute(lambda: _service().evaluate_access(body), created=True)


@bp.route("/security-governance/egress", methods=["GET"])
@require_permissions(SECURITY_GOVERNANCE_READ)
def security_list_egress():
    return _execute(lambda: _service().list_egress_requests(status=request.args.get("status")))


@bp.route("/security-governance/egress", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def security_create_egress():
    return _execute(
        lambda: _service().submit_egress_request(request.get_json(silent=True) or {}, actor_uid=g.current_user["id"]),
        created=True,
        versioned=True,
    )


@bp.route("/security-governance/egress/<request_uid>/reconcile", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def security_reconcile_egress(request_uid):
    return _execute(
        lambda: _service().reconcile_egress_request(request_uid, expected_version=_version(), actor_uid=g.current_user["id"]),
        versioned=True,
    )


@bp.route("/security-governance/retention", methods=["GET"])
@require_permissions(SECURITY_GOVERNANCE_READ)
def security_list_retention():
    return _execute(_service().list_retention_policies)


@bp.route("/security-governance/retention", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def security_create_retention():
    return _execute(
        lambda: _service().create_retention_policy(request.get_json(silent=True) or {}, actor_uid=g.current_user["id"]),
        created=True,
    )


@bp.route("/security-governance/retention/candidates", methods=["GET"])
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def security_retention_candidates():
    value = request.args.get("as_of") or datetime.now(UTC).isoformat()
    return _execute(lambda: _service().retention_candidates(as_of=value, limit=int(request.args.get("limit", 100))))


@bp.route("/security-governance/siem/sinks", methods=["GET"])
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def security_list_siem_sinks():
    return _execute(_service().list_siem_sinks)


@bp.route("/security-governance/siem/sinks", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def security_create_siem_sink():
    return _execute(
        lambda: _service().create_siem_sink(request.get_json(silent=True) or {}, actor_uid=g.current_user["id"]),
        created=True,
    )


@bp.route("/security-governance/siem/deliveries", methods=["GET"])
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def security_siem_deliveries():
    return _execute(lambda: _service().list_siem_deliveries(sink_uid=request.args.get("sink_uid")))


@bp.route("/security-governance/siem/sinks/<sink_uid>/dispatch", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def security_siem_dispatch(sink_uid):
    return _execute(lambda: _service().dispatch_siem_events(sink_uid, request.get_json(silent=True) or {}, actor_uid=g.current_user["id"]), created=True)


@bp.route("/security-governance/sboms", methods=["GET"])
@require_permissions(SECURITY_GOVERNANCE_READ)
def security_list_sboms():
    return _execute(_service().list_sboms)


@bp.route("/security-governance/sboms", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def security_register_sbom():
    return _execute(lambda: _service().register_sbom(request.get_json(silent=True) or {}, actor_uid=g.current_user["id"]), created=True)


@bp.route("/security-governance/sboms/<sbom_uid>/vulnerabilities", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def security_import_vulnerabilities(sbom_uid):
    return _execute(lambda: _service().ingest_vulnerabilities(sbom_uid, request.get_json(silent=True) or {}, actor_uid=g.current_user["id"]), created=True)


@bp.route("/security-governance/vulnerabilities", methods=["GET"])
@require_permissions(SECURITY_GOVERNANCE_READ)
def security_vulnerabilities():
    return _execute(lambda: _service().list_vulnerabilities(status=request.args.get("status"), severity=request.args.get("severity")))


def _vulnerability_change(finding_uid, action):
    service = _service()
    method = getattr(service, f"{action}_vulnerability")
    return _execute(
        lambda: method(finding_uid, request.get_json(silent=True) or {}, expected_version=_version(), actor_uid=g.current_user["id"]),
        versioned=True,
    )


@bp.route("/security-governance/vulnerabilities/<finding_uid>/assign", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def security_assign_vulnerability(finding_uid):
    return _vulnerability_change(finding_uid, "assign")


@bp.route("/security-governance/vulnerabilities/<finding_uid>/resolve", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def security_resolve_vulnerability(finding_uid):
    return _vulnerability_change(finding_uid, "resolve")


@bp.route("/security-governance/vulnerabilities/<finding_uid>/close", methods=["POST"])
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def security_close_vulnerability(finding_uid):
    return _vulnerability_change(finding_uid, "close")