+─────────────────────────────────────────────────────────────────────────────+
|               FINANCIAL DATA PRIVACY & COMPLIANCE STATUTES                  |
+─────────────────────────────────────────────────────────────────────────────+
|                                                                             |
|  1. Gramm-Leach-Bliley Act (GLBA Safeguards Rule - 16 CFR Part 314):        |
|     Mandates administrative and technical safeguards for Non-Public         |
|     Personal Information (NPI). Prohibits transmission to third-party       |
|     multitenant clouds without explicit custodial data agreements.          |
|                                                                             |
|  2. SEC Rule 17a-4 & FINRA Rule 4511 (Books and Records):                   |
|     Requires all electronic records, audit trails, and financial queries   |
|     to be stored in Write-Once-Read-Many (WORM) immutable formats with      |
|     verifiable cryptographic timestamps. Multitenant APIs cannot provide   |
|     reproducible, tamper-evident query verification.                        |
|                                                                             |
|  3. GDPR Article 9 & 28 / BaFin Cloud Banking Circular 10/2018:             |
|     Restricts cross-border data egress for banking secrets and customer     |
|     financial profiles. Prohibits model training on custodial accounts.     |
|                                                                             |
|  4. PCI Security Standards Council (PCI-DSS v4.0 Requirement 3.4):          |
|     Primary Account Numbers (PAN) and sensitive authentication data         |
|     must be cryptographically rendered unreadable across all endpoints.    |
|                                                                             |
+─────────────────────────────────────────────────────────────────────────────+

flowchart TD
    Client["Financial Auditor / Analyst<br/>(Internal Banking Portal)"] -->|mTLS 1.3 Strict Auth| Ingress["Zero-Trust API Gateway<br/>(RBAC / ABAC Verification)"]
    
    subgraph VPC_SECURE_PERIMETER ["Air-Gapped Private VPC Enclave (Zero WAN Egress)"]
        direction TB
        
        Ingress --> Redaction["Streaming PII/NPI Redaction Gateway<br/>(HMAC-SHA256 Token Vault)"]
        
        subgraph RETRIEVAL_ENGINE ["Dual-Index Hybrid Retrieval Mesh"]
            Redaction --> QueryRouter{"Intent Classifier<br/>(Structured vs. Unstructured)"}
            
            QueryRouter -->|Structured Ledger Balance| TextToSQL["Deterministic Text-to-SQL<br/>(Constrained AST Validator)"]
            TextToSQL --> LedgerDB[("PostgreSQL 16 & ClickHouse<br/>(Double-Entry Journals + RLS)")]
            
            QueryRouter -->|Unstructured Audit Memo| DenseEmbed["Local Embedding Tensor<br/>(BGE-large-en-v1.5 on GPU)"]
            DenseEmbed --> VectorDB[("pgvector HNSW Store<br/>(SOX Workpapers & SEC Filings)")]
        end
        
        subgraph CONFIDENTIAL_COMPUTE ["Nitro Enclave / AMD SEV-SNP Memory Sandbox"]
            RRF["Reciprocal Rank Fusion (RRF)<br/>Context Assembly & Source Citations"]
            LedgerDB -.-> RRF
            VectorDB -.-> RRF
            
            RRF --> LocalLLM["vLLM / TensorRT-LLM Inference Node<br/>(Llama 3.1 70B / Mixtral 8x22B)"]
        end
        
        LocalLLM --> ResponseValidator{"Deterministic Output Gate<br/>(JSON Schema / Hallucination Check)"}
    end
    ResponseValidator -->|Cryptographically Verified Answer| Client
    ResponseValidator -.->|WORM Audit Envelope| S3WORM[("SEC 17a-4 Compliant WORM Vault<br/>(Immutable S3 Object Lock)")]

