Mengizinkan aplikasi klien terhubung langsung ke provider upstream seperti OpenAI, Anthropic, atau open-weights cluster membuka celah keamanan dan pemborosan biaya. Ketika sebuah varian model diidentifikasi memiliki kerentanan prompt injection baru, perilaku halusinasi kritis, atau harga inferensi melonjak tajam, tim keamanan harus mampu memblokir model tersebut secara instan. Menunggu siklus redeploy aplikasi bukan solusi yang layak pada sistem produksi.

Solusinya adalah menempatkan reverse proxy cerdas sebagai LLM gateway untuk mengontrol seluruh lalu lintas egress AI. Gateway bertindak sebagai sentral penegakan kebijakan (Policy Enforcement Point) yang memvalidasi setiap payload permintaan sebelum mencapai upstream API.

Arsitektur Egress Gateway LLM dan Dynamic Policy Store

Arsitektur ini memisahkan kontrol model dari kode aplikasi backend. Permintaan HTTP dari aplikasi diarahkan ke gateway internal, gateway membaca status kebijakan dari cache terdistribusi (seperti Redis atau in-memory TTL store), lalu memutuskan apakah request diteruskan atau ditolak.

Komponen utama arsitektur ini mencakup:

  • Inspection Middleware: Membaca stream body JSON, mengekstrak field model, lalu merekonstruksi stream body agar upstream dapat membaca data tanpa error.
  • Policy Engine: Melakukan lookup O(1) ke policy store untuk mengecek apakah ID model masuk daftar blokir (misal: gpt-4-base, claude-v1-unrestricted) atau memerlukan rerouting otomatis.
  • Dynamic Policy Store: Redis dengan pub/sub atau short-lived cache lokal (TTL 30 detik) yang memungkinkan penambahan model terlarang tanpa restart gateway.
  • Upstream Proxy: Meneruskan request valid via koneksi HTTP/2 atau HTTP/1.1 terkelola ke endpoint upstream.

Desain Fail-Secure dan Format Error RFC 7807

Gateway keamanan harus mengadopsi prinsip fail-secure (fail-closed). Jika policy store mengalami timeout, network partition, atau payload JSON rusak sehingga field model tidak dapat diuraikan, gateway wajib menolak permintaan secara default daripada meloloskannya ke upstream tanpa verifikasi.

Saat pemblokiran terjadi, client API harus menerima konteks masalah yang jelas tanpa mengekspos rahasia infrastruktur internal. Standar RFC 7807 (Problem Details for HTTP APIs) adalah format respons yang tepat untuk skenario ini dengan header Content-Type: application/problem+json.

HTTP/1.1 403 Forbidden
Content-Type: application/problem+json

{
  "type": "https://api.internal/errors/forbidden-model",
  "title": "Model Usage Forbidden",
  "status": 403,
  "detail": "Model 'text-davinci-003' is blocked under Security Policy SEC-LLM-04.",
  "instance": "/v1/chat/completions",
  "invalid_params": [
    {
      "name": "model",
      "reason": "Model deprecated due to safety non-compliance"
    }
  ]
}

Implementasi Reverse Proxy dan Middleware di Go

Berikut adalah implementasi reverse proxy minimalis menggunakan paket standar net/http dan net/http/httputil di Go. Kode ini menginspeksi body, memvalidasi model terhadap policy store, menangani safe model fallback, dan mencatat audit log terstruktur.

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/http/httputil"
	"net/url"
	"os"
	"sync"
)

// RFC 7807 Problem Details representation
type ProblemDetails struct {
	Type     string `json:"type"`
	Title    string `json:"title"`
	Status   int    `json:"status"`
	Detail   string `json:"detail"`
	Instance string `json:"instance"`
}

type PolicyStore interface {
	IsBlocked(model string) (bool, error)
	GetFallback(model string) (string, bool)
}

type MemoryPolicyStore struct {
	mu        sync.RWMutex
	blocklist map[string]bool
	fallbacks map[string]string
}

func (m *MemoryPolicyStore) IsBlocked(model string) (bool, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.blocklist[model], nil
}

