Blockchain
Verifiable RAG on AWS: Cryptographic Provenance for Retrieval Results
A regulated enterprise wants RAG outputs that are auditable and tamper-evident. Their concern is not only hallucination, but also poisoned corpora and undocumented retrieval provenance.
Verifiable RAG on AWS: Cryptographic Provenance for Retrieval Results
Scenario
A regulated enterprise wants RAG outputs that are auditable and tamper-evident. Their concern is not only hallucination, but also poisoned corpora and undocumented retrieval provenance.
Problem statement
Traditional RAG can cite chunks, but citations alone do not guarantee integrity. Security teams need cryptographic evidence that:
- chunk content was not modified after indexing
- retrieved chunks belong to an approved corpus version
- answer claims were grounded in verified evidence
Architecture summary
graph TD
Docs[Documents] --> Chunk[Chunk + Embed Pipeline]
Chunk --> Sign[Chunk Hash + Signature]
Sign --> Merkle[Merkle Root Builder]
Sign --> Vector[(Vector Index)]
Merkle --> Ledger[(Immutable Root Store: QLDB or Blockchain Anchor)]
Query[User Query] --> Retrieve[Retriever]
Retrieve --> Verify[Signature + Merkle Proof Verification]
Verify --> LLM[Answer Generation]
Verify --> Audit[(Audit Log + Evidence Receipt)]
Trade-offs
- Stronger integrity verification adds latency and operational complexity.
- Provenance verification proves origin, not truthfulness.
- Best pattern is provenance + semantic verification + curation.
Implementation tutorial
1) Build signed chunks
import hashlib
import json
from nacl.signing import SigningKey
signing_key = SigningKey.generate()
verify_key = signing_key.verify_key
def sign_chunk(chunk_text: str) -> dict:
digest = hashlib.sha256(chunk_text.encode("utf-8")).hexdigest()
sig = signing_key.sign(digest.encode("utf-8")).signature.hex()
return {"chunk": chunk_text, "sha256": digest, "signature": sig}
2) Construct a Merkle root
import hashlib
def h(x: str) -> str:
return hashlib.sha256(x.encode("utf-8")).hexdigest()
def merkle_root(leaves: list[str]) -> str:
nodes = [h(v) for v in leaves]
if not nodes:
return ""
while len(nodes) > 1:
if len(nodes) % 2 == 1:
nodes.append(nodes[-1])
nodes = [h(nodes[i] + nodes[i+1]) for i in range(0, len(nodes), 2)]
return nodes[0]
3) Store corpus root and metadata
export AWS_REGION=us-east-1
export PROJECT=verifiable-rag
export ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
aws dynamodb create-table \
--table-name ${PROJECT}-corpus-roots \
--attribute-definitions AttributeName=corpus_id,AttributeType=S AttributeName=version,AttributeType=S \
--key-schema AttributeName=corpus_id,KeyType=HASH AttributeName=version,KeyType=RANGE \
--billing-mode PAY_PER_REQUEST \
--sse-specification Enabled=true
$env:AWS_REGION = "us-east-1"
$env:PROJECT = "verifiable-rag"
$env:ACCOUNT_ID = (aws sts get-caller-identity --query Account --output text)
aws dynamodb create-table `
--table-name "$($env:PROJECT)-corpus-roots" `
--attribute-definitions AttributeName=corpus_id,AttributeType=S AttributeName=version,AttributeType=S `
--key-schema AttributeName=corpus_id,KeyType=HASH AttributeName=version,KeyType=RANGE `
--billing-mode PAY_PER_REQUEST `
--sse-specification Enabled=true
4) Verification at query time
from nacl.signing import VerifyKey
def verify_chunk_signature(verify_key_hex: str, digest: str, signature_hex: str) -> bool:
vk = VerifyKey(bytes.fromhex(verify_key_hex))
try:
vk.verify(digest.encode("utf-8"), bytes.fromhex(signature_hex))
return True
except Exception:
return False
5) FastAPI middleware for proof-gated answers
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.post("/answer")
def answer(payload: dict):
verified_chunks = payload.get("verified_chunks", [])
if not verified_chunks:
raise HTTPException(status_code=412, detail="No verified evidence available")
# Call LLM only after verification passes.
return {"status": "ok", "evidence_count": len(verified_chunks)}
6) Optional anchoring pattern
You can store Merkle roots in:
- AWS QLDB for immutable journals
- a chosen public chain for external transparency
Use async anchoring jobs so retrieval path remains low latency.
Security controls
- keep signing keys in AWS KMS/HSM-backed workflows
- isolate indexing and serving roles
- require proof verification before answer release
- log every failed verification event
Monitoring
Track:
- verification pass rate
- proof-generation latency
- rejected responses due to missing proofs
- corpus version drift
Cost optimization
- verify top-k only, not full corpus
- cache proof bundles for frequent queries
- batch anchoring operations
Pricing reminder: verify current costs for DynamoDB, KMS, Lambda, and chosen anchoring service.
Production checklist
- Key rotation process defined
- Corpus versioning policy documented
- Proof validation integrated into runtime gate
- Incident playbook for signature verification failures
- Red-team scenarios include corpus poisoning
References
- https://docs.aws.amazon.com/prescriptive-guidance/latest/retrieval-augmented-generation-options/choosing-option.html
- https://aws.amazon.com/blogs/security/
- https://himjoe.github.io/proof-carrying-answers/
Source
platform/archive/articles/verifiable-rag-on-aws-cryptographic-provenance.md