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.

Risk	What happens in production
Unchecked tool execution	Agent extractions flow straight into DB writes, wire transfers, claims payouts, or underwriting approvals with no cryptographic gate.
Silent drift & injection	Digit drops ($150,000 → $150.00), OCR column shifts, and PDF-footer prompt injections alter parameters that rules engines treat as truth.
No cryptographic auditability	After an automated decision, there is no sealed, verifiable artifact proving which evidence authorized which parameters under which policy.

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

TrustDecision	Runtime behavior
ACCEPT	Sealed FlatBuffer + Ed25519 signature returned to the DAG
REVIEW	Escalation queued → EscalationRequiredException (HITL)
REFUSE	Hard block → SecurityTrustException

---

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

Capability	Implementation
Canonical numeric claims	prismmanifest.canonicalize (micro-units + BLAKE3 digit fingerprints)
Sealed artifact	FlatBuffer ParameterManifest via binary_codec.encode_manifest
Signing	KmsEnvelopeBackend (Azure Key Vault / AWS KMS / GCP / local) — Ed25519 after envelope unwrap
Fail-closed keys	No silent key minting unless allow_ephemeral_keys=True and PRISMMANIFEST_KMS_MODE=local
In-process C++ gate	cpp_bridge.enforce_fb / prismmanifest_c shared library
Python hard gate	gate.enforce_group3_boundary (attestation + DAG/hash pin + optional replay)
Consumer verify	prismmanifest.prism_shield.verify_attestation(...)

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.

Plan	Price	Included executions	Headline gates
Developer	$0 / mo	10,000	Python gateway, local keys, 3 policies
Team	$299 / mo	100,000 (+ $0.002 overage)	Cloud KMS, hosted HITL (5 seats)
Business	$999 / mo	500,000 (+ $0.0012 overage)	C++ sidecar, HITL (20 seats), HSM
Enterprise	$35k–$120k+ / yr	Custom	Self-host VPC, SSO, 24/7, SA

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.

Control theme	Prism-Shield capability
Change & config	Versioned policy_id + published schema_hash allowlists
Logical access	Fail-closed KMS; optional bound_tenant_id + allowed_tenant_ids
System operations	FlatBuffer contracts; C++ when DLL loaded (gate_path=inprocess), else Python hard gate
Audit logging	DecisionAuditLog on every decision + escalation queue
Risk mitigation	ACCEPT / REVIEW / REFUSE; strict policies can require span + digit lock

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

Variable / knob	Purpose
PRISMMANIFEST_KMS_STORE	Directory for KMS envelope key material
PRISMMANIFEST_KMS_MODE	local | azure | aws | gcp
PRISMMANIFEST_SHIELD_KEY_ID	Default envelope key id when not passed explicitly
PRISMMANIFEST_ESCALATION_ROOT	HITL queue root for EscalationQueue
PRISMMANIFEST_REVIEW_TOKEN	Shared secret for review UI mutations
PRISMMANIFEST_SHIELD_REPLAY_STORE	Durable replay jsonl path (default under temp/prism_shield); cross-process lock-safe when shared

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