Record client IP and user agent in the audit log with trusted-proxy unwrap

.env.example

@@ -26,6 +26,8 @@ CODEBUDDY2API_ADMIN_CSRF=true
 # Extra trusted management origins behind reverse proxies (comma separated; bare domains use HTTPS).
 # Unset leaves the list configurable in the WebUI; an explicit value locks it.
 # CODEBUDDY2API_ADMIN_ORIGINS=https://chat.example.com
+# 可信代理 IP/CIDR（逗号分隔）；命中时审计改记 X-Forwarded-For/X-Real-IP 的真实客户端地址
+# CODEBUDDY2API_TRUSTED_PROXIES=127.0.0.1,10.0.0.0/8
 
 # Unset leaves tool metadata configurable in the WebUI; an explicit Boolean locks it.
 # CODEBUDDY2API_KEEP_TOOL_METADATA=true

app/admin_api.py

@@ -486,7 +486,7 @@ async def logs_get(request):
             raise ValueError("limit 必须为整数") from None
         if not 1 <= limit <= 200:
             raise ValueError("limit 必须为 1 至 200")
-        filters = {key: params[key] for key in ("model", "credential", "profile", "status", "search") if key in params}
+        filters = {key: params[key] for key in ("model", "credential", "profile", "status", "search", "client_ip") if key in params}
         if any(len(value) > 256 for value in filters.values()) or len(params.get("cursor", "")) > 2048:
             raise ValueError("日志筛选参数过长")
         result = await run_in_threadpool(audit.list_records, kind, limit, params.get("cursor"), **filters)

app/audit_store.py

@@ -15,7 +15,7 @@
 import uuid
 from typing import Any
 
