Prism-Shield: Zero-Trust AI-to-DAG Execution Gateway

Seal AI extractions into cryptographically attested ParameterManifest FlatBuffers, enforce them at the Group-3 boundary, and escalate uncertain cases to human review — before agents touch databases, payments, or underwriting DAGs.

Stage: v0.2 pilot-hardening — suitable for design-partner pilots under EULA. Not a SOC2-certified product and not a guaranteed sub-millisecond C++ path unless the Business/Enterprise C++ sidecar is installed (gate_path=inprocess).

Pricing (list): Developer $0 · Team $299/mo · Business $999/mo · Enterprise $35k–$120k+/yr — full packaging on insightits.com; engineering source of truth in docs/PRICING.md.

The Enterprise Problem ("Excessive Agency & Unvetted Inputs")

Security and risk teams are being asked to ship LLM and multi-agent systems into workflows that already move money, change underwriting decisions, and write to systems of record. The failure mode is not a chat hallucination—it is unvetted probabilistic output becoming deterministic side effects.

Prism-Shield is the zero-trust runtime boundary between agent reasoning and enterprise DAGs: canonicalize → policy/schema check → seal a Group-3 ParameterManifest FlatBuffer → C++/Python enforce → KMS attest → execute or escalate to human review.

Quickstart (3-Line Execution Guard)

pip install prism-shield

Protect any business DAG so it cannot run unless Prism-Shield accepts and seals the extraction:

from prismmanifest.prism_shield import (
    protect_execution_dag,
    AgentExtractionPayload,
    ExecutionContext,
)

@protect_execution_dag(
    policy_id="policy-underwriting-v1",
    signing_backend=kms_backend,       # KmsEnvelopeBackend or KeyRing
    schema_registry=schema_registry,   # fail-closed schema allowlist
)
async def execute_loan_disbursement(signed_manifest_bytes: bytes):
    # Runs only after C++/Python gate ACCEPT + KMS attestation.
    # signed_manifest_bytes is a sealed FlatBuffer ParameterManifest.
    return loan_engine.disburse(signed_manifest_bytes)


# Caller supplies the agent extraction + tenant context:
await execute_loan_disbursement(
    AgentExtractionPayload(
        agent_id="extractor-01",
        dag_target_id="capital_gains_v3",
        schema_hash=SCHEMA_HASH,  # 64-hex published schema
        extracted_parameters=[...],
    ),
    ExecutionContext(
        session_id="sess-42",
        tenant_id="tenant-acme",
        policy_id="policy-underwriting-v1",
    ),
)

Decisions

How Prism-Shield Works

┌─────────────────────┐     ┌──────────────────────────────┐     ┌─────────────────────┐
│  AI Agent / Graph   │────▶│  PrismShieldGateway          │────▶│  Enterprise DAG     │
│  extractions        │     │  policy · schema · replay    │     │  DB / payments /    │
│  AgentExtraction…   │     │  seal ParameterManifest FB   │     │  underwriting       │
└─────────────────────┘     │  C++ enforce_fb / Python gate│     └─────────────────────┘
                            │  KMS envelope Ed25519        │
                            │  HITL escalation queue       │
                            └──────────────────────────────┘

Core types live in prismmanifest.prism_core.models:

• AgentExtractionPayload — agent_id, dag_target_id, schema_hash, proposed parameters + evidence
• ExecutionContext — session_id, tenant_id, policy_id, environment
• VerificationOutcome — decision, sealed bytes, signature, reasons, escalation_id, execution_time_ms
• TrustDecision — ACCEPT | REVIEW | REFUSE

Gateway entrypoint (prismmanifest.prism_shield.gateway.PrismShieldGateway):

