DataOps-platform/migrations/versions/20260811_485_trusted_delivery_hardening.py

"""Close WP06 asset-state races and database-side trusted-delivery guards."""

from alembic import op

revision = "20260811_485"
down_revision = "20260811_484"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(r"""
    CREATE OR REPLACE FUNCTION public.lock_trusted_delivery_asset(p_asset UUID)
    RETURNS void LANGUAGE plpgsql AS $$
    BEGIN
      PERFORM pg_advisory_xact_lock(hashtextextended(p_asset::text, 485));
    END $$;
    CREATE OR REPLACE FUNCTION public.guard_trusted_delivery_hold_asset()
    RETURNS trigger LANGUAGE plpgsql AS $$
    BEGIN
      PERFORM public.lock_trusted_delivery_asset(NEW.asset_uid);
      IF NEW.status='active' THEN UPDATE public.trusted_delivery_destruction_approvals SET status='blocked' WHERE asset_uid=NEW.asset_uid AND status='approved'; END IF;
      RETURN NEW;
    END $$;
    CREATE TRIGGER trg_trusted_delivery_hold_asset_lock BEFORE INSERT OR UPDATE OF status ON public.trusted_delivery_legal_holds FOR EACH ROW EXECUTE FUNCTION public.guard_trusted_delivery_hold_asset();
    CREATE OR REPLACE FUNCTION public.guard_trusted_delivery_destruction_asset()
    RETURNS trigger LANGUAGE plpgsql AS $$
    BEGIN
      PERFORM public.lock_trusted_delivery_asset(NEW.asset_uid);
      IF NEW.status='approved' AND EXISTS(SELECT 1 FROM public.trusted_delivery_legal_holds WHERE asset_uid=NEW.asset_uid AND status='active') THEN RAISE EXCEPTION 'destruction blocked by active legal hold'; END IF;
      RETURN NEW;
    END $$;
    DROP TRIGGER IF EXISTS trg_trusted_delivery_destruction_hold ON public.trusted_delivery_destruction_approvals;
    CREATE TRIGGER trg_trusted_delivery_destruction_asset_lock BEFORE INSERT OR UPDATE OF asset_uid,status ON public.trusted_delivery_destruction_approvals FOR EACH ROW EXECUTE FUNCTION public.guard_trusted_delivery_destruction_asset();
    CREATE OR REPLACE FUNCTION public.enforce_trusted_delivery_grant_active()
    RETURNS trigger LANGUAGE plpgsql AS $$
    BEGIN
      IF NEW.status='active' THEN
        IF NEW.approval_ref IS NULL OR NEW.approval_digest IS NULL OR NEW.purpose IS NULL OR NEW.environment IS NULL OR NEW.target_capability IS NULL OR jsonb_typeof(NEW.target_capability)<>'object' THEN RAISE EXCEPTION 'active trusted delivery grant requires bound approval and projection'; END IF;
        IF NOT EXISTS(SELECT 1 FROM public.trusted_delivery_policy_versions p WHERE p.uid=NEW.policy_uid AND p.status='active') OR NOT EXISTS(SELECT 1 FROM public.trusted_delivery_policy_transitions t WHERE t.policy_uid=NEW.policy_uid AND t.operation IN ('activate','rollback')) THEN RAISE EXCEPTION 'active trusted delivery grant requires active approved policy'; END IF;
      END IF;
      RETURN NEW;
    END $$;
    CREATE TRIGGER trg_trusted_delivery_grant_active_guard BEFORE INSERT OR UPDATE OF status,policy_uid,approval_ref,approval_digest,purpose,environment,target_capability ON public.trusted_delivery_grants FOR EACH ROW EXECUTE FUNCTION public.enforce_trusted_delivery_grant_active();
    CREATE OR REPLACE FUNCTION public.enforce_trusted_delivery_subscription_guard()
    RETURNS trigger LANGUAGE plpgsql AS $$
    DECLARE g RECORD; s RECORD;
    BEGIN
      IF TG_TABLE_NAME='trusted_delivery_subscriptions' THEN
        IF NEW.status='active' THEN SELECT subject_uid,asset_uid,purpose,expires_at,status INTO g FROM public.trusted_delivery_grants WHERE uid=NEW.grant_uid FOR KEY SHARE; IF NOT FOUND OR g.status<>'active' OR g.expires_at<=clock_timestamp() OR g.asset_uid<>NEW.asset_uid OR g.purpose<>NEW.purpose OR NEW.created_by<>g.subject_uid OR NEW.expires_at<=clock_timestamp() OR NEW.expires_at>g.expires_at OR EXISTS(SELECT 1 FROM public.trusted_delivery_legal_holds h WHERE h.asset_uid=NEW.asset_uid AND h.status='active') THEN RAISE EXCEPTION 'subscription activation requires current subject-bound grant without legal hold'; END IF; END IF;
      ELSE
        SELECT sub.status,sub.expires_at,sub.asset_uid,sub.grant_uid,gr.status AS grant_status,gr.expires_at AS grant_expires_at INTO s FROM public.trusted_delivery_subscriptions sub JOIN public.trusted_delivery_grants gr ON gr.uid=sub.grant_uid WHERE sub.uid=NEW.subscription_uid FOR KEY SHARE; IF NOT FOUND OR s.status<>'active' OR s.expires_at<=clock_timestamp() OR s.grant_status<>'active' OR s.grant_expires_at<=clock_timestamp() OR EXISTS(SELECT 1 FROM public.trusted_delivery_legal_holds h WHERE h.asset_uid=s.asset_uid AND h.status='active') THEN RAISE EXCEPTION 'subscription delivery requires current grant without legal hold'; END IF;
      END IF;
      RETURN NEW;
    END $$;
    CREATE TRIGGER trg_trusted_delivery_subscription_guard BEFORE INSERT OR UPDATE OF status,grant_uid,asset_uid,purpose,expires_at,created_by ON public.trusted_delivery_subscriptions FOR EACH ROW EXECUTE FUNCTION public.enforce_trusted_delivery_subscription_guard();
    CREATE TRIGGER trg_trusted_delivery_subscription_delivery_guard BEFORE INSERT OR UPDATE OF status,subscription_uid ON public.trusted_delivery_subscription_deliveries FOR EACH ROW EXECUTE FUNCTION public.enforce_trusted_delivery_subscription_guard();
    CREATE UNIQUE INDEX uq_trusted_delivery_single_revoke ON public.trusted_delivery_deliveries(grant_uid) WHERE operation='revoke';
    DO $$ BEGIN IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname='dataops_app_runtime') AND EXISTS (SELECT 1 FROM pg_roles WHERE rolname='dataops_edge_evidence_owner') THEN ALTER TABLE public.trusted_delivery_evidence OWNER TO dataops_edge_evidence_owner; ALTER TABLE public.trusted_delivery_receipts OWNER TO dataops_edge_evidence_owner; ALTER FUNCTION public.protect_trusted_delivery_evidence() OWNER TO dataops_edge_evidence_owner; ALTER FUNCTION public.protect_trusted_delivery_receipt() OWNER TO dataops_edge_evidence_owner; REVOKE INSERT,UPDATE,DELETE,TRUNCATE ON public.trusted_delivery_evidence,public.trusted_delivery_receipts FROM dataops_app_runtime; GRANT SELECT ON public.trusted_delivery_evidence,public.trusted_delivery_receipts TO dataops_app_runtime; END IF; END $$;
    """)


