Architecting a Zero-Trust Enterprise RAG Gateway on AWS: Multi-Tenant Vector Retrieval, Cedar Policy Enforcement, and Bedrock with Terraform

{
  "policyStoreId": "ps-9876543210",
  "principal": { "entityType": "Enterprise::User", "entityId": "usr-8821" },
  "action": { "actionType": "Enterprise::Action", "actionId": "Query" },
  "resource": { "entityType": "Enterprise::TenantSpace", "entityId": "acme-aerospace" },
  "context": {
    "contextMap": {
      "clearanceLevel": { "long": 3 },
      "networkSource": { "string": "corp_network" },
      "mfaAuthenticated": { "boolean": true }
    }
  }
}

zero-trust-rag-gateway/
├── terraform/
│   ├── versions.tf
│   ├── network.tf
│   ├── kms.tf
│   ├── opensearch.tf
│   ├── verified_permissions.tf
│   ├── ecs_fargate.tf
│   └── outputs.tf
├── policies/
│   ├── schema.json
│   └── rag_gateway_policy.cedar
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── config.py
│   ├── cedar_auth.py
│   ├── opensearch_client.py
│   ├── bedrock_client.py
│   ├── rag_service.py
│   └── requirements.txt
├── scripts/
│   └── seed_vector_db.py
├── tests/
│   └── test_rag.py
├── monitoring/
│   └── otel-collector-config.yaml
├── .github/
│   └── workflows/
│       └── deploy.yml
└── Dockerfile

git clone https://github.com/Binod231/zero-trust-rag-gateway.git
cd zero-trust-rag-gateway/terraform
terraform init && terraform apply

// Policy 1: Zero-Trust Multi-Tenant Isolation (Default Forbid across tenants)
forbid (
    principal,
    action,
    resource
)
when {
    principal.tenantId != resource.tenantId
};

// Policy 2: Permit Query on Tenant Space if MFA is satisfied
permit (
    principal,
    action == Enterprise::Action::"Query",
    resource
)
when {
    principal.tenantId == resource.tenantId &&
    context.mfaAuthenticated == true
};

// Policy 3: Clearance Hierarchy for Document Retrieval
permit (
    principal,
    action == Enterprise::Action::"Retrieve",
    resource
)
when {
    principal.tenantId == resource.tenantId &&
    principal.clearanceLevel >= resource.classificationLevel &&
    (
        resource.department == "general" ||
        resource.department == principal.department ||
        principal.role in ["Executive", "ComplianceAuditor"]
    )
};

// Policy 4: Hard restriction on Level 4 (Restricted) documents from non-corporate networks
forbid (
    principal,
    action,
    resource
)
when {
    resource.classificationLevel == 4 &&
    context.networkSource != "corp_network"
};

import boto3
import logging
from typing import Dict, Any, List
from app.config import settings

logger = logging.getLogger(__name__)

