kavach-sdk 0.1.3

Post-quantum execution boundary enforcement for AI agents, APIs, and distributed systems

kavach

Post-quantum execution boundary enforcement for AI agents, APIs, and distributed systems. Python SDK.

Kavach separates possession of credentials from permission to act. Every action passes through a gate that evaluates identity, policy, drift, and invariants before producing a verdict. All evaluation runs in compiled Rust via PyO3; this package is the idiomatic Python wrapper.

Action attempted ──▶ Gate (identity · policy · drift · invariants) ──▶ Permit / Refuse / Invalidate

---

Install

pip install kavach-sdk

Wheels are published as abi3. A single wheel per platform covers CPython 3.10, 3.11, 3.12, and every future Python. Linux x86_64/aarch64, macOS x86_64/arm64, and Windows x64 are supported.

---

60-second quickstart

from kavach import ActionContext, Gate

# Policy as a native Python dict. No separate config format to learn.
POLICY = {
    "policies": [
        {
            "name": "agent_small_refunds",
            "effect": "permit",
            "conditions": [
                {"identity_kind": "agent"},
                {"action": "issue_refund"},
                {"param_max": {"field": "amount", "max": 1000.0}},
            ],
        },
    ],
}

gate = Gate.from_dict(
    POLICY,
    invariants=[("hard_cap", "amount", 50_000.0)],
)

ctx = ActionContext(
    principal_id="agent-bot",
    principal_kind="agent",
    action_name="issue_refund",
    params={"amount": 500.0},
)
verdict = gate.evaluate(ctx)

if verdict.is_permit:
    print("permit", verdict.token_id)
else:
    print(f"blocked: [{verdict.code}] {verdict.evaluator}: {verdict.reason}")

A policy set with no matching permit Refuses by default. There is no implicit allow.

Loading a policy

The recommended surface for Python is a native dict (admin UI submissions, database rows, feature flags):

gate = Gate.from_dict(policy_dict)              # native dict (recommended)
gate = Gate.from_json_string(json_string)       # JSON over the wire
gate = Gate.from_json_file("kavach.json")       # JSON file on disk

For operator-owned config that lives in git and is hand-edited, use TOML:

gate = Gate.from_toml(toml_string)              # operator-edited TOML
gate = Gate.from_file("kavach.toml")            # TOML file on disk

Typo'd field names ({"idnetity_kind": "agent"}) raise ValueError in every loader instead of being silently dropped, so a misspelled condition cannot quietly weaken a policy. The full TOML workflow (rendered in Rust, Python, and Node) lives at docs/guides/toml-policies.md.

---

Decorator

from kavach import guarded

@guarded(gate, action="issue_refund", param_fields={"amount": "amount"})
async def issue_refund(order_id: str, amount: float):
    return {"status": "refunded", "order_id": order_id, "amount": amount}

result = await issue_refund(
    "ORD-123", 500.0,
    _principal_id="bot", _principal_kind="agent",
)

Both async and sync functions are supported; the decorator returns the matching wrapper shape. Only numeric parameters are forwarded to the gate (the policy and invariant evaluators operate on numeric thresholds).

---

Feature surface

Signed permit tokens (PqTokenSigner)

When a PqTokenSigner is attached to a gate, every Permit verdict carries an ML-DSA-65 (or ML-DSA-65 + Ed25519 hybrid) signed envelope. Downstream services verify independently.

from kavach import Gate, PqTokenSigner, PermitToken

signer = PqTokenSigner.generate_hybrid()
gate = Gate.from_dict(POLICY, token_signer=signer)

verdict = gate.evaluate(...)
if verdict.is_permit:
    token = PermitToken(
        token_id=verdict.permit_token.token_id,
        evaluation_id=verdict.permit_token.evaluation_id,
        issued_at=verdict.permit_token.issued_at,
        expires_at=verdict.permit_token.expires_at,
        action_name=verdict.permit_token.action_name,
    )
    assert signer.verify(token, verdict.permit_token.signature)