def downgrade() -> None:
    op.execute("""
    DO $$ BEGIN IF EXISTS(SELECT 1 FROM public.trusted_delivery_destruction_approvals WHERE status='blocked') THEN RAISE EXCEPTION 'downgrade requires approved hardening archival migration'; END IF; END $$;
    DROP INDEX public.uq_trusted_delivery_single_revoke;
    DROP TRIGGER trg_trusted_delivery_subscription_delivery_guard ON public.trusted_delivery_subscription_deliveries;
    DROP TRIGGER trg_trusted_delivery_subscription_guard ON public.trusted_delivery_subscriptions;
    DROP FUNCTION public.enforce_trusted_delivery_subscription_guard();
    DROP TRIGGER trg_trusted_delivery_grant_active_guard ON public.trusted_delivery_grants;
    DROP FUNCTION public.enforce_trusted_delivery_grant_active();
    DROP TRIGGER trg_trusted_delivery_destruction_asset_lock ON public.trusted_delivery_destruction_approvals;
    CREATE TRIGGER trg_trusted_delivery_destruction_hold BEFORE INSERT ON public.trusted_delivery_destruction_approvals FOR EACH ROW EXECUTE FUNCTION public.block_trusted_delivery_destruction_on_hold();
    DROP FUNCTION public.guard_trusted_delivery_destruction_asset();
    DROP TRIGGER trg_trusted_delivery_hold_asset_lock ON public.trusted_delivery_legal_holds;
    DROP FUNCTION public.guard_trusted_delivery_hold_asset();
    DROP FUNCTION public.lock_trusted_delivery_asset(UUID);
    """)