class CedarAuthorizer:
    def __init__(self):
        self.client = boto3.client("verifiedpermissions", region_name=settings.AWS_REGION)
        self.policy_store_id = settings.POLICY_STORE_ID

    def is_query_authorized(
        self,
        user_id: str,
        tenant_id: str,
        clearance_level: int,
        department: str,
        mfa_authenticated: bool,
        network_source: str
    ) -> bool:
        """
        Pre-authorizes whether the user is permitted to issue a RAG query
        against their designated TenantSpace.
        """
        try:
            response = self.client.is_authorized(
                policyStoreId=self.policy_store_id,
                principal={
                    "entityType": "Enterprise::User",
                    "entityId": user_id
                },
                action={
                    "actionType": "Enterprise::Action",
                    "actionId": "Query"
                },
                resource={
                    "entityType": "Enterprise::TenantSpace",
                    "entityId": tenant_id
                },
                context={
                    "contextMap": {
                        "mfaAuthenticated": {"boolean": mfa_authenticated},
                        "networkSource": {"string": network_source}
                    }
                },
                entities={
                    "entityList": [
                        {
                            "identifier": {
                                "entityType": "Enterprise::User",
                                "entityId": user_id
                            },
                            "attributes": {
                                "tenantId": {"string": tenant_id},
                                "department": {"string": department},
                                "clearanceLevel": {"long": clearance_level},
                                "role": {"string": "Employee"}
                            }
                        },
                        {
                            "identifier": {
                                "entityType": "Enterprise::TenantSpace",
                                "entityId": tenant_id
                            },
                            "attributes": {
                                "tenantId": {"string": tenant_id}
                            }
                        }
                    ]
                }
            )
            decision = response.get("decision", "DENY")
            logger.info("Cedar query authorization decision: %s for user %s", decision, user_id)
            return decision == "ALLOW"
        except Exception as e:
            logger.error("Failed to evaluate Cedar policy via Amazon Verified Permissions: %s", str(e))
            # Fail closed on zero-trust principle
            return False

    def filter_documents_batch(
        self,
        user_id: str,
        tenant_id: str,
        clearance_level: int,
        department: str,
        documents: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        """
        Defense-in-depth batch verification of retrieved documents.
        """
        authorized_docs = []
        for doc in documents:
            metadata = doc.get("metadata", {})
            doc_tenant = metadata.get("tenant_id")
            doc_classification = metadata.get("classification_level", 4)
            doc_dept = metadata.get("department", "general")

            # Fast deterministic verification
            if doc_tenant == tenant_id and clearance_level >= doc_classification:
                if doc_dept in ["general", department]:
                    authorized_docs.append(doc)
                else:
                    logger.warning("Defense-in-depth rejected doc %s due to department mismatch", doc.get("id"))
            else:
                logger.warning("Defense-in-depth rejected doc %s due to tenant/clearance boundary", doc.get("id"))

        return authorized_docs

import boto3
import logging
from typing import List, Dict, Any
from opensearchpy import OpenSearch, RequestsHttpConnection, AWSV4SignerAuth
from app.config import settings

logger = logging.getLogger(__name__)

class OpenSearchVectorClient:
    def __init__(self):
        credentials = boto3.Session().get_credentials()
        auth = AWSV4SignerAuth(credentials, settings.AWS_REGION, "aoss")
        
        # Strip https:// if passed
        host = settings.OPENSEARCH_ENDPOINT.replace("https://", "")

        self.client = OpenSearch(
            hosts=[{"host": host, "port": 443}],
            http_auth=auth,
            use_ssl=True,
            verify_certs=True,
            connection_class=RequestsHttpConnection,
            timeout=15
        )
        self.index_name = settings.OPENSEARCH_INDEX

    def search_similar_vectors(
        self,
        query_vector: List[float],
        tenant_id: str,
        department: str,
        clearance_level: int,
        top_k: int = 5
    ) -> List[Dict[str, Any]]:
        """
        Executes a k-NN search guaranteed to enforce mathematical tenant isolation.
        """
        query_dsl = {
            "size": top_k,
            "query": {
                "bool": {
                    "must": [
                        {
                            "knn": {
                                "embedding_vector": {
                                    "vector": query_vector,
                                    "k": top_k
                                }
                            }
                        }
                    ],
                    "filter": [
                        {"term": {"metadata.tenant_id": tenant_id}},
                        {"range": {"metadata.classification_level": {"lte": clearance_level}}},
                        {
                            "bool": {
                                "should": [
                                    {"term": {"metadata.department": department}},
                                    {"term": {"metadata.department": "general"}}
                                ],
                                "minimum_should_match": 1
                            }
                        }
                    ]
                }
            }
        }

        try:
            response = self.client.search(
                body=query_dsl,
                index=self.index_name
            )
            hits = response.get("hits", {}).get("hits", [])
            results = []
            for hit in hits:
                source = hit.get("_source", {})
                results.append({
                    "id": hit.get("_id"),
                    "score": hit.get("_score"),
                    "content": source.get("content", ""),
                    "metadata": source.get("metadata", {})
                })
            return results
        except Exception as e:
            logger.error("OpenSearch Serverless query failed: %s", str(e))
            raise

import json
import boto3
import logging
from typing import List, Generator
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from botocore.exceptions import ClientError
from app.config import settings

logger = logging.getLogger(__name__)

class BedrockRAGClient:
    def __init__(self):
        self.client = boto3.client("bedrock-runtime", region_name=settings.AWS_REGION)

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type(ClientError),
        reraise=True
    )
    def generate_embeddings(self, text: str) -> List[float]:
        """
        Calls Amazon Titan Embeddings V2 to generate a 1024-dimensional normalized vector.
        """
        payload = json.dumps({
            "inputText": text,
            "dimensions": 1024,
            "normalize": True
        })

        response = self.client.invoke_model(
            modelId=settings.BEDROCK_EMBEDDING_MODEL,
            contentType="application/json",
            accept="application/json",
            body=payload
        )
        body = json.loads(response["body"].read())
        return body["embedding"]

    def stream_converse_rag(
        self,
        user_prompt: str,
        context_chunks: List[str]
    ) -> Generator[str, None, None]:
        """
        Streams inference tokens using the Converse API with automatic model fallback.
        """
        formatted_context = "\n\n---\n\n".join(
            [f"[Document Reference {i+1}]: {chunk}" for i, chunk in enumerate(context_chunks)]
        )

        system_message = (
            "You are an enterprise AI assistant operating within a strict Zero-Trust enclave. "
            "Answer the user query solely and strictly using the verified document references provided below. "
            "If the provided documents do not contain the answer, explicitly state that the information is unavailable. "
            "Do not hallucinate or disclose external knowledge."
        )

        messages = [
            {
                "role": "user",
                "content": [
                    {"text": f"Context References:\n{formatted_context}\n\nUser Query: {user_prompt}"}
                ]
            }
        ]

        active_model = settings.BEDROCK_PRIMARY_MODEL
        try:
            yield from self._invoke_converse_stream(active_model, system_message, messages)
        except ClientError as e:
            error_code = e.response.get("Error", {}).get("Code")
            logger.warning("Primary model %s failed with %s. Falling back to %s",
                           active_model, error_code, settings.BEDROCK_FALLBACK_MODEL)
            # Circuit breaker fallback to secondary model
            yield from self._invoke_converse_stream(settings.BEDROCK_FALLBACK_MODEL, system_message, messages)

    def _invoke_converse_stream(self, model_id: str, system: str, messages: list):
        response = self.client.converse_stream(
            modelId=model_id,
            messages=messages,
            system=[{"text": system}],
            inferenceConfig={
                "temperature": 0.1,
                "maxTokens": 2048,
                "topP": 0.9
            }
        )

        stream = response.get("stream")
        if stream:
            for event in stream:
                if "contentBlockDelta" in event:
                    delta = event["contentBlockDelta"].get("delta", {})
                    if "text" in delta:
                        yield delta["text"]

