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

"""No-store HTTP boundary for closed trusted-delivery contracts."""
from datetime import UTC, datetime

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,
    require_permissions,
)
from app.core.system.trusted_delivery import TrustedDeliveryService
from app.core.system.trusted_delivery_repository import (
    SqlAlchemyTrustedDeliveryRepository,
)
from app.core.system.trusted_delivery_subscription_repository import (
    SqlAlchemyTrustedDeliverySubscriptionRepository,
)
from app.core.system.trusted_delivery_subscriptions import (
    TrustedDeliverySubscriptionService,
)
from app.models.result import failed, success

MAX_TRUSTED_DELIVERY_REQUEST_BYTES = 16 * 1024


def _service():
    return TrustedDeliveryService(SqlAlchemyTrustedDeliveryRepository(db.session))


def _subscription_service():
    return TrustedDeliverySubscriptionService(
        SqlAlchemyTrustedDeliverySubscriptionRepository(db.session)
    )


def _execute(operation, *, created=False):
    try:
        value = operation()
        # Service/repository methods deliberately leave transaction ownership to
        # the HTTP boundary.  Commit before returning success so a second
        # connection can observe a successful response immediately.
        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 request denied", code=403))
    except (IntegrityError, DBAPIError):
        db.session.rollback()
        response = jsonify(failed("trusted delivery request conflict", code=409))
    except (LookupError, RuntimeError, ValueError, TypeError):
        db.session.rollback()
        response = jsonify(failed("trusted delivery request rejected", code=400))
    response.headers["Cache-Control"] = "no-store"
    return response, response.get_json()["code"]


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


@bp.post("/trusted-delivery/policy-versions")
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def create_trusted_delivery_policy():
    return _execute(lambda: _service().create_policy_version(_body({"code", "version", "selector", "resource_rules"}), actor_uid=g.current_user["id"]), created=True)


@bp.post("/trusted-delivery/policy-versions/activate")
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def activate_trusted_delivery_policy():
    return _execute(lambda: _service().activate_policy_version(_body({"code", "version", "approval_ref", "approval_digest", "idempotency_key"}), actor_uid=g.current_user["id"]))


@bp.post("/trusted-delivery/policy-versions/rollback")
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def rollback_trusted_delivery_policy():
    return _execute(lambda: _service().rollback_policy_version(_body({"code", "version", "approval_ref", "approval_digest", "idempotency_key"}), actor_uid=g.current_user["id"]))


@bp.post("/trusted-delivery/decisions")
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def evaluate_trusted_delivery():
    allowed = {"subject_uid", "roles", "business_domain_uid", "asset_uid", "purpose", "environment", "action", "classification", "requested_fields"}
    body = _body(allowed)
    body["subject_uid"] = g.current_user["id"]
    body["roles"] = list(g.current_user["roles"])
    return _execute(lambda: _service().evaluate_gateway(body))


@bp.post("/trusted-delivery/provisions")
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def provision_trusted_delivery():
    allowed = {"policy_uid", "approval_ref", "approval_digest", "asset_uid", "business_domain_uid", "purpose", "environment", "action", "classification", "requested_fields", "expires_at", "target_capability", "provider", "idempotency_key"}
    def operation():
        body = _body(allowed)
        # Browser callers never choose the identity facts used for policy matching.
        body["subject_uid"] = g.current_user["id"]
        return _service().provision(body, actor_uid=g.current_user["id"], subject_roles=list(g.current_user["roles"]))
    return _execute(operation, created=True)


@bp.post("/trusted-delivery/legal-holds")
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def create_trusted_delivery_hold():
    return _execute(lambda: _service().create_legal_hold(_body({"asset_uid", "operation", "approver_refs", "evidence_digest"}), actor_uid=g.current_user["id"]), created=True)


@bp.post("/trusted-delivery/legal-holds/release")
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def release_trusted_delivery_hold():
    return _execute(lambda: _service().release_legal_hold(_body({"hold_uid", "approval_ref", "approval_digest", "idempotency_key"}), actor_uid=g.current_user["id"]))


@bp.post("/trusted-delivery/reclaim/preview")
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def trusted_delivery_reclaim_preview():
    body = _body({"as_of"})
    as_of = body.get("as_of") or datetime.now(UTC).isoformat()
    return _execute(lambda: {"candidates": _service().preview_reclaim(as_of=as_of, actor_uid=g.current_user["id"]), "executed": False})


@bp.post("/trusted-delivery/reclaim/execute")
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def trusted_delivery_reclaim_execute():
    return _execute(lambda: _service().execute_reclaim(_body({"grant_uid", "idempotency_key"}), actor_uid=g.current_user["id"]))


@bp.post("/trusted-delivery/subscriptions")
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def create_trusted_delivery_subscription():
    allowed = {"grant_uid", "asset_uid", "trigger", "purpose", "expires_at", "idempotency_key"}
    return _execute(lambda: _subscription_service().create_subscription(_body(allowed), actor_uid=g.current_user["id"]), created=True)


@bp.post("/trusted-delivery/subscriptions/<subscription_uid>/pause")
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def pause_trusted_delivery_subscription(subscription_uid):
    return _execute(lambda: _subscription_service().pause_subscription(subscription_uid, actor_uid=g.current_user["id"]))


@bp.post("/trusted-delivery/subscriptions/<subscription_uid>/activate")
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def activate_trusted_delivery_subscription(subscription_uid):
    return _execute(lambda: _subscription_service().activate_subscription(subscription_uid, actor_uid=g.current_user["id"]))


@bp.post("/trusted-delivery/subscriptions/<subscription_uid>/resume")
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def resume_trusted_delivery_subscription(subscription_uid):
    return _execute(lambda: _subscription_service().resume_subscription(subscription_uid, actor_uid=g.current_user["id"]))


@bp.post("/trusted-delivery/subscriptions/<subscription_uid>/terminate")
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def terminate_trusted_delivery_subscription(subscription_uid):
    return _execute(lambda: _subscription_service().terminate_subscription(subscription_uid, actor_uid=g.current_user["id"]))


@bp.post("/trusted-delivery/subscriptions/deliveries/<delivery_uid>/compensate")
@require_permissions(SECURITY_GOVERNANCE_MANAGE)
def compensate_trusted_delivery_subscription_delivery(delivery_uid):
    return _execute(lambda: _subscription_service().compensate_dead_letter(
        delivery_uid, **_body({"reason_code", "receipt_code"}), actor_uid=g.current_user["id"]
    ))


@bp.post("/trusted-delivery/subscriptions/anomalies")
@require_permissions(SECURITY_GOVERNANCE_OPERATE)
def report_trusted_delivery_subscription_anomaly():
    allowed = {"kind", "asset_uid", "subscription_uid", "incident_uid", "summary", "correlation_id"}
    return _execute(lambda: _subscription_service().report_anomaly(_body(allowed), actor_uid=g.current_user["id"]), created=True)