1. Resolve policy + schema (unknown IDs → REFUSE)
2. Canonicalize extraction with session/tenant binding
3. Python pre-filter (confidence floors, required fields, DAG allowlist)
4. On ACCEPT candidate: map → ParameterManifest → KMS sign → FlatBuffer seal
5. Enforce via C++ enforce_fb when the in-process library is available, else Python enforce_group3_boundary
6. Record replay receipt (session_id + canonical digest)
7. On REVIEW: sign PASS_WITH_HUMAN and enqueue EscalationQueue for the review UI

Framework Adapters

Generic decorator

from prismmanifest.prism_shield import protect_execution_dag

@protect_execution_dag(policy_id="policy-underwriting-v1", signing_backend=backend, ...)
async def my_dag(signed_manifest_bytes: bytes, **kwargs):
    ...

LangGraph / ChorusGraph / CrewAI node

from prismmanifest.prism_shield import PrismShieldNode

shield = PrismShieldNode(
    "policy-underwriting-v1",
    signing_backend=backend,
    schema_registry=schemas,
)

# Graph state must include proposed_extraction, session_id, and tenant_id
# (adapters fail closed — no default_session / default_tenant).
state = await shield(state)
# state["prism_decision"] in {"accept", "review", "refuse"}
# state["prism_outcome"]  → VerificationOutcome.model_dump()

Wire a conditional edge on prism_decision so refuse / review never reach the execution node.

Imperative gateway

from prismmanifest.prism_shield import PrismShieldGateway, verify_attestation

gateway = PrismShieldGateway(
    policy_id="policy-underwriting-v1",
    signing_backend=kms_backend,
    schema_registry=schema_registry,
    policy_registry=policy_registry,
    replay_guard=ReplayGuard(),
    prefer_cpp=True,
)

outcome = await gateway.verify_and_authorize(payload, context)
if outcome.decision.value == "ACCEPT":
    manifest = verify_attestation(
        outcome.signed_manifest_bytes,
        kms_backend,
        expected_dag_id="capital_gains_v3",
        expected_schema_hash=SCHEMA_HASH,
        replay_guard=False,  # already recorded at the gateway
    )

Cryptographic Attestation & C++ Gate

from prismmanifest.keys.kms_envelope import KmsEnvelopeBackend

# Production: load an already-provisioned envelope key (fail closed).
kms_backend = KmsEnvelopeBackend.load("/var/prism/kms", "prod-underwriting-01")

# Local/CI only:
# kms_backend = KmsEnvelopeBackend.generate(store, "dev-key", mode="local")

Human-in-the-Loop (HITL) Review

When confidence or gate policy requires review, Prism-Shield:

1. Seals a PASS_WITH_HUMAN ParameterManifest with a real attestation (not a placeholder)
2. Enqueues via prismmanifest.audit.escalation.EscalationQueue
3. Surfaces an escalation_id for the review dashboard (prismmanifest.review_ui.app.ReviewApp)

Operators approve or reject with a shared review token (PRISMMANIFEST_REVIEW_TOKEN). Approved clearance can mint a human-approval token for Group-3 execution under controlled allow_pass_with_human paths.

REVIEW → EscalationQueue (jsonl) → Review UI → Approve / Reject → Audit trail

Policy & Schema Registries (Fail Closed)

Unknown policy_id or schema_hash refuses—Prism-Shield never invents trust.

from prismmanifest.prism_shield import (
    PolicyRegistry,
    TrustPolicy,
    SchemaRegistry,
    SchemaRecord,
)

policies = PolicyRegistry()
policies.register(
    TrustPolicy(
        policy_id="policy-underwriting-v1",
        accept_confidence=0.85,
        review_confidence=0.50,
        allowed_dag_ids=frozenset({"capital_gains_v3"}),
        required_fields=("agi_usd",),
    )
)

schemas = SchemaRegistry()
schemas.register(
    SchemaRecord(
        schema_hash="a" * 64,  # publish the real blake/sha digest of your schema
        dag_id="capital_gains_v3",
        required_fields=("agi_usd",),
        field_units={"agi_usd": "USD"},
    )
)

Replay protection binds session_id + canonical digest so the same sealed decision cannot be replayed into a DAG.