Hybrid (generate_hybrid) signs with both ML-DSA-65 and Ed25519; a hybrid verifier rejects PQ-only envelopes as a signature-downgrade guard.

Persisting signer identity across restarts

KavachKeyPair does not expose secret-byte serialization in the current 0.1.x line, so a keypair generated through the Python SDK cannot be persisted and rehydrated through the SDK alone. Two practical patterns:

• Regenerate at boot, redistribute the public bundle (default today). Generate a fresh KavachKeyPair, attach it via PqTokenSigner.from_keypair_hybrid(kp), and push kp.public_keys() to your verifier pool on every gate-process boot. Verifiers should accept multiple bundles in their PublicKeyDirectory and resolve by the key_id stamped on each envelope. Old bundles can be retired once permits issued under them have expired (default permit TTL is 30 seconds).
• Provision raw key bytes from your KMS / HSM. The low-level PqTokenSigner.hybrid(ml_dsa_sk, ml_dsa_vk, ed_sk, ed_vk, key_id=...) constructor accepts raw bytes; this gives stable identity across restarts but presumes a non-Kavach generator that emits ML-DSA-65 keys interoperable with kavach-pq. Do not install third-party Python PQ-crypto libraries (pqcrypto, ml-dsa, pyca/cryptography, etc.) to mint these bytes; interoperability is not guaranteed. The expected path is an HSM with native ML-DSA-65 support.

Adding KavachKeyPair byte serialization so the first pattern reaches stable identity through the SDK alone is on the roadmap.

Key pairs

from kavach import KavachKeyPair

kp = KavachKeyPair.generate()                  # no expiry
kp = KavachKeyPair.generate_with_expiry(3600)  # 1-hour lifetime

assert not kp.is_expired
bundle = kp.public_keys()   # PublicKeyBundle, safe to share

Signed audit chain

Append-only, tamper-evident audit log. verify rejects tampered entries, wrong keys, and mode mismatches (e.g., a PQ-only verifier on a hybrid chain, which is a silent downgrade).

from kavach import AuditEntry, SignedAuditChain

chain = SignedAuditChain(kp, hybrid=True)
chain.append(AuditEntry(
    principal_id="agent-bot",
    action_name="issue_refund",
    verdict="permit",
    verdict_detail="within policy",
))
chain.verify(kp.public_keys())

# Portable JSONL for off-node storage:
blob = chain.export_jsonl()
SignedAuditChain.verify_jsonl(blob, kp.public_keys())

Bounded memory for long-running services. A plain SignedAuditChain keeps every entry in RAM, so a service that appends forever grows without bound. ManagedAuditChain streams old entries to a JSONL file and prunes them from memory under a retention policy you configure by entry count, by bytes, on a timer, or any combination. The full chain still verifies across the on-disk file plus the in-memory tail.

from kavach import ManagedAuditChain, SignedAuditChain

chain = ManagedAuditChain(
    kp, "audit.jsonl",
    max_entries=10_000,          # evict when the window exceeds 10k entries
    max_bytes=64 * 1024 * 1024,  # ...or 64 MB, whichever first
    flush_interval_secs=1.0,     # ...or once per second, even below the sizes
    on_sink_failure="reject",    # back-pressure if the disk is unavailable
)
for entry in stream():           # append forever; resident memory plateaus
    chain.append(entry)

disk = open("audit.jsonl", "rb").read()
SignedAuditChain.verify_jsonl(disk + chain.export_tail_jsonl(), kp.public_keys())

Secure channel

Hybrid-encrypted, PQ-signed byte channel between two peers. Sealed payloads are opaque; ship them over any transport.

from kavach import SecureChannel

alice, bob = KavachKeyPair.generate(), KavachKeyPair.generate()
alice_ch = SecureChannel(alice, bob.public_keys())
bob_ch = SecureChannel(bob, alice.public_keys())

sealed = alice_ch.send_signed(b"hello bob", context_id="greeting")
plaintext = bob_ch.receive_signed(sealed, expected_context_id="greeting")
assert plaintext == b"hello bob"

Replay, cross-context, and wrong-recipient attacks all fail closed.