-SCHEMA_VERSION = 2
+SCHEMA_VERSION = 3
 _CLEANUP_BATCH = 128
 _CLEANUP_SECONDS = 0.05
 METRICS = ("input_tokens", "output_tokens", "cache_read_tokens", "cache_creation_tokens",
@@ -33,6 +33,16 @@ def safe_label(value: Any, limit: int = 160) -> str | None:
     return value
 
 
+
+def safe_agent(value: Any) -> str | None:
+    """Single-line printable User-Agent snippet; never header-shaped secrets."""
+    if not isinstance(value, str):
+        return None
+    cleaned = " ".join(value.split())[:200]
+    if not cleaned or not all(0x20 <= ord(char) <= 0x7E for char in cleaned):
+        return None
+    return None if _SECRET.search(cleaned) else cleaned
+
 def number(value: Any) -> int | float | None:
     if isinstance(value, bool) or not isinstance(value, (int, float)):
         return None
@@ -119,7 +129,7 @@ def __init__(self, path, max_bytes=256 * 1024 * 1024, retention_days=30,
             self._db.execute("BEGIN IMMEDIATE")
             version = self._db.execute("PRAGMA user_version").fetchone()[0]
             tables = self._db.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
-            if version not in (0, 1, SCHEMA_VERSION) or (version == 0 and tables):
+            if version not in (0, 1, 2, SCHEMA_VERSION) or (version == 0 and tables):
                 raise ValueError("unsupported audit schema")
             self._db.execute("CREATE TABLE IF NOT EXISTS state (id INTEGER PRIMARY KEY CHECK(id=1), epoch INTEGER NOT NULL, detail_generation INTEGER NOT NULL, cleared_at REAL NOT NULL)")
             self._db.execute("INSERT OR IGNORE INTO state VALUES(1,0,0,0)")
@@ -133,6 +143,7 @@ def __init__(self, path, max_bytes=256 * 1024 * 1024, retention_days=30,
             for table in ("stats_hourly", "stats_daily", "stats_totals"):
                 self._db.execute(f"CREATE TABLE IF NOT EXISTS {table} (bucket INTEGER NOT NULL, dimension TEXT NOT NULL, dimension_key TEXT NOT NULL, payload TEXT NOT NULL, PRIMARY KEY(bucket,dimension,dimension_key))")
             self._migrate_ingest_time()
+            self._migrate_client_ip()
             self._db.execute(f"PRAGMA user_version={SCHEMA_VERSION}")
             self._epoch = self._db.execute("SELECT epoch FROM state").fetchone()[0]
             self._db.execute("COMMIT")
@@ -150,6 +161,15 @@ def _migrate_ingest_time(self):
                 (SELECT started_at FROM events WHERE events.id=ingest.id), ?)""", (time.time(),))
         self._db.execute("CREATE INDEX IF NOT EXISTS ingest_time ON ingest(created_at)")
 
+    def _migrate_client_ip(self):
+        # Additive v3 migration: record the audit-visible client address per request.
+        columns = [row[1] for row in self._db.execute("PRAGMA table_info(requests)").fetchall()]
+        if "client_ip" not in columns:
+            self._db.execute("ALTER TABLE requests ADD COLUMN client_ip TEXT")
+        # Exact-match client filters must not scan the retained time range.
+        self._db.execute(
+            "CREATE INDEX IF NOT EXISTS requests_client ON requests(client_ip, started_at)")
+
     def _init_accounting(self):
         # Additive v1 migration: scan legacy details only once, under the same
         # write transaction that installs triggers. Reopening never recounts.
@@ -265,6 +285,8 @@ def _aggregate(self, record):
     def _sanitize_record(self, source):
         result = {key: safe_label(source.get(key)) for key in
                   ("upstream_model", "profile", "credential", "protocol", "error_code", "usage_source")}
+        result["client_ip"] = safe_label(source.get("client_ip"))
+        result["user_agent"] = safe_agent(source.get("user_agent"))
         result["stream_mode"] = (source.get("stream_mode")
                                  if source.get("stream_mode") in ("compatible", "realtime") else None)
         projection_mode = source.get("responses_projection_mode")
@@ -318,10 +340,12 @@ def commit():
                     (generation is None or generation == state["detail_generation"]))
             if keep:
                 payload = json.dumps(data, separators=(",", ":"))
-                self._db.execute("INSERT INTO requests VALUES(?,?,?,?,?,?,?,?,?)", (
-                    data["id"], data["started_at"], data["model"], data["profile"], data["credential"],
-                    data["outcome"], data["status_code"], payload,
-                    len(payload.encode()) + sum(len(json.dumps(a).encode()) + 32 for a in data["attempts"]) + 256))
+                self._db.execute(
+                    "INSERT INTO requests(id,started_at,model,profile,credential,outcome,status_code,payload,logical_bytes,client_ip) VALUES(?,?,?,?,?,?,?,?,?,?)",
+                    (data["id"], data["started_at"], data["model"], data["profile"], data["credential"],
+                     data["outcome"], data["status_code"], payload,
+                     len(payload.encode()) + sum(len(json.dumps(a).encode()) + 32 for a in data["attempts"]) + 256,
+                     data["client_ip"]))
                 for i, attempt in enumerate(data["attempts"]):
                     self._db.execute("INSERT INTO attempts VALUES(?,?,?)", (data["id"], i, json.dumps(attempt)))
             self._prune()
@@ -435,7 +459,7 @@ def list_records(self, kind="request", limit=50, cursor=None, **filters):
             clauses.append("(started_at<? OR (started_at=? AND id<?))")
             params.extend((stamp, stamp, record_id))
         if kind == "request":
-            for key in ("model", "profile", "credential", "outcome", "status_code"):
+            for key in ("model", "profile", "credential", "outcome", "status_code", "client_ip"):
                 value = filters.get(key)
                 if value is not None and value != "":
                     clauses.append(f"{key}=?")

app/observability.py

@@ -1,23 +1,98 @@
 """Observe inference metadata with bounded response parsing; never inspect request bodies or headers.
+
+The only headers read are audit client metadata (user-agent and forwarded-for
+chains behind a configured trusted proxy); they are never forwarded upstream.
 Preserve the ASGI wire and exclude credential files or tokens from route metadata.
 """
 from __future__ import annotations
 
 import asyncio
+import ipaddress
 from contextvars import ContextVar
 from dataclasses import dataclass, field
 import json
 import time
 from typing import Any
 
-from app.audit_store import AuditStore, METRICS, number, safe_attempt, safe_label
+from app.audit_store import AuditStore, METRICS, number, safe_agent, safe_attempt, safe_label
 from app.request_context import current_context, ensure_context
 
 _PATHS = {"/v1/chat/completions": "chat", "/v1/responses": "responses", "/v1/messages": "messages"}
 _PARSE_LIMIT = 16384
+_META_HEADERS = ("user-agent", "x-forwarded-for", "x-real-ip")
+_proxy_networks: dict[str, tuple] = {}
 _current: ContextVar[_Observation | None] = ContextVar("audit_observation", default=None)
 
 
+def _trusted_networks(raw):
+    """Parse a normalized trusted_proxies setting; cached per setting value."""
+    if not isinstance(raw, str) or not raw.strip():
+        return ()
+    cached = _proxy_networks.get(raw)
+    if cached is None:
+        import ipaddress
+        try:
+            cached = tuple(ipaddress.ip_network(entry.strip(), strict=False)
+                           for entry in raw.split(",") if entry.strip())
+        except ValueError:
+            return ()
+        if len(_proxy_networks) > 8:
+            _proxy_networks.clear()
+        _proxy_networks[raw] = cached
+    return cached
+
+
+def client_meta(scope, trusted_proxies):
+    """(client_ip, user_agent) for the audit record.
+
+    Forwarded headers count only when the direct peer is a trusted proxy;
+    otherwise they are ignored so clients cannot spoof their address.
+    """
+    headers = {}
+    forwarded: list[str] = []
+    for key, value in scope.get("headers") or []:
+        name = key.decode("latin-1").lower() if isinstance(key, (bytes, bytearray)) else str(key).lower()
+        clean = value.decode("latin-1", "replace").strip() if isinstance(value, (bytes, bytearray)) else str(value).strip()
+        if name == "x-forwarded-for":
+            # A trusted proxy may append its hop as a separate header field;
+            # combine the fields in order before walking the chain.
+            if clean:
+                forwarded.append(clean)
+        elif name in _META_HEADERS and name not in headers:
+            headers[name] = clean
+    client = scope.get("client")
+    remote = client[0] if client and isinstance(client[0], str) else None
+
+    def valid_ip(candidate: str | None) -> str | None:
+        try:
+            return str(ipaddress.ip_address(candidate)) if candidate else None
+        except ValueError:
+            return None
+
+    remote_ip = valid_ip(remote)
+    remote_obj = ipaddress.ip_address(remote_ip) if remote_ip else None
+    networks = _trusted_networks(trusted_proxies)
+    ip = remote
+    if remote_obj is not None and any(remote_obj in network for network in networks):
+        hops = [hop.strip() for hop in (",".join(forwarded)).split(",") if hop.strip()][-32:]
+        chosen = None
+        broken = False
+        for hop in reversed(hops):
+            hop_ip = valid_ip(hop)
+            if hop_ip is None:
+                # A malformed hop breaks the chain; never fall back to the
+                # client-controlled leftmost entry in that case.
+                broken = True
+                break
+            chosen = hop_ip
+            if not any(ipaddress.ip_address(hop_ip) in network for network in networks):
+                break
+        if chosen is None or broken:
+            chosen = valid_ip(headers.get("x-real-ip"))
+        ip = chosen or remote
+    return ip, safe_agent(headers.get("user-agent"))
+
+
 def _mapping(value):
     return value if isinstance(value, dict) else {}
 
@@ -333,6 +408,9 @@ async def lifespan_receive():
                                     "detail_generation": -1, "started_at": started,
                                     "protocol": _PATHS[scope["path"]],
                                     **{key: None for key in METRICS}}, monotonic_start)
+        observation.record.update(zip(("client_ip", "user_agent"),
+                                      client_meta(scope, self.config.get("trusted_proxies")
+                                                  if isinstance(self.config, dict) else None)))
         token = _current.set(observation)
         parser = _Parser(observation)
         cancelled = False

app/settings.py

@@ -33,6 +33,23 @@ def normalize_allowed_origins(value):
     return ",".join(normalized)
 
 
+def normalize_trusted_proxies(value):
+    """Normalize a comma/space separated list of trusted proxy IPs or CIDRs."""
+    import ipaddress
+    entries = [entry for entry in re.split(r"[\s,]+", value.strip()) if entry]
+    if len(entries) > 32:
+        raise ValueError("trusted_proxies: 代理数量超出上限")
+    normalized = []
+    for entry in entries:
+        try:
+            network = str(ipaddress.ip_network(entry, strict=False))
+        except ValueError:
+            raise ValueError(f"trusted_proxies: 地址无效 {entry!r}") from None
+        if network not in normalized:
+            normalized.append(network)
+    return ",".join(normalized)
+
+
 def validate_projection_max_bytes(value):
     """Allow zero or enough room for a bounded head/tail warning."""
     if value != 0 and value < 256:
@@ -105,6 +122,9 @@ def _item(default, type_, label, *, mode="hot", env=None, minimum=None, maximum=
     "audit_max_bytes": _item(256 * 1024 * 1024, "integer", "审计明细预算", minimum=1024**2, maximum=1024**4),
     "audit_retention_days": _item(30, "integer", "审计明细保留天数", minimum=1, maximum=36500),
     "audit_diagnostic_bytes": _item(8192, "integer", "失败诊断最大字节", minimum=0, maximum=8192),
+    "trusted_proxies": _item("", "string", "可信代理地址（IP/CIDR，逗号分隔）",
+                             env="CODEBUDDY2API_TRUSTED_PROXIES", allow_empty=True,
+                             max_length=2000, validator=normalize_trusted_proxies),
 }
 
 

docker-compose.yml

@@ -22,6 +22,8 @@ services:
       CODEBUDDY2API_AUTO_ACCEPT_BUDDY: ${CODEBUDDY2API_AUTO_ACCEPT_BUDDY:-false}
       # Unset variables remain configurable through the WebUI.
       CODEBUDDY2API_ADMIN_ORIGINS:
+      # Trusted proxy IPs/CIDRs; forwarded headers count only from these peers.
+      CODEBUDDY2API_TRUSTED_PROXIES:
       CODEBUDDY2API_KEEP_TOOL_METADATA:
       CODEBUDDY2API_MAX_IMAGES: ${CODEBUDDY2API_MAX_IMAGES:-16}
       CODEBUDDY2API_IMAGE_POLICY: ${CODEBUDDY2API_IMAGE_POLICY:-truncate}

docs/deployment.md

@@ -13,6 +13,7 @@ On first setup, run `cp .env.example .env` and set `CODEBUDDY2API_KEY` to your o
 | `CODEBUDDY2API_IMAGE` | Compose image; the template uses `codebuddy2api:local` |
 | `CODEBUDDY2API_BIND` / `CODEBUDDY2API_PORT` | Native listener and Compose host mapping; default `127.0.0.1:8787` |
 | `CODEBUDDY2API_AUTH_PATH` | Compose host data directory; defaults to `./auth`, mounted at `/data/auth` |
+| `CODEBUDDY2API_TRUSTED_PROXIES` | Trusted proxy IPs/CIDRs (comma separated); when the direct peer matches, the audit log records the client from `X-Forwarded-For`/`X-Real-IP` instead of the proxy address |
 | `CODEBUDDY_AUTH_DIR` | Local Python data directory; defaults to the repository's `auth/`. Compose sets it to `/data/auth` inside the container |
 | `CODEBUDDY_IMPORT_DIR` | Optional import directory; defaults to `imports/` under the data directory. Use container paths with Compose |
 The example lists all active runtime variables, including inbound/aggregate byte limits, concurrency, tool retries and failover. Compose forwards these limits; unset optional settings (tool metadata, Responses projection, origin allowlist, failover and similar) remain configurable in the WebUI. Zero disables aggregate/concurrency limits, per-item Responses trimming or extra retries, not the required positive inbound limit.
@@ -75,6 +76,8 @@ Without uv, run `python3 -m venv .venv`, activate it, install dependencies with
 
 Both commands optionally read `.env` in the current working directory, never parent directories. Precedence: explicit CLI > process environment > `.env` > saved SQLite values > defaults. Overrides do not replace the saved default key; an explicitly empty key still locks management. Listener changes require restart; `CODEBUDDY2API_IMAGE/AUTH_PATH` are Compose-only.
 
+> The audit schema is single-writer: stop the old instance before starting a new version, and never run two gateway versions against one auth directory.
+
 ### Upgrade and state migration
 
 Stop the gateway and back up the entire data directory first. Startup imports legacy JSON sessions, cooldowns, credit/trial history, catalogs and usage into `control.sqlite3` once. Old files remain as backups and are no longer read or updated. Invalid critical state or a failed migration stops startup; repair it or restore a backup rather than deleting ledgers to bypass validation.

docs/deployment.zh-CN.md

@@ -13,6 +13,7 @@
 | `CODEBUDDY2API_IMAGE` | Compose 镜像；模板默认 `codebuddy2api:local` |
 | `CODEBUDDY2API_BIND` / `CODEBUDDY2API_PORT` | 本地监听与 Compose 宿主机映射；默认 `127.0.0.1:8787` |
 | `CODEBUDDY2API_AUTH_PATH` | Compose 宿主机数据目录；默认 `./auth`，挂载到容器 `/data/auth` |
+| `CODEBUDDY2API_TRUSTED_PROXIES` | 可信代理 IP/CIDR（逗号分隔）；直连地址命中时审计改记 `X-Forwarded-For`/`X-Real-IP` 里的真实客户端地址，而不是代理地址 |
 | `CODEBUDDY_AUTH_DIR` | 本地 Python 数据目录；默认仓库内 `auth/`。Compose 在容器内固定为 `/data/auth` |
 | `CODEBUDDY_IMPORT_DIR` | 可选导入目录；默认数据目录下的 `imports/`。Compose 中请使用容器路径 |
 
@@ -76,6 +77,8 @@ uv run converter.py
 
 两种启动方式均可选读取当前工作目录的 `.env`，不搜索父目录。优先级：显式 CLI > 进程环境变量 > `.env` > SQLite 保存值 > 默认值。覆盖不改写已保存的默认 key；显式空 key 仍锁定管理。修改监听需重启，`CODEBUDDY2API_IMAGE/AUTH_PATH` 仅用于 Compose。
 
+> 审计库为单写者设计：升级新版本前先停止旧实例，不要让两个版本的网关共用同一 auth 目录。
+
 ### 升级与数据迁移
 
 升级前停止网关并备份整个数据目录。首次启动将旧 JSON 会话、冷却、积分/领取、目录和用量状态一次性导入 `control.sqlite3`；原文件保留作备份，不再参与读写。关键数据损坏或迁移失败会阻止启动，请修复或恢复备份，不要删除账本绕过检查。