Pricing & packaging

Primary metric: Verified Execution = one verify_and_authorize() call.

Full matrix, overage rules, and upgrade levers: docs/PRICING.md.
Long-form buyer copy, ROI, and order forms: www.insightits.com (keep list prices synced with docs/PRICING.md).

from prismmanifest.prism_shield.metering import ExecutionMeter, Plan

meter = ExecutionMeter(plan=Plan.DEVELOPER)  # enforces 10k/mo locally
gateway = PrismShieldGateway(..., execution_meter=meter)

Enterprise Controls (control-mapped, not certified)

See the full matrix in docs/SOC2_CONTROL_MAP.md and SKUs in docs/SKU.md.

Pair with Prism-Eval in CI to adversarial-test digit drops and injections before Shield enforces the boundary in production.

CI (Prism-Eval)  ──▶  merge   ──▶  Runtime (Prism-Shield gateway)

Configuration Reference

CI also runs a clean-install job: build the overlay wheel, install into a fresh venv, and run tests with the vendored repo tree not on PYTHONPATH so green matches what pip install prism-shield gets. | PRISMMANIFEST_SHIELD_AUDIT_ROOT | Decision audit journal root | | PRISMMANIFEST_C_DLL / PRISMMANIFEST_C_LIB | Optional in-process C++ gate library | | prefer_cpp=True | Prefer enforce_fb when the DLL is loaded | | bound_tenant_id | Hard-bind gateway instance to one tenant | | allow_ephemeral_keys=False | Keep false in production |

Interactive Demo

Walk ACCEPT / REVIEW / REFUSE / replay / FlatBuffer inspect in your terminal:

pip install -e ".[dev]"
python demos/interactive_demo.py
# or:
prism-shield-demo

Try it on GitHub: open this repo in Codespaces — the container installs the package and runs the CI smoke path; then run python demos/interactive_demo.py for the menu-driven demo.

Commercial pilots & sales: www.insightits.com.

Non-interactive CI smoke (used by GitHub Actions):

python demos/ci_smoke_demo.py

Install & Verify

pip install prism-shield
# pulls dependency: prismmanifest>=0.3.4
# or from source:
pip install -e ".[dev]"
pytest tests/test_prism_shield_gateway.py -v
python demos/ci_smoke_demo.py

Publish maintainers: see docs/PUBLISHING.md (TestPyPI → PyPI).

Self-host stubs: deploy/README.md (Docker / Helm / K8s).

Expected: ACCEPT seals a FlatBuffer verifiable by enforce_group3_boundary / verify_attestation; REVIEW escalates; REFUSE blocks; replay of the same session receipt is rejected.

Package Layout

prismmanifest/
  prism_core/           # Pydantic domain models + adapters
    models.py
    adapters/
      generic.py        # @protect_execution_dag
      chorusgraph.py    # PrismShieldNode
      langgraph.py      # PrismShieldNode
  prism_shield/         # Runtime gateway
    gateway.py          # PrismShieldGateway.verify_and_authorize
    engine.py           # seal · enforce · KMS · escalation bindings
    policy.py           # PolicyRegistry
    schema_registry.py  # SchemaRegistry
  cpp_bridge.py         # C++ FlatBuffer enforce
  keys/kms_envelope.py  # Cloud/HSM envelope signing
  audit/escalation.py   # HITL queue
  review_ui/app.py      # Review dashboard

License & Contact

• Author: Amin Parva
• Company: www.insightits.com
• GitHub: insightitsGit/Prism-Shield
• PyPI: prism-shield
• Open core (this repo): Apache-2.0 — see LICENSE
• Commercial SKU / EULA: COMMERCIAL_LICENSE.md
• Support tiers: docs/SUPPORT.md
• Security reports: docs/SECURITY.md

Prism-Shield — zero-trust execution for AI agents that touch real money and real systems of record.

Author: Amin Parva · InsightITS · GitHub · PyPI · Prism-Eval · Issues