func (m *MemoryPolicyStore) GetFallback(model string) (string, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	target, exists := m.fallbacks[model]
	return target, exists
}

type Gateway struct {
	policy PolicyStore
	proxy  *httputil.ReverseProxy
	logger *slog.Logger
}

func NewGateway(target *url.URL, policy PolicyStore) *Gateway {
	logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
	proxy := httputil.NewSingleHostReverseProxy(target)
	return &Gateway{
		policy: policy,
		proxy:  proxy,
		logger: logger,
	}
}

func (g *Gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	// Hanya inspeksi rute completions/chat
	if r.Method != http.MethodPost || r.URL.Path != "/v1/chat/completions" {
		g.proxy.ServeHTTP(w, r)
		return
	}

	bodyBytes, err := io.ReadAll(r.Body)
	if err != nil {
		g.writeProblem(w, r.URL.Path, http.StatusBadRequest, "Malformed Request", "Unable to read request payload.")
		return
	}
	r.Body.Close()

	var payload map[string]interface{}
	if err := json.Unmarshal(bodyBytes, &payload); err != nil {
		// Fail-secure: jika payload tidak valid, tolak
		g.writeProblem(w, r.URL.Path, http.StatusBadRequest, "Invalid JSON", "Body must be valid JSON.")
		return
	}

	modelRaw, ok := payload["model"]
	modelName, isString := modelRaw.(string)
	if !ok || !isString || modelName == "" {
		// Fail-secure: model tidak ada di payload
		g.writeProblem(w, r.URL.Path, http.StatusUnprocessableEntity, "Missing Model", "The 'model' field is mandatory.")
		return
	}

	// 1. Cek Dynamic Policy
	blocked, err := g.policy.IsBlocked(modelName)
	if err != nil {
		// Fail-secure: jika pengecekan policy error, tolak
		g.logger.Error("policy check failed", "error", err, "model", modelName)
		g.writeProblem(w, r.URL.Path, http.StatusInternalServerError, "Policy Error", "Unable to verify egress security policy.")
		return
	}

	if blocked {
		// Cek fallback terverifikasi
		fallbackModel, hasFallback := g.policy.GetFallback(modelName)
		if hasFallback {
			payload["model"] = fallbackModel
			modifiedBody, _ := json.Marshal(payload)
			bodyBytes = modifiedBody
			r.ContentLength = int64(len(modifiedBody))
			r.Header.Set("Content-Length", fmt.Sprintf("%d", len(modifiedBody)))

			g.logger.Warn("egress policy reroute",
				"action", "REROUTED",
				"client_ip", r.RemoteAddr,
				"original_model", modelName,
				"target_model", fallbackModel,
			)
		} else {
			g.logger.Warn("egress policy violation",
				"action", "BLOCKED",
				"client_ip", r.RemoteAddr,
				"model", modelName,
			)
			g.writeProblem(w, r.URL.Path, http.StatusForbidden, "Model Forbidden",
				fmt.Sprintf("Model '%s' is prohibited under current egress security policy.", modelName))
			return
		}
	} else {
		g.logger.Info("egress policy allowed", "action", "ALLOWED", "model", modelName)
	}

	// Rekonstruksi r.Body untuk diteruskan ke upstream
	r.Body = io.NopCloser(bytes.NewReader(bodyBytes))
	g.proxy.ServeHTTP(w, r)
}

func (g *Gateway) writeProblem(w http.ResponseWriter, instance string, status int, title, detail string) {
	w.Header().Set("Content-Type", "application/problem+json")
	w.WriteHeader(status)
	prob := ProblemDetails{
		Type:     "https://api.internal/errors/llm-egress-policy",
		Title:    title,
		Status:   status,
		Detail:   detail,
		Instance: instance,
	}
	_ = json.NewEncoder(w).Encode(prob)
}

Audit Logging Terstruktur dan Model Fallback

