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

"""No-store HTTP boundary for trusted-delivery control and release gate summaries."""

from flask import g, jsonify, request
from sqlalchemy.exc import DBAPIError, IntegrityError

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.trusted_delivery_controls import TrustedDeliveryControlError
from app.core.system.trusted_delivery_controls_repository import (
    SqlAlchemyTrustedDeliveryControlsRepository,
)
from app.core.system.trusted_delivery_controls_service import (
    TrustedDeliveryControlsService,
)
from app.models.result import failed, success

MAX_TRUSTED_DELIVERY_CONTROL_REQUEST_BYTES = 16 * 1024


def _service():
    return TrustedDeliveryControlsService(SqlAlchemyTrustedDeliveryControlsRepository(db.session))


def _body(allowed):
    if request.content_length is not None and request.content_length > MAX_TRUSTED_DELIVERY_CONTROL_REQUEST_BYTES:
        raise TrustedDeliveryControlError("trusted delivery control request is too large")
    if len(request.get_data(cache=True)) > MAX_TRUSTED_DELIVERY_CONTROL_REQUEST_BYTES:
        raise TrustedDeliveryControlError("trusted delivery control request is too large")
    body = request.get_json(silent=True)
    if not isinstance(body, dict) or set(body) - allowed:
        raise TrustedDeliveryControlError("closed request schema")
    return body


def _execute(operation, *, created=False):
    try:
        value = operation()
        # Do not acknowledge a durable control mutation before its commit has
        # succeeded.  Failure takes the common rollback/redacted-error path.
        db.session.commit()
        response = jsonify(success(value, code=201 if created else 200))
        response.headers["Cache-Control"] = "no-store"
        return (response, 201) if created else response
    except PermissionError:
        db.session.rollback()
        response = jsonify(failed("trusted delivery control request denied", code=403))
    except (LookupError, RuntimeError, TrustedDeliveryControlError, ValueError, TypeError, IntegrityError, DBAPIError):
        db.session.rollback()
        response = jsonify(failed("trusted delivery control request rejected", code=400))
    response.headers["Cache-Control"] = "no-store"
    return response, response.get_json()["code"]


@bp.get("/trusted-delivery/controls/profiles/<profile_id>/active")
@require_permissions(SECURITY_GOVERNANCE_READ)
def trusted_delivery_active_control_profile(profile_id):
    return _execute(lambda: _service().active_profile_summary(profile_id))


@bp.post("/trusted-delivery/controls/release-gates/evaluate")
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def trusted_delivery_release_gate():
    return _execute(lambda: _service().evaluate_release_gate(_body({"artifact_digest", "sbom_digest", "license_policy_digest", "vulnerability_scan_digest", "approval_id", "approval_version", "scan_decision", "evidence_expires_at", "approval_ref", "approval_digest", "idempotency_key"}), actor_uid=g.current_user["id"]), created=True)


@bp.post("/trusted-delivery/controls/profiles/activate")
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def trusted_delivery_activate_control_profile():
    return _execute(lambda: _service().activate_profile(_body({"profile_id", "version", "approval_ref", "approval_digest", "idempotency_key"}), actor_uid=g.current_user["id"]), created=True)


@bp.post("/trusted-delivery/controls/profiles/rollback")
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def trusted_delivery_rollback_control_profile():
    return _execute(lambda: _service().activate_profile(_body({"profile_id", "version", "approval_ref", "approval_digest", "idempotency_key"}), actor_uid=g.current_user["id"], operation="rollback"))


@bp.post("/trusted-delivery/controls/capability-approvals")
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def trusted_delivery_capability_approval():
    return _execute(lambda: _service().approve_capability(_body({"provider", "capability", "version", "config_digest", "approval_ref", "approval_digest", "idempotency_key"}), actor_uid=g.current_user["id"]), created=True)


@bp.get("/trusted-delivery/controls/evidence")
@require_permissions(SECURITY_GOVERNANCE_READ)
def trusted_delivery_control_evidence():
    return _execute(lambda: _service().evidence_summary())


@bp.post("/trusted-delivery/controls/destruction-approvals")
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def trusted_delivery_destruction_approval():
    return _execute(lambda: _service().approve_destruction(_body({"asset_uid", "approval_refs", "evidence_digest", "idempotency_key"}), actor_uid=g.current_user["id"]), created=True)