+─────────────────────────────────────────────────────────────────────────────+
|               AIR-GAPPED PRIVATE VPC RETRIEVAL TOPOLOGY                     |
+─────────────────────────────────────────────────────────────────────────────+
|                                                                             |
|  [Auditor Web Client] ──► [mTLS 1.3 Ingress] ──► [PII Redaction Gateway]    |
|                                                          │                  |
|          ┌───────────────────────────────────────────────┘                  |
|          ▼                                                                  |
|  [Dual-Path Retrieval Mesh]                                                 |
|  ├─► Structured Ledger: Deterministic Text-to-SQL (PostgreSQL 16 / RLS)    |
|  └─► Unstructured Memos: Dense pgvector HNSW (BGE-large-en-v1.5)            |
|          │                                                                  |
|          ▼ (Strict In-Memory Aggregation: Reciprocal Rank Fusion)           |
|  [Hardware-Isolated Confidential Sandbox (AMD SEV-SNP / Nitro Enclave)]     |
|  - In-VPC Model Serving: vLLM Running Llama 3.1 70B Instruct                |
|  - Network Policy: 0.0.0.0/0 Explicitly DROPPED (Zero WAN Gateway)          |
|  - Deterministic JSON Schema Enforcement                                    |
|          │                                                                  |
|          ├─────────────────────────────────────────────────┐                |
|          ▼                                                 ▼                |
|  [Verified Auditor Response]                  [SEC 17a-4 WORM Audit Vault]   |
|  - Grounded Ledger Citation                   - SHA-256 Hash of Prompt/SQL  |
|  - Mathematical Reconciliation                - Immutable 7-Year Retention  |
|                                                                             |
+─────────────────────────────────────────────────────────────────────────────+

# Cilium Network Policy: Air-Gapped Financial AI Enclave
apiVersion: "cilium.io/v2"
kind: CiliumNetworkPolicy
metadata:
  name: "enforce-financial-ai-airgap"
  namespace: "fintech-ai-core"
spec:
  endpointSelector:
    matchLabels:
      app: "private-rag-engine"
  ingress:
fromEndpoints:
matchLabels:
            app: "internal-banking-gateway"
      toPorts:
ports:
port: "8443"
              protocol: "TCP"
  egress:
    # Allow communication ONLY to internal PostgreSQL and ClickHouse endpoints
toEndpoints:
matchLabels:
            app: "core-ledger-postgres"
      toPorts:
ports:
port: "5432"
              protocol: "TCP"
toEndpoints:
matchLabels:
            app: "sec17a4-audit-vault"
      toPorts:
ports:
port: "9000"
              protocol: "TCP"
    # EXPLICIT DENY: All external CIDRs (0.0.0.0/0) dropped at kernel layer

+─────────────────────────────────────────────────────────────────────────────+
|                MATHEMATICAL FORMULATIONS: HYBRID RETRIEVAL                  |
+─────────────────────────────────────────────────────────────────────────────+
|                                                                             |
|  1. Reciprocal Rank Fusion (RRF) for Dual-Mesh Alignment:                   |
|                                                                             |
|     RRF_Score(d in D) = Sum_{m in M} ( w_m / ( k + r_m(d) ) )               |
|                                                                             |
|     Where:                                                                  |
|     - M = { Dense Vector Search (HNSW), Sparse Lexical (BM25) }             |
|     - r_m(d) is the ordinal rank of document d in retrieval model m         |
|     - k is the rank smoothing hyperparameter (typically k = 60)             |
|     - w_m is the domain weight (w_dense = 0.65, w_sparse = 0.35)            |
|                                                                             |
|  2. Cosine Similarity with Row-Level Security Masking:                      |
|                                                                             |
|     Sim(q, v_i) = ( q . v_i ) / ( ||q||  ||v_i|| )  M_RLS(u, entity_i)   |
|                                                                             |
|     Where M_RLS(u, entity_i) evaluates to:                                  |
|     - 1.0 if user u possesses cryptographic clearance for legal entity i    |
|     - 0.0 (Hard Mask) if clearance is lacking, nullifying vector score      |
|                                                                             |
|  3. SEC 17a-4 Cryptographic Audit Envelope Hash:                            |
|                                                                             |
|     H_audit = SHA-256( Prompt || SQL_Query || Chunk_Hashes || Model_Output  |
|                        || Timestamp || Prev_Block_Hash )                    |
|                                                                             |
+─────────────────────────────────────────────────────────────────────────────+