Salah satu syarat utama kepatuhan (SOC 2, ISO 27001) dalam penggunaan generative AI adalah keterlacakan (traceability). Gateway harus mencatat setiap keputusan egress yang diambil. Pada implementasi di atas, slog digunakan untuk menghasilkan log JSON yang dapat langsung diparsing oleh log aggregator (seperti Grafana Loki atau Datadog).

Pola Model Fallback memberikan fleksibilitas operasional: daripada menggagalkan permintaan yang menyebabkan degradasi sistem pada klien, model berisiko tinggi atau mahal secara transparan diganti dengan model aman yang setara (misalnya mengarahkan gpt-4-legacy ke gpt-4o-mini).

Pengujian Gateway: Runnable Self-Check

Berikut adalah pengujian fungsional menggunakan httptest untuk memvalidasi bahwa gateway bekerja sesuai spesifikasi: meloloskan model valid, meredireksi model fallback, dan menolak model terlarang dengan format RFC 7807.

package main

import (
	"bytes"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"net/url"
	"testing"
)

func TestLLMGatewayPolicy(t *testing.T) {
	// Mock Upstream server
	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var req map[string]interface{}
		_ = json.NewDecoder(r.Body).Decode(&req)
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusOK)
		_ = json.NewEncoder(w).Encode(map[string]string{"status": "ok", "model_served": req["model"].(string)})
	}))
	defer upstream.Close()

	upstreamURL, _ := url.Parse(upstream.URL)
	store := &MemoryPolicyStore{
		blocklist: map[string]bool{
			"dangerous-model-v1": true,
			"deprecated-fast":    true,
		},
		fallbacks: map[string]string{
			"deprecated-fast": "safe-fast-v2",
		},
	}

	gw := NewGateway(upstreamURL, store)

	// Kasus 1: Model diblokir tanpa fallback -> 403 Forbidden
	req1 := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewBufferString(`{"model":"dangerous-model-v1"}`))
	rec1 := httptest.NewRecorder()
	gw.ServeHTTP(rec1, req1)

	if rec1.Code != http.StatusForbidden {
		t.Fatalf("Expected 403 Forbidden, got %d", rec1.Code)
	}
	if rec1.Header().Get("Content-Type") != "application/problem+json" {
		t.Errorf("Expected RFC 7807 content type, got %s", rec1.Header().Get("Content-Type"))
	}

	// Kasus 2: Model dialihkan via fallback -> 200 OK dengan model baru
	req2 := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewBufferString(`{"model":"deprecated-fast"}`))
	rec2 := httptest.NewRecorder()
	gw.ServeHTTP(rec2, req2)

	if rec2.Code != http.StatusOK {
		t.Fatalf("Expected 200 OK, got %d", rec2.Code)
	}
	var resBody map[string]string
	_ = json.NewDecoder(rec2.Body).Decode(&resBody)
	if resBody["model_served"] != "safe-fast-v2" {
		t.Errorf("Expected safe-fast-v2, got %s", resBody["model_served"])
	}

	// Kasus 3: Model diizinkan -> 200 OK
	req3 := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewBufferString(`{"model":"standard-gpt"}`))
	rec3 := httptest.NewRecorder()
	gw.ServeHTTP(rec3, req3)

	if rec3.Code != http.StatusOK {
		t.Fatalf("Expected 200 OK, got %d", rec3.Code)
	}
}

Pertimbangan Operasional dan Trade-Off

  • Latency Overhead: Membaca dan mengurai body JSON memperkenalkan latensi marginal (sub-milidetik pada payload tipikal). Untuk payload raksasa dengan riwayat chat panjang, gunakan streaming parser (seperti json.Decoder dengan token traversal) alih-alih json.Unmarshal penuh pada seluruh buffer.
  • Kesiapan Streaming (SSE): Blokir model dievaluasi pada saat request masuk, sehingga koneksi Server-Sent Events (SSE) di sisi response upstream tidak terganggu oleh logic interceptor request ini.
  • Keandalan Policy Cache: Di production, gunakan Redis dengan local read-through cache (misal: Ristretto atau sync.Map dengan TTL) guna mencegah lonjakan latensi jika cluster Redis mengalami lonjakan beban.