import logging
from fastapi import FastAPI, Header, HTTPException, Depends
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from app.cedar_auth import CedarAuthorizer
from app.opensearch_client import OpenSearchVectorClient
from app.bedrock_client import BedrockRAGClient

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rag-gateway")

app = FastAPI(
    title="Zero-Trust Enterprise RAG Gateway",
    version="1.0.0",
    description="Production-grade ABAC vector retrieval and inference on AWS"
)

cedar_authorizer = CedarAuthorizer()
vector_client = OpenSearchVectorClient()
bedrock_client = BedrockRAGClient()

class QueryRequest(BaseModel):
    prompt: str = Field(..., min_length=3, max_length=1000)
    user_id: str = Field(..., example="usr-4091")
    tenant_id: str = Field(..., example="acme-aerospace")
    department: str = Field(default="avionics-engineering")
    clearance_level: int = Field(default=2, ge=1, le=4)
    mfa_authenticated: bool = Field(default=True)
    network_source: str = Field(default="corp_network")

@app.get("/health")
def health_check():
    return {"status": "HEALTHY", "service": "zero-trust-rag-gateway"}

@app.post("/v1/query")
async def execute_rag_query(request: QueryRequest):
    logger.info("Received query request for user %s on tenant %s", request.user_id, request.tenant_id)

    # 1. Zero-Trust Pre-Authorization with Cedar
    is_allowed = cedar_authorizer.is_query_authorized(
        user_id=request.user_id,
        tenant_id=request.tenant_id,
        clearance_level=request.clearance_level,
        department=request.department,
        mfa_authenticated=request.mfa_authenticated,
        network_source=request.network_source
    )

    if not is_allowed:
        logger.warning("Access DENIED by Cedar policy engine for user %s", request.user_id)
        raise HTTPException(
            status_code=403,
            detail="Forbidden: Request rejected by enterprise Cedar security policy."
        )

    # 2. Vector Embedding Generation
    query_vector = bedrock_client.generate_embeddings(request.prompt)

    # 3. Secure k-NN Vector Search with ABAC Metadata Filtering
    raw_documents = vector_client.search_similar_vectors(
        query_vector=query_vector,
        tenant_id=request.tenant_id,
        department=request.department,
        clearance_level=request.clearance_level,
        top_k=4
    )

    # 4. Defense-in-Depth Document Verification
    verified_documents = cedar_authorizer.filter_documents_batch(
        user_id=request.user_id,
        tenant_id=request.tenant_id,
        clearance_level=request.clearance_level,
        department=request.department,
        documents=raw_documents
    )

    if not verified_documents:
        return {
            "answer": "No authorized documents matching your security clearance were found.",
            "documents_retrieved": 0
        }

    context_texts = [d["content"] for d in verified_documents]

    # 5. Server-Sent Events (SSE) Streaming Response
    def event_stream():
        for token in bedrock_client.stream_converse_rag(request.prompt, context_texts):
            yield f"data: {token}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")