"""
scripts/financial_pii_sanitizer.py
Technology: Python 3.11, Regex, Microsoft Presidio Core, Cryptography
Function: High-throughput, deterministic Non-Public Personal Information (NPI) redaction
"""
import re
import hmac
import hashlib
from typing import Dict, Tuple
class FinancialDataSanitizer:
    def __init__(self, vault_secret_key: bytes):
        self.secret_key = vault_secret_key
        
        # Financial Regex Patterns (PCI-DSS & GLBA Scope)
        self.patterns = {
            # Primary Account Numbers (13 to 19 digits with Luhn algorithm validation)
            "PAN": re.compile(r'\b(?:\d{4}[-\s]?){3}\d{4,7}\b'),
            # US Social Security Numbers (SSN)
            "SSN": re.compile(r'\b\d{3}-\d{2}-\d{4}\b'),
            # International Bank Account Numbers (IBAN)
            "IBAN": re.compile(r'\b[A-Z]{2}\d{2}[A-Z0-9]{4}\d{7}([A-Z0-9]?){0,16}\b'),
            # SWIFT Business Identifier Codes (BIC)
            "SWIFT_BIC": re.compile(r'\b[A-Z]{6}[A-Z0-9]{2}([A-Z0-9]{3})?\b'),
            # Employer Identification Numbers (EIN)
            "EIN": re.compile(r'\b\d{2}-\d{7}\b'),
        }
    def _generate_hmac_token(self, entity_type: str, raw_value: str) -> str:
        """Generates deterministic pseudo-token for entity without exposing raw value"""
        clean_value = re.sub(r'[-\s]', '', raw_value)
        h = hmac.new(self.secret_key, clean_value.encode('utf-8'), hashlib.sha256)
        token_hash = h.hexdigest()[:12]
        return f"[TOKEN_{entity_type}_{token_hash}]"
    def sanitize_text(self, text: str) -> Tuple[str, Dict[str, str]]:
        """
        Sanitizes text, replacing sensitive NPI with reversible tokens.
        Returns: (sanitized_text, token_mapping_vault)
        """
        sanitized_text = text
        token_vault: Dict[str, str] = {}
        # 1. Redact and tokenize structured financial entities
        for entity_type, regex in self.patterns.items():
            matches = regex.findall(sanitized_text)
            for match in matches:
                # Handle regex match tuples
                raw_match = match if isinstance(match, str) else match[0]
                if not raw_match:
                    continue
                
                # Verify Luhn checksum if evaluating credit card PAN
                if entity_type == "PAN" and not self._verify_luhn(raw_match):
                    continue
                token = self._generate_hmac_token(entity_type, raw_match)
                sanitized_text = sanitized_text.replace(raw_match, token)
                token_vault[token] = raw_match
        return sanitized_text, token_vault
    def desanitize_output(self, generated_text: str, token_vault: Dict[str, str]) -> str:
        """Restores original values on final client-side render if authorized"""
        restored = generated_text
        for token, original in token_vault.items():
            restored = restored.replace(token, original)
        return restored
    @staticmethod
    def _verify_luhn(card_number: str) -> bool:
        """Standard Luhn checksum verification for credit card PANs"""
        digits = [int(c) for c in re.sub(r'\D', '', card_number)]
        if len(digits) < 13 or len(digits) > 19:
            return False
        checksum = 0
        reverse_digits = digits[::-1]
        for i, d in enumerate(reverse_digits):
            if i % 2 == 1:
                doubled = d  2
                checksum += (doubled - 9) if doubled > 9 else doubled
            else:
                checksum += d
        return checksum % 10 == 0
Usage Example

if __name__ == "__main__":
    secret = b"hardware_enclave_secret_key_884912"
    sanitizer = FinancialDataSanitizer(secret)
    
    audit_memo = (
        "Reconciliation Report: Customer John Doe transferred $1,450,000 from "
        "Account 4532-0192-8834-1120 (IBAN DE89370400440532013000) to Barclays SWIFT BARCGB22."
    )
    
    clean_memo, vault = sanitizer.sanitize_text(audit_memo)
    print("Sanitized for In-VPC LLM:")
    print(clean_memo)
    print("\nIsolated In-Memory Token Vault:")
    print(vault)