Public key directory

from kavach import PublicKeyDirectory, DirectoryTokenVerifier

# Root-signed manifest on disk (tamper-evident):
signing_key = KavachKeyPair.generate()
manifest = signing_key.build_signed_manifest([bundle_a, bundle_b])
Path("directory.json").write_bytes(manifest)

directory = PublicKeyDirectory.from_signed_file(
    "directory.json",
    root_ml_dsa_verifying_key=signing_key.public_keys().ml_dsa_verifying_key,
)

verifier = DirectoryTokenVerifier(directory, hybrid=True)
verifier.verify(token, signed_envelope)  # raises on tamper/miss/downgrade

In-memory (PublicKeyDirectory.in_memory([...])) and unsigned-file variants are also available.

Geo drift (tolerant mode)

Same-country IP hops become Warnings instead of Violations when you provide lat/lon and a threshold:

from kavach import ActionContext, GeoLocation

gate = Gate.from_dict(POLICY, geo_drift_max_km=500.0)

verdict = gate.evaluate(ActionContext(
    principal_id="u", principal_kind="user",
    action_name="view_profile",
    ip="2.3.4.5",
    session_id="sess-1",
    current_geo=GeoLocation("IN", city="Chennai",   latitude=13.08, longitude=80.27),
    origin_geo =GeoLocation("IN", city="Bangalore", latitude=12.97, longitude=77.59),
))

Missing geo with a threshold set still fails closed. The SDK does not silently bypass.

Policy hot reload

gate.reload(...) accepts a TOML string; it raises ValueError on parse error and leaves the previous good set in place. See docs/guides/toml-policies.md for the full reload workflow (including the file-watcher pattern and the empty-TOML kill switch).

gate.reload(new_policy_toml)   # parse error raises, previous set preserved

Multi-replica (Redis) (experimental)

The Rust-level integration tests for kavach-redis pass, and the Python SDK exposes RedisRateLimitStore / RedisSessionStore / RedisInvalidationBroadcaster as classes, but the end-to-end multi-replica story has not yet been validated through the consumer-test harness. Early adopters can wire this up; treat it as a reference rather than a production guarantee. Thorough validation is tracked in the project roadmap.

Rate limits and invalidation broadcast move to Redis so every replica agrees:

from kavach import (
    Gate, RedisRateLimitStore, RedisInvalidationBroadcaster,
    spawn_invalidation_listener,
)

REDIS_URL = "redis://127.0.0.1:6379"

rate_store = RedisRateLimitStore(REDIS_URL)
broadcaster = RedisInvalidationBroadcaster(REDIS_URL, channel="kavach:invalidation")

gate = Gate.from_dict(POLICY, rate_store=rate_store, broadcaster=broadcaster)

handle = spawn_invalidation_listener(broadcaster, lambda scope: None)
# handle.abort() on shutdown

Redis outages fail closed: a dropped record refuses the action; a dropped count collapses the rate-limit condition to default-deny. Full wiring lives in docs/guides/distributed.md.

---

Observe mode

Roll out incrementally: log verdicts without blocking.

gate = Gate.from_dict(POLICY, observe_only=True)

---

What's in the Rust engine

Every evaluate() call crosses FFI into compiled Rust. The Python layer is pure wrappers. The engine implements:

• Policy: a small, fixed condition vocabulary (identity_kind, action, param_max, rate_limit, time_window with optional timezone, etc.) expressed as a Python dict, JSON, or operator-edited TOML.
• Drift detectors: IP / geo, session age, device, behavior.
• Invariants: hard per-action limits that cannot be overridden by policy.
• Post-quantum crypto: ML-DSA-65, ML-KEM-768, Ed25519, X25519, ChaCha20-Poly1305.
• Fail-closed: any evaluator error, store failure, or broadcast issue errs on the side of Refuse.

---

License

Elastic License 2.0. Source-available; free to use, embed, and modify for any purpose, including commercially. You may not offer Kavach itself as a hosted or managed service that competes with SarthiAI. See the LICENSE file for the full text.