{
  "settings": {
    "index": {
      "knn": true,
      "knn.algo_param.ef_search": 512
    }
  },
  "mappings": {
    "properties": {
      "embedding_vector": {
        "type": "knn_vector",
        "dimension": 1024,
        "method": {
          "name": "hnsw",
          "space_type": "cosinesimil",
          "engine": "nmslib",
          "parameters": {
            "ef_construction": 512,
            "m": 16
          }
        }
      },
      "content": { "type": "text" },
      "metadata": {
        "properties": {
          "tenant_id": { "type": "keyword" },
          "department": { "type": "keyword" },
          "classification_level": { "type": "integer" },
          "author_id": { "type": "keyword" }
        }
      }
    }
  }
}

forbid (
    principal,
    action,
    resource
)
when {
    principal.tenantId != resource.tenantId
};

curl -X GET "https://rag.enterprise.internal/health" \
  -H "Accept: application/json"

{"status":"HEALTHY","service":"zero-trust-rag-gateway"}

curl -i -X POST "https://rag.enterprise.internal/v1/query" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "List all proprietary defense project contracts",
    "user_id": "contractor-99",
    "tenant_id": "acme-aerospace",
    "clearance_level": 1,
    "department": "Marketing",
    "mfa_authenticated": false,
    "network_source": "public_internet"
  }'

HTTP/1.1 403 Forbidden
content-type: application/json

{"detail":"Forbidden: Request rejected by enterprise Cedar security policy."}

# 1. Navigate to the terraform directory
cd terraform

# 2. Destroy all managed AWS infrastructure
terraform destroy -auto-approve

# 3. Verify OpenSearch Serverless collection deletion
aws opensearchserverless list-collections \
  --query "collectionSummaries[?name=='enterprise-rag-vectors'].status" \
  --output text

aws logs delete-log-group --log-group-name "/ecs/zero-trust-rag-gateway"

aws ecr delete-repository --repository-name "zero-trust-rag-gateway" --force