i run three different LLMs locally. Ollama for quick stuff, NanoGPT for longer tasks, and a small fine-tuned model for code completion. but here's the thing nobody talks about: just because the model runs locally doesn't mean your prompts are safe.
every one of those tools has its own API endpoint, its own logging behavior, and its own idea of what "ephemeral" means. ollama keeps a request log by default. your web UI stores conversation history in sqlite. your reverse proxy writes access logs with full request bodies if you configured it wrong.
i got tired of wondering where my prompts were ending up. so i built a middleware layer that sits between my tools and my models. every prompt goes through it. every response comes back through it. everything gets encrypted and stored locally, and i can search it whenever i want.
here's the full build.
the architecture
the idea is simple. instead of talking directly to ollama's API at localhost:11434, my tools talk to my audit logger at localhost:9400. the logger:
- receives the prompt and metadata
- encrypts it with a local key
- stores it in a sqlite database
- forwards the request to the actual LLM endpoint
- encrypts and stores the response
- returns the response to the client
the client never knows the difference. it just sees a normal openai-compatible API. the logger is invisible.
┌──────────────┐ ┌──────────────────┐ ┌──────────────┐
│ your tools │────▶│ audit logger │────▶│ ollama │
│ (openwebui, │ │ (localhost:9400) │ │ (11434) │
│ scripts) │◀────│ encrypts+stores │◀────│ │
└──────────────┘ └──────────────────┘ └──────────────┘
the project structure
ai-audit-logger/
├── main.py # fastapi app
├── crypto.py # encryption helpers
├── storage.py # database layer
├── proxy.py # forwarding logic
├── config.py # settings
├── models.py # pydantic schemas
├── search.py # search endpoint
├── requirements.txt
└── docker-compose.yml
config.py — keeping secrets out of code
i use environment variables for everything sensitive. the encryption key gets derived from a passphrase you set once and never type again.
# config.py
import os
from pathlib import Path
LLM_BASE_URL = os.getenv("LLM_BASE_URL", "http://localhost:11434")
LISTEN_HOST = os.getenv("LISTEN_HOST", "127.0.0.1")
LISTEN_PORT = int(os.getenv("LISTEN_PORT", "9400"))
DB_PATH = Path(os.getenv("DB_PATH", "./audit.db"))
PASSPHRASE = os.getenv("AUDIT_PASSPHRASE", "change-me-before-deploying")
LOG_DIR = Path(os.getenv("LOG_DIR", "./logs"))
LOG_DIR.mkdir(exist_ok=True)
the passphrase is the only secret. you set it as an env var or put it in a .env file that never leaves your machine. i'll show how the key derivation works next.
crypto.py — encryption at rest
i use cryptography's Fernet for symmetric encryption. it's fast, audited, and good enough for local storage. the key gets derived from your passphrase using PBKDF2 with 600k iterations.
# crypto.py
import os
import base64
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
_SALT_FILE = ".audit_salt"
def _get_or_create_salt() -> bytes:
if os.path.exists(_SALT_FILE):
return open(_SALT_FILE, "rb").read()
salt = os.urandom(16)
open(_SALT_FILE, "wb").write(salt)
return salt
def derive_key(passphrase: str) -> bytes:
salt = _get_or_create_salt()
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=salt,
iterations=600_000,
)
return base64.urlsafe_b64encode(kdf.derive(passphrase.encode()))
def encrypt(plaintext: str, key: bytes) -> bytes:
return Fernet(key).encrypt(plaintext.encode())
def decrypt(ciphertext: bytes, key: bytes) -> str:
return Fernet(key).decrypt(ciphertext).decode()
the salt gets generated once and stored in a file. if you lose the salt, you lose your data. that's the tradeoff. i back up the salt file alongside my database.
models.py — what we track
every logged request includes metadata that most people never think about. which tool sent the prompt, what model was targeted, how long it took, and whether the response was streamed.
# models.py
from pydantic import BaseModel
from typing import Optional
class ChatMessage(BaseModel):
role: str
content: str
class ChatCompletionRequest(BaseModel):
model: str
messages: list[ChatMessage]
temperature: float = 0.7
max_tokens: Optional[int] = None
stream: bool = False
user: Optional[str] = None # which tool sent this
class AuditEntry(BaseModel):
id: Optional[int] = None
timestamp: float
model: str
user_agent: str
source_ip: str
prompt_encrypted: bytes
response_encrypted: bytes
duration_ms: float
tokens_prompt: int
tokens_response: int
stream: bool
i log the user agent and source IP so i can tell which tool sent the prompt. if open webui sends something vs. a python script, i want to know.
storage.py — sqlite with encrypted fields
sqlite because it's zero-config and lives in a single file i can back up with a simple cp.
# storage.py
import sqlite3
import time
from pathlib import Path
from config import DB_PATH
def get_db() -> sqlite3.Connection:
conn = sqlite3.connect(str(DB_PATH))
conn.execute("""
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp REAL NOT NULL,
model TEXT NOT NULL,
user_agent TEXT,
source_ip TEXT,
prompt_encrypted BLOB NOT NULL,
response_encrypted BLOB NOT NULL,
duration_ms REAL,
tokens_prompt INTEGER DEFAULT 0,
tokens_response INTEGER DEFAULT 0,
stream BOOLEAN DEFAULT 0
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_timestamp ON audit_log(timestamp)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_model ON audit_log(model)
""")
conn.commit()
return conn
def store_entry(conn, entry):
conn.execute("""
INSERT INTO audit_log
(timestamp, model, user_agent, source_ip, prompt_encrypted,
response_encrypted, duration_ms, tokens_prompt, tokens_response, stream)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
entry.timestamp, entry.model, entry.user_agent, entry.source_ip,
entry.prompt_encrypted, entry.response_encrypted,
entry.duration_ms, entry.tokens_prompt, entry.tokens_response,
entry.stream
))
conn.commit()
the prompt and response are stored as encrypted blobs. anyone who opens the sqlite file just sees random bytes. they need the passphrase to read anything.
proxy.py — the forwarding logic
this is the core. it receives the openai-compatible request, logs it, forwards to the real endpoint, logs the response, and sends it back.
# proxy.py
import time
import httpx
from models import ChatCompletionRequest
from crypto import encrypt
from storage import store_entry
from config import LLM_BASE_URL, PASSPHRASE
from config import LOG_DIR
from pathlib import Path
import json
_key = None
def get_key():
global _key
if _key is None:
from crypto import derive_key
_key = derive_key(PASSPHRASE)
return _key
async def forward_request(
request: ChatCompletionRequest,
user_agent: str,
source_ip: str,
) -> dict:
key = get_key()
start = time.time()
# encrypt the prompt before forwarding
prompt_text = json.dumps(
[m.model_dump() for m in request.messages], ensure_ascii=False
)
prompt_encrypted = encrypt(prompt_text, key)
# forward to the real LLM
payload = {
"model": request.model,
"messages": [m.model_dump() for m in request.messages],
"temperature": request.temperature,
"stream": False, # handle streaming separately
}
if request.max_tokens:
payload["max_tokens"] = request.max_tokens
async with httpx.AsyncClient(timeout=300) as client:
resp = await client.post(
f"{LLM_BASE_URL}/v1/chat/completions",
json=payload,
)
result = resp.json()
duration_ms = (time.time() - start) * 1000
# encrypt the response
response_text = json.dumps(result, ensure_ascii=False)
response_encrypted = encrypt(response_text, key)
# count tokens (rough estimate from response)
usage = result.get("usage", {})
tokens_prompt = usage.get("prompt_tokens", 0)
tokens_response = usage.get("completion_tokens", 0)
entry = {
"timestamp": time.time(),
"model": request.model,
"user_agent": user_agent,
"source_ip": source_ip,
"prompt_encrypted": prompt_encrypted,
"response_encrypted": response_encrypted,
"duration_ms": duration_ms,
"tokens_prompt": tokens_prompt,
"tokens_response": tokens_response,
"stream": False,
}
# store in database
from storage import get_db
conn = get_db()
store_entry(conn, entry)
# also write a human-readable log line (no content, just metadata)
log_line = (
f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] "
f"model={request.model} "
f"duration={duration_ms:.0f}ms "
f"tokens={tokens_prompt}+{tokens_response} "
f"source={user_agent}\n"
)
log_file = LOG_DIR / f"{time.strftime('%Y-%m-%d')}.log"
log_file.open("a").write(log_line)
return result
notice the human-readable log line at the bottom. it stores metadata only, no prompt content. i can quickly check "what models ran today" without decrypting anything.
main.py — the fastapi app
# main.py
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from models import ChatCompletionRequest
from proxy import forward_request
from storage import get_db
from crypto import decrypt, get_key
from config import LISTEN_HOST, LISTEN_PORT, PASSPHRASE
from crypto import derive_key
import json
import time
app = FastAPI(title="AI Audit Logger")
@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
body = await request.json()
req = ChatCompletionRequest(**body)
user_agent = request.headers.get("user-agent", "unknown")
source_ip = request.client.host if request.client else "unknown"
result = await forward_request(req, user_agent, source_ip)
return result
@app.get("/v1/audit/search")
def search_prompts(
q: str = "",
model: str = "",
limit: int = 50,
offset: int = 0,
):
key = derive_key(PASSPHRASE)
conn = get_db()
query = "SELECT * FROM audit_log WHERE 1=1"
params = []
if model:
query += " AND model = ?"
params.append(model)
query += " ORDER BY timestamp DESC LIMIT ? OFFSET ?"
params.extend([limit, offset])
rows = conn.execute(query, params).fetchall()
results = []
for row in rows:
prompt_decrypted = decrypt(row[5], key)
response_decrypted = decrypt(row[6], key)
# if search query provided, filter
if q and q.lower() not in prompt_decrypted.lower():
continue
results.append({
"id": row[0],
"timestamp": row[1],
"model": row[2],
"user_agent": row[3],
"duration_ms": row[7],
"prompt": prompt_decrypted,
"response": response_decrypted,
})
return {"count": len(results), "results": results}
@app.get("/v1/audit/stats")
def audit_stats():
conn = get_db()
total = conn.execute("SELECT COUNT(*) FROM audit_log").fetchone()[0]
models = conn.execute(
"SELECT model, COUNT(*) as cnt FROM audit_log GROUP BY model ORDER BY cnt DESC"
).fetchall()
today = conn.execute(
"SELECT COUNT(*) FROM audit_log WHERE timestamp > ?",
(time.time() - 86400,)
).fetchone()[0]
return {
"total_logged": total,
"today": today,
"models": [{"model": m[0], "count": m[1]} for m in models],
}
@app.get("/health")
def health():
return {"status": "ok"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host=LISTEN_HOST, port=LISTEN_PORT)
the search endpoint decrypts entries on the fly. it's not fast for large datasets, but for personal use with a few thousand entries, it's instant. the stats endpoint gives you a quick overview without touching encrypted data.
docker-compose.yml — running it
# docker-compose.yml
version: "3.8"
services:
audit-logger:
build: .
ports:
- "127.0.0.1:9400:9400"
environment:
- LLM_BASE_URL=http://host.docker.internal:11434
- AUDIT_PASSPHRASE=${AUDIT_PASSPHRASE}
- DB_PATH=/data/audit.db
volumes:
- ./data:/data
- ./logs:/app/logs
restart: unless-stopped
bind to localhost only. never expose this to a network. the audit logger has your decrypted prompts in memory and the encryption key loaded. it should never be reachable from outside your machine.
the requirements.txt
fastapi==0.115.0
uvicorn==0.30.0
httpx==0.27.0
cryptography==43.0.0
pydantic==2.9.0
five dependencies. that's it. no heavy ML libraries, no database drivers, no ORM. sqlite is built into python.
how to use it
start the logger first, then point your tools at it:
# start the audit logger
cd ai-audit-logger
export AUDIT_PASSPHRASE="your-secret-passphrase"
docker compose up -d
# point ollama-compatible tools at it
export OLLAMA_BASE_URL=http://localhost:9400/v1
now when you use open webui, any python script, or any tool that talks to an openai-compatible API, everything goes through the logger first.
to search your prompts:
curl "http://localhost:9400/v1/audit/search?q=medical&limit=10"
to see stats:
curl "http://localhost:9400/v1/audit/stats"
what i learned building this
the biggest surprise was how many tools silently log your prompts. ollama writes to its own log files. open webui stores everything in its sqlite database. even curl writes to your shell history. the audit logger doesn't solve the shell history problem, but it gives you one place to look instead of five.
the encryption overhead is negligible. i benchmarked 10k encrypt/decrypt cycles and it added maybe 2ms per request. for a local tool that's nothing.
the real value is the search. when i'm debugging a model's behavior on a specific prompt from three weeks ago, i can just query it. no more scrolling through web UI history or grepping log files.
if you want to see more privacy-focused tools and self-hosted setups, i keep a running list at ai-privacy-tools.vercel.app with honest reviews of what actually works. i update it whenever i find something worth recommending or something that turned out to be garbage.
the full code for this project is meant to be copied, modified, and run on your own hardware. the encryption is real, the storage is local, and nothing leaves your machine. that's the whole point.
Top comments (0)