Gejala: Data Profil Pengguna Bocor Lintas Sesi

Saat menerapkan layer proteksi dan personalisasi (Guardian LLM) di depan upstream API LLM, proxy backend bertugas memvalidasi prompt dan menginjeksikan profil kepatuhan pengguna. Masalah muncul ketika User B menerima respons yang memuat entitas personal (seperti nama lengkap, ID rekening, atau riwayat transaksi) milik User A.

Kegagalan ini terdeteksi saat beban konkurensi meningkat. Data boundary antar-sesi runtuh di layer proxy, menyebabkan model Guardian mengevaluasi prompt menggunakan variabel konteks dari request yang mendahuluinya.

Analisis Log Transaksi

Pemeriksaan log backend menunjukkan urutan request yang saling tumpang tindih dalam waktu milidetik:

2025-02-15T10:14:02.102Z [REQ-001] INFO  auth: User authenticated: user_id=usr_alice_99
2025-02-15T10:14:02.105Z [REQ-001] DEBUG context: Injected profile context: tier=enterprise, name=Alice
2025-02-15T10:14:02.108Z [REQ-002] INFO  auth: User authenticated: user_id=usr_bob_42
2025-02-15T10:14:02.112Z [REQ-001] INFO  llm_call: Dispatching prompt with 1420 tokens
2025-02-15T10:14:02.115Z [REQ-002] DEBUG context: Injected profile context: tier=free, name=Bob
2025-02-15T10:14:02.450Z [REQ-001] INFO  llm_resp: Generated completion for usr_alice_99
2025-02-15T10:14:02.452Z [REQ-002] INFO  llm_call: Dispatching prompt with 1420 tokens
2025-02-15T10:14:02.810Z [REQ-002] WARN  llm_resp: Response contains leaked token match: 'Alice'

Meskipun autentikasi REQ-002 mencatat identitas Bob, muatan token yang dikirim ke upstream LLM mempertahankan data dari REQ-001. Penelusuran trace payload mengonfirmasi bahwa system instruction milik Bob telah terkontaminasi metadata milik Alice.

Root Cause: Shared Mutable State pada Context-Builder

Akar masalah berasal dari implementasi singleton class pada service context-builder. Service tersebut menggunakan list atau dictionary di level instance untuk menampung history dan profil tanpa isolasi per-request:

# KODE BERMASALAH: State dibagikan antar coroutine / request
class GuardianContextBuilder:
    def __init__(self):
        self.context_store = {}  # In-memory dictionary bersama

    def set_user_context(self, user_id: str, profile_data: dict):
        self.context_store["current_user"] = user_id
        self.context_store["profile"] = profile_data

    def build_prompt(self, user_prompt: str) -> str:
        # Race condition: Timpaan nilai dari request lain sebelum prompt sempat dibaca
        profile = self.context_store.get("profile", {})
        return f"System: User context {profile}\nPrompt: {user_prompt}"

# Instance singleton di level modul
builder_instance = GuardianContextBuilder()

Pada arsitektur asinkron (seperti FastAPI/Uvicorn atau Node.js event loop), beberapa request diproses secara konkuren dalam satu proses. Saat REQ-001 mengeksekusi operasi async I/O (misal fetch database), eksekusi berpindah ke REQ-002. Mutasi pada self.context_store menimpa state milik REQ-001, atau sebaliknya state REQ-001 tertinggal dan terbaca oleh REQ-002 karena ketiadaan namespace spesifik.

Langkah Isolasi Bug dengan Request-Scoped Context

Untuk memutus shared state drift, implementasi arsitektur context-builder harus memenuhi prinsip immutability dan explicit isolation:

  1. Hapus module-level state: Hindari atribut mutable pada class singleton untuk menampung data transient.
  2. Gunakan ContextVar: Manfaatkan primitive bawaan runtime untuk mengisolasi context per execution-chain (coroutine).
  3. Explicit Passing: Lewatkan konteks sebagai parameter murni fungsi (pure function payload) alih-alih menyimpannya di instance object.

Implementasi Perbaikan (Python / FastAPI)

Solusi berikut menggunakan dataclasses murni untuk immutability dan modul contextvars untuk menjamin data boundary di setiap asynchronous task:

from contextvars import ContextVar
from dataclasses import dataclass
from typing import Optional
from fastapi import FastAPI, Depends, Request

@dataclass(frozen=True)
class UserContext:
    user_id: str
    tenant_id: str
    compliance_rules: list[str]

# Inisialisasi request-scoped context variable
_request_context: ContextVar[Optional[UserContext]] = ContextVar("user_context", default=None)

class GuardianContextManager:
    @staticmethod
    def set_context(context: UserContext):
        _request_context.set(context)

    @staticmethod
    def get_context() -> UserContext:
        ctx = _request_context.get()
        if ctx is None:
            raise RuntimeError("Data boundary error: Uninitialized request context")
        return ctx

class SafeContextBuilder:
    @staticmethod
    def construct_system_prompt(raw_prompt: str, context: UserContext) -> str:
        # Menghasilkan prompt terisolasi tanpa efek samping ke state global
        rules_joined = "; ".join(context.compliance_rules)
        return (
            f"[IDENTITY: {context.user_id}]\n"
            f"[RULES: {rules_joined}]\n"
            f"User Input: {raw_prompt}"
        )

app = FastAPI()

# Dependency injection menjamin context di-resolve per-request
async def get_current_user_context(request: Request) -> UserContext:
    user_id = request.headers.get("X-User-ID", "anonymous")
    # Ambil profile dari session storage terisolasi (contoh: DB atau JWT claims)
    return UserContext(
        user_id=user_id,
        tenant_id=request.headers.get("X-Tenant-ID", "default"),
        compliance_rules=["BLOCK_PII_LEAK", "ENFORCE_TONE_SAFETY"]
    )

@app.post("/v1/guardian/proxy")
async def handle_prompt(
    prompt: str,
    context: UserContext = Depends(get_current_user_context)
):
    GuardianContextManager.set_context(context)
    
    # Context di-pass secara eksplisit, menghilangkan shared mutable drift
    scoped_context = GuardianContextManager.get_context()
    sanitized_payload = SafeContextBuilder.construct_system_prompt(prompt, scoped_context)
    
    # Eksekusi call ke upstream LLM terbebas dari kebocoran data
    return {"status": "success", "composed_prompt": sanitized_payload}

Verifikasi dan Pencegahan Regresi

Untuk memastikan data boundary bekerja di bawah beban tinggi, jalankan concurrency testing menggunakan script evaluasi race condition:

import asyncio
import httpx

async def send_check(user_id: str):
    async with httpx.AsyncClient(base_url="http://127.0.0.1:8000") as client:
        resp = await client.post(
            "/v1/guardian/proxy",
            params={"prompt": f"Request from {user_id}"},
            headers={"X-User-ID": user_id}
        )
        data = resp.json()
        assert f"[IDENTITY: {user_id}]" in data["composed_prompt"], f"Leak detected for {user_id}"

async def test_concurrency():
    tasks = [send_check(f"usr_{i}") for i in range(100)]
    await asyncio.gather(*tasks)
    print("All concurrent sessions passed data boundary assertion.")

if __name__ == "__main__":
    asyncio.run(test_concurrency())

Dengan mengalihkan arsitektur dari shared in-memory state ke request-scoped context, parameter isolasi antar-sesi terjaga penuh, mencegah celah privasi dan anomali inferensi pada proxy Guardian LLM.