-- database/schema/financial_audit_rag.sql
-- PostgreSQL 16 Enterprise with pgvector & RLS Enforcement
CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS pg_trgm;
-- Legal Entity Directory (Multi-Tenant Banking Boundaries)
CREATE TABLE legal_entities (
    entity_id VARCHAR(32) PRIMARY KEY,
    name VARCHAR(128) NOT NULL,
    jurisdiction VARCHAR(3) NOT NULL -- e.g. USA, GBR, DEU, CHE
);
-- Audit Documents & Workpapers Table
CREATE TABLE audit_documents (
    doc_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    entity_id VARCHAR(32) NOT NULL REFERENCES legal_entities(entity_id),
    title VARCHAR(256) NOT NULL,
    fiscal_year SMALLINT NOT NULL,
    classification VARCHAR(32) NOT NULL, -- e.g. 'SOX_404', 'BSA_AML', 'KYC_RISK'
    chunk_index INT NOT NULL,
    content TEXT NOT NULL,
    content_tsv TSVECTOR GENERATED ALWAYS AS (to_tsvector('english', content)) STORED,
    embedding VECTOR(1024), -- BGE-large-en-v1.5 dimension
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
-- Hierarchical Navigable Small World (HNSW) Vector Index for sub-10ms retrieval
CREATE INDEX idx_audit_docs_hnsw ON audit_documents 
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- GIN Index for Fast Lexical BM25 Sparse Matching
CREATE INDEX idx_audit_docs_tsv ON audit_documents USING GIN (content_tsv);
-- Enforce Row-Level Security (RLS)
ALTER TABLE audit_documents ENABLE ROW LEVEL SECURITY;
-- Dynamic RLS Policy: User can only search documents matching their clearance session variable
CREATE POLICY auditor_entity_access_policy ON audit_documents
FOR SELECT
USING (
    entity_id = CURRENT_SETTING('app.current_auditor_entity', true)
    OR CURRENT_SETTING('app.is_global_compliance_officer', true) = 'true'
);

"""
scripts/hybrid_ledger_retriever.py
Executes parallel vector + sparse retrieval inside the private VPC
"""
import os
import psycopg2
from psycopg2.extras import RealDictCursor
from sentence_transformers import SentenceTransformer
Load local sovereign embedding model from internal NVMe cache (Zero WAN)
EMBED_MODEL_PATH = "/models/bge-large-en-v1.5"
model = SentenceTransformer(EMBED_MODEL_PATH, device="cuda")
def hybrid_audit_search(auditor_id: str, legal_entity: str, query: str, top_k: int = 5):
    # Generate query embedding locally in 14ms
    query_vector = model.encode(query, normalize_embeddings=True).tolist()
    vector_str = "[" + ",".join(map(str, query_vector)) + "]"
    conn = psycopg2.connect(
        host=os.getenv("LEDGER_DB_HOST", "127.0.0.1"),
        dbname="financial_ai_core",
        user="rag_worker",
        password=os.getenv("LEDGER_DB_PASSWORD"),
        port=5432
    )
    
    with conn.cursor(cursor_factory=RealDictCursor) as cur:
        # 1. Set Session Variables to enforce Row-Level Security
        cur.execute("SET LOCAL app.current_auditor_entity = %s;", (legal_entity,))
        
        # 2. Reciprocal Rank Fusion (RRF) SQL Query combining Dense Vector + Full-Text Search
        rrf_query = """
        WITH dense_search AS (
            SELECT doc_id, content, title, fiscal_year,
                   ROW_NUMBER() OVER (ORDER BY embedding <=> %s::vector) AS dense_rank
            FROM audit_documents
            ORDER BY embedding <=> %s::vector
            LIMIT 30
        ),
        sparse_search AS (
            SELECT doc_id, content, title, fiscal_year,
                   ROW_NUMBER() OVER (ORDER BY ts_rank_cd(content_tsv, plainto_tsquery('english', %s)) DESC) AS sparse_rank
            FROM audit_documents
            WHERE content_tsv @@ plainto_tsquery('english', %s)
            LIMIT 30
        )
        SELECT 
            COALESCE(d.doc_id, s.doc_id) AS doc_id,
            COALESCE(d.title, s.title) AS title,
            COALESCE(d.fiscal_year, s.fiscal_year) AS fiscal_year,
            COALESCE(d.content, s.content) AS content,
            (COALESCE(1.0 / (60 + d.dense_rank), 0.0)  0.65 +
             COALESCE(1.0 / (60 + s.sparse_rank), 0.0)  0.35) AS rrf_score
        FROM dense_search d
        FULL OUTER JOIN sparse_search s ON d.doc_id = s.doc_id
        ORDER BY rrf_score DESC
        LIMIT %s;
        """
        cur.execute(rrf_query, (vector_str, vector_str, query, query, top_k))
        results = cur.fetchall()
        
    conn.close()
    return results

"""
scripts/secure_inference_orchestrator.py
Connects hybrid retrieval to local vLLM serving with SEC 17a-4 WORM audit logging
"""
import time
import json
import hashlib
import requests
from typing import Dict, Any
VLLM_INTERNAL_URL = "http://10.0.4.15:8000/v1/chat/completions"
SYSTEM_PROMPT = """You are an air-gapped Financial Audit AI Assistant operating inside a private banking VPC.
You must adhere strictly to the provided context.
Rules:
Every numeric finding, journal entry, or balance must cite its specific Document Title and Fiscal Year.
If the context does not contain sufficient data to reconcile a balance, state: "INSUFFICIENT_AUDIT_EVIDENCE".
Never extrapolate or assume ledger balances.
Output must strictly follow the requested JSON format.
"""
def generate_reconciled_audit_response(
    auditor_id: str, 
    query: str, 
    retrieved_chunks: list
) -> Dict[str, Any]:
    
    start_time = time.perf_counter()
    # Assemble ground-truth context
    context_text = "\n\n".join([
        f"[SOURCE: {c['title']} (FY{c['fiscal_year']})]\n{c['content']}"
        for c in retrieved_chunks
    ])
    user_payload = f"Audit Investigation Query: {query}\n\nRetrieved Audit Context:\n{context_text}"
    # Request local vLLM inside VPC enclave
    req_body = {
        "model": "meta-llama/Llama-3.1-70B-Instruct",
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_payload}
        ],
        "temperature": 0.05, # Near-deterministic for audit accuracy
        "max_tokens": 1024,
        "response_format": {"type": "json_object"}
    }
    resp = requests.post(VLLM_INTERNAL_URL, json=req_body, timeout=12.0)
    resp_data = resp.json()
    model_output = resp_data["choices"][0]["message"]["content"]
    elapsed_ms = (time.perf_counter() - start_time)  1000
    # Generate SEC 17a-4 Cryptographic Audit Envelope
    chunk_hashes = [hashlib.sha256(c['content'].encode()).hexdigest() for c in retrieved_chunks]
    audit_payload = {
        "auditor_id": auditor_id,
        "timestamp_utc": int(time.time()),
        "query": query,
        "retrieved_chunk_hashes": chunk_hashes,
        "model_output": model_output,
        "latency_ms": round(elapsed_ms, 2),
        "model_id": "Llama-3.1-70B-Instruct-vLLM-Enclave"
    }
    envelope_hash = hashlib.sha256(json.dumps(audit_payload, sort_keys=True).encode()).hexdigest()
    audit_payload["immutable_envelope_sha256"] = envelope_hash
    # Commit to Write-Once-Read-Many (WORM) Storage (S3 Object Lock / MinIO)
    commit_to_worm_vault(envelope_hash, audit_payload)
    return {
        "answer": json.loads(model_output),
        "audit_hash": envelope_hash,
        "latency_ms": elapsed_ms
    }
def commit_to_worm_vault(object_key: str, data: dict):
    """Writes to S3 WORM bucket with Legal Hold / Governance Object Lock"""
    # In production: boto3 client configured with PutObject LegalHoldStatus='ON'
    pass