test: add MCP smoke tests over stdio and streamable HTTP, from source and from the built image

The audit logger is opened at startup (exit 2 on failure) and obtained before
every call, which is refused with audit_unavailable otherwise. Audit records only
take validated path and sudo values and use the request id as correlation id. The
rate limiter evicts idle keys, caps tracked keys, audits the first excess per key
and window and uses the settings passed to the app. FORWARDED_ALLOW_IPS is now a
setting, and startup warns about a missing host allow-list and unavailable OAuth.

.env.example

@@ -21,6 +21,11 @@ MCP_BEARER_TOKEN=
 PUBLIC_BASE_URL=
 # Extra Host header values accepted behind a reverse proxy (comma separated).
 ALLOWED_HOSTS=
+# Proxy addresses whose X-Forwarded-For is trusted (comma separated IPs or
+# networks). The default only trusts a proxy on this host. Behind a Docker proxy
+# on another address list that address; use * only if the port is reachable
+# solely from the proxy.
+FORWARDED_ALLOW_IPS=127.0.0.1
 
 # --- Per-user OAuth (HTTP only) -----------------------------------------------
 OAUTH_ENABLED=false

README.md

@@ -237,8 +237,11 @@ when the proxy forwards a different `Host`.
 Bind rules: the server binds to `127.0.0.1` by default. A non-loopback `MCP_HOST` refuses to
 start (exit code 2) unless `MCP_BEARER_TOKEN` or a complete OAuth setup is configured. Requests
 with an unexpected `Host` header are rejected (DNS-rebinding protection). The MCP endpoint is
-rate limited (`RATE_LIMIT_PER_MINUTE`), keyed by a hash of the Authorization header, else the
-client IP.
+rate limited (`RATE_LIMIT_PER_MINUTE`) per client IP and per hash of the Authorization header.
+Behind a proxy on another address set `FORWARDED_ALLOW_IPS` so the real client IP is used. A
+non-loopback bind without `PUBLIC_BASE_URL` or `ALLOWED_HOSTS` logs a startup warning (remote
+`Host` headers get 421). The server also exits with code 2 when the audit log cannot be opened
+for append, and refuses tool calls it cannot audit.
 
 More in [docs/deployment.md](docs/deployment.md) and [docs/hardening.md](docs/hardening.md).
 

docs/audit-logging.md

@@ -55,6 +55,13 @@ keep elsewhere.
 The log is written to `AUDIT_LOG_PATH`, default `DATA_DIR/audit.log` (`/var/lib/aegis-mcp` in the
 image). Keep it on durable storage, readable and writable only by the service account.
 
+The server opens the file for append at startup and exits with code 2 (message on stderr) when
+that fails, so it never runs unaudited. Each call also obtains the audit logger before it runs;
+if that fails the call is refused with `audit_unavailable` and never reaches Gitea. A write
+failure after a call has executed is only logged to stderr: the Gitea side effect already
+happened and the caller must still receive the result. The `correlation_id` of a tool entry is
+the request id, which is also the entry's `request_id`.
+
 ## Rotation
 
 The server appends to one file and does not rotate it, because rotating mid-stream would break

docs/configuration.md

@@ -30,7 +30,8 @@ when OAuth is on.
 | `MCP_PATH` | `/mcp` | Starts with a single `/`, no query, spaces or trailing slash, not just `/`. |
 | `MCP_BEARER_TOKEN` | unset | Static bearer token for HTTP (admin identity). At least 32 characters. Generate with `openssl rand -hex 32`. |
 | `PUBLIC_BASE_URL` | unset | Public URL behind a reverse proxy. Same URL rules as `GITEA_URL`, and `https` unless the host is loopback. Required for OAuth. |
-| `ALLOWED_HOSTS` | empty | Extra `Host` header values accepted behind a proxy. The loopback names with the port and the host of `PUBLIC_BASE_URL` are always allowed. Other `Host` values are rejected (DNS-rebinding protection). |
+| `ALLOWED_HOSTS` | empty | Extra `Host` header values accepted behind a proxy. The loopback names with the port and the host of `PUBLIC_BASE_URL` are always allowed. Other `Host` values are rejected (DNS-rebinding protection). A non-loopback `MCP_HOST` with neither `PUBLIC_BASE_URL` nor `ALLOWED_HOSTS` logs a startup warning on stderr, because remote clients then get 421. |
+| `FORWARDED_ALLOW_IPS` | `127.0.0.1` | Comma list of IP addresses or networks (or `*`) whose `X-Forwarded-For` and `X-Forwarded-Proto` headers are trusted. Set it to the address of the reverse proxy when it is not on the same host, for example a Docker network. Otherwise the per-IP rate limit sees the proxy address for every client. `*` trusts every peer and logs a warning: only use it when the port is reachable solely from the proxy, since clients could otherwise spoof their IP. |
 
 ## Per-user OAuth (HTTP only)
 
@@ -50,7 +51,7 @@ when OAuth is on.
 | Variable | Default | Rules |
 |---|---|---|
 | `DATA_DIR` | `~/.local/state/aegis-gitea-mcp` (`/var/lib/aegis-mcp` in the image) | `~` is expanded. Holds the token store (`oauth-tokens.json`, mode 0600) and the audit log. |
-| `AUDIT_LOG_PATH` | `DATA_DIR/audit.log` | Path of the hash-chained audit log. See [audit-logging.md](audit-logging.md). |
+| `AUDIT_LOG_PATH` | `DATA_DIR/audit.log` | Path of the hash-chained audit log. See [audit-logging.md](audit-logging.md). The server opens it for append at startup and exits with code 2 and a message on stderr when it cannot. |
 
 ## Tool exposure
 
@@ -74,7 +75,7 @@ listed nor callable.
 | `SECRET_DETECTION_MODE` | `mask` | `off`, `mask` or `block`. See below. |
 | `LOG_LEVEL` | `INFO` | `DEBUG`, `INFO`, `WARNING`, `ERROR` or `CRITICAL`. Logs go to stderr. |
 | `EXPOSE_ERROR_DETAILS` | `false` | When true, the exception class name is added to internal and transport errors. Never a stack trace. Do not enable in production. |
-| `RATE_LIMIT_PER_MINUTE` | `120` | At least 1. Per client on the HTTP MCP endpoint, keyed by a hash of the Authorization header, else the client IP. Excess requests get 429. |
+| `RATE_LIMIT_PER_MINUTE` | `120` | At least 1. Per client on the HTTP MCP endpoint, keyed by a hash of the Authorization header, else the client IP. Excess requests get 429. Only the first rejected request per client and minute is written to the audit log. Tracking is in memory and bounded (idle clients are forgotten). |
 
 Secret detection: `mask` replaces secret-like strings in results with a placeholder, `block`
 withholds a whole result that contains one (`secret_blocked`), `off` disables pattern matching.

docs/deployment.md

@@ -81,7 +81,15 @@ in mind for long tool calls.
 
 The server rejects requests whose `Host` is not allowed (DNS-rebinding protection). Allowed hosts
 are the loopback names with the port, the host of `PUBLIC_BASE_URL`, and `ALLOWED_HOSTS`. If
-you reach it under another name, add that name to `ALLOWED_HOSTS`.
+you reach it under another name, add that name to `ALLOWED_HOSTS`. With a non-loopback
+`MCP_HOST` (the container) and neither `PUBLIC_BASE_URL` nor `ALLOWED_HOSTS`, every remote `Host`
+is rejected with 421, and the server logs a warning at startup naming the two variables.
+
+The server trusts `X-Forwarded-For` only from `FORWARDED_ALLOW_IPS` (default `127.0.0.1`). When
+the proxy reaches the container from another address, for example a Docker network, add that
+address or network (`FORWARDED_ALLOW_IPS=172.18.0.0/16`). Without it every request looks like it
+comes from the proxy and shares one per-IP rate limit bucket. `*` trusts any peer, so use it only
+when the port is reachable solely from the proxy.
 
 ## OAuth application in Gitea
 

docs/hardening.md

@@ -40,6 +40,7 @@ authentication), but it does not decide what is safe. This checklist is how you
   rejected.
 - Restrict who can reach the endpoint at the proxy or firewall too (VPN, allow list), as a
   second layer in front of bearer authentication.
+- Set `FORWARDED_ALLOW_IPS` to the proxy address only (not `*`), so a client cannot spoof its IP.
 - Leave `RATE_LIMIT_PER_MINUTE` on. Put request-size and connection limits at the proxy.
 - Use https for `GITEA_URL`. Plain http sends the token unencrypted (setup warns about it).
 

docs/troubleshooting.md

@@ -34,6 +34,7 @@ client fails to parse the stream. Logs go to stderr; look at the client's MCP se
 | `redirect` | Gitea answered with a redirect, which is never followed. Use the final URL (for example https) as `GITEA_URL`. |
 | `payload_too_large` | The response is above `MAX_BINARY_BYTES`. Raise it or fetch a smaller object. |
 | `secret_blocked` | `SECRET_DETECTION_MODE=block` withheld a result containing secret-like content. Use `mask` if that is acceptable. |
+| `audit_unavailable` | The audit log could not be opened, so the call was refused before it reached Gitea. Fix the permissions of `AUDIT_LOG_PATH` (or `DATA_DIR`) and restart. |
 | `rate_limited` (429) | Gitea or this server's `RATE_LIMIT_PER_MINUTE` limit. |
 
 A result with a `truncated` object was cut to `MAX_TOOL_RESPONSE_ITEMS` or
@@ -48,9 +49,15 @@ A tool the client cannot see is probably filtered by `TOOLSETS`, `READ_ONLY` or
   10 seconds). Fix connectivity from the server's host.
 - 401 on `/mcp`: no or wrong bearer token, and no valid OAuth token. The response has a
   `WWW-Authenticate` header that points clients to the OAuth metadata.
-- Request rejected with an invalid Host: the name you used is not allowed. Add it to
-  `ALLOWED_HOSTS` or set `PUBLIC_BASE_URL`.
-- 429: the client exceeded `RATE_LIMIT_PER_MINUTE`.
+- Request rejected with an invalid Host (421): the name you used is not allowed. Add it to
+  `ALLOWED_HOSTS` or set `PUBLIC_BASE_URL`. With a non-loopback `MCP_HOST` and neither variable
+  the server prints a warning at startup that says so.
+- 429: the client exceeded `RATE_LIMIT_PER_MINUTE`. If every client seems to share one limit,
+  the proxy address is not in `FORWARDED_ALLOW_IPS`, so all requests look like they come from
+  the proxy.
+- The server exits with code 2 and "cannot open the audit log": the audit file or its directory
+  is not writable. Fix the permissions, or point `AUDIT_LOG_PATH` or `DATA_DIR` at a writable
+  location (in Docker, check the volume ownership). The server never runs without an audit log.
 - Responses arrive in one chunk or time out behind nginx: set `proxy_buffering off;`.
 
 ## OAuth sign-in

pyproject.toml

@@ -86,6 +86,10 @@ ignore = [
     "B008",  # function calls in argument defaults (pydantic Field style)
 ]
 
+[tool.ruff.lint.isort]
+# tests/mcp is a directory, not the SDK: keep "mcp" classified as third party.
+known-third-party = ["mcp"]
+
 [tool.ruff.lint.per-file-ignores]
 "__init__.py" = ["F401"]
 "tests/integration/harness.py" = [
@@ -129,6 +133,7 @@ pythonpath = ["src"]
 asyncio_mode = "auto"
 markers = [
     "integration: needs a live Gitea in Docker (opt-in via AEGIS_INTEGRATION=1)",
+    "mcp_smoke: MCP client smoke tests over real transports (opt-in via AEGIS_INTEGRATION=1)",
 ]
 
 [tool.coverage.run]

src/aegis_gitea_mcp/auth.py

@@ -9,6 +9,7 @@ token to the :class:`~aegis_gitea_mcp.identity.Identity` a tool call runs as.
 
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import hmac
 import time
@@ -103,6 +104,7 @@ class AdminIdentityResolver:
         self._executor = executor
         self._login: str | None = None
         self._retry_at = 0.0
+        self._lookup_lock = asyncio.Lock()
 
     async def resolve(self, access: AccessToken | None) -> Identity:
         """Return the admin identity, or raise for an access token of another kind."""
@@ -112,6 +114,9 @@ class AdminIdentityResolver:
         if secret is None:
             raise ToolError("admin-token mode needs GITEA_TOKEN, which is not configured")
         token = secret.get_secret_value()
+        if self._login is None and time.monotonic() >= self._retry_at:
+            # One lookup for concurrent first calls: later callers wait, then see the result.
+            async with self._lookup_lock:
                 if self._login is None and time.monotonic() >= self._retry_at:
                     self._login = await self._lookup_login(token)
                     if self._login is None:

src/aegis_gitea_mcp/cli.py

@@ -17,6 +17,17 @@ from aegis_gitea_mcp.config import Settings, get_settings, reset_settings
 from aegis_gitea_mcp.logging_utils import configure_logging
 
 
+def _port(value: str) -> int:
+    """Parse ``--port`` (argparse ``type``): an integer from 1 to 65535."""
+    try:
+        port = int(value)
+    except ValueError:
+        raise argparse.ArgumentTypeError("must be an integer from 1 to 65535") from None
+    if not 1 <= port <= 65535:
+        raise argparse.ArgumentTypeError("must be between 1 and 65535")
+    return port
+
+
 def build_parser() -> argparse.ArgumentParser:
     """Build the argument parser."""
     parser = argparse.ArgumentParser(
@@ -35,9 +46,8 @@ def build_parser() -> argparse.ArgumentParser:
     serve.add_argument("--host", default=None, help="HTTP bind address (default: MCP_HOST)")
     serve.add_argument(
         "--port",
-        type=int,
+        type=_port,
         default=None,
-        choices=range(1, 65536),
         metavar="PORT",
         help="HTTP port (default: MCP_PORT)",
     )

src/aegis_gitea_mcp/config.py

@@ -128,6 +128,7 @@ class Settings(BaseSettings):
     mcp_bearer_token: SecretStr | None = Field(default=None)
     public_base_url: str | None = Field(default=None)
     allowed_hosts: Annotated[list[str], NoDecode] = Field(default_factory=list)
+    forwarded_allow_ips: str = Field(default="127.0.0.1")
 
     # Per-user OAuth (HTTP only)
     oauth_enabled: bool = Field(default=False)
@@ -199,6 +200,28 @@ class Settings(BaseSettings):
             raise ValueError("PUBLIC_BASE_URL must use https unless the host is loopback")
         return normalized
 
+    @field_validator("forwarded_allow_ips", mode="before")
+    @classmethod
+    def _validate_forwarded_allow_ips(cls, value: Any) -> Any:
+        """Accept ``*`` or a comma list of IP addresses and networks (blank means the default)."""
+        if not isinstance(value, str):
+            return value
+        items = [item.strip() for item in value.split(",") if item.strip()]
+        if not items:
+            return "127.0.0.1"
+        if "*" in items and len(items) > 1:
+            raise ValueError("FORWARDED_ALLOW_IPS: '*' cannot be combined with other entries")
+        for item in items:
+            if item == "*":
+                continue
+            try:
+                ipaddress.ip_network(item, strict=False)
+            except ValueError:
+                raise ValueError(
+                    "FORWARDED_ALLOW_IPS must be '*' or a comma list of IP addresses or networks"
+                ) from None
+        return ",".join(items)
+
     @field_validator("mcp_path")
     @classmethod
     def _validate_mcp_path(cls, value: str) -> str:

src/aegis_gitea_mcp/rate_limit.py

@@ -1,14 +1,29 @@
-"""In-memory request rate limiting for MCP endpoints."""
+"""In-memory request rate limiting for the HTTP MCP endpoint.
+
+The limiter runs before authentication, so every key it sees is attacker
+controlled. Memory is therefore bounded: a key whose window has emptied is
+dropped, and each limiter holds at most ``max_keys`` keys (least recently used
+keys are evicted first). Per-IP limiting always applies; the per-token bucket is
+keyed by a hash of the Authorization header and is only a second line.
+"""
 
 from __future__ import annotations
 
 import hashlib
+import logging
 import time
-from collections import defaultdict, deque
+from collections import OrderedDict, deque
 from dataclasses import dataclass
 
 from aegis_gitea_mcp.audit import get_audit_logger
-from aegis_gitea_mcp.config import get_settings
+from aegis_gitea_mcp.config import Settings
+
+logger = logging.getLogger(__name__)
+
+#: Maximum number of distinct client IPs tracked at once.
+MAX_IP_KEYS = 100_000
+#: Maximum number of distinct Authorization hashes tracked at once.
+MAX_TOKEN_KEYS = 10_000
 
 
 @dataclass(frozen=True)
@@ -20,25 +35,52 @@ class RateLimitDecision:
 
 
 class SlidingWindowLimiter:
-    """Sliding-window limiter keyed by arbitrary identifiers."""
+    """Sliding-window limiter keyed by arbitrary identifiers, with bounded memory."""
 
-    def __init__(self, max_requests: int, window_seconds: int) -> None:
-        """Initialize a fixed-window limiter.
+    def __init__(self, max_requests: int, window_seconds: int, max_keys: int = MAX_IP_KEYS) -> None:
+        """Initialize a sliding-window limiter.
 
         Args:
             max_requests: Maximum allowed requests within window.
             window_seconds: Rolling time window length.
+            max_keys: Maximum number of tracked keys; the least recently used key is
+                evicted beyond it.
         """
         self.max_requests = max_requests
         self.window_seconds = window_seconds
-        self._events: dict[str, deque[float]] = defaultdict(deque)
+        self.max_keys = max_keys
+        self._events: OrderedDict[str, deque[float]] = OrderedDict()
+        self._reported: dict[str, float] = {}
+        self._next_sweep = 0.0
+
+    def __len__(self) -> int:
+        return len(self._events)
+
+    def _sweep(self, now: float) -> None:
+        """Drop keys with no event inside the window (at most once per window)."""
+        if now < self._next_sweep:
+            return
+        self._next_sweep = now + self.window_seconds
+        boundary = now - self.window_seconds
+        for key in [k for k, events in self._events.items() if not events or events[-1] < boundary]:
+            del self._events[key]
+        for key in [k for k in self._reported if k not in self._events]:
+            del self._reported[key]
 
     def allow(self, key: str) -> bool:
         """Check and record request for the provided key."""
         now = time.time()
         boundary = now - self.window_seconds
+        self._sweep(now)
 
-        events = self._events[key]
+        events = self._events.get(key)
+        if events is None:
+            events = self._events[key] = deque()
+            while len(self._events) > self.max_keys:
+                evicted, _ = self._events.popitem(last=False)
+                self._reported.pop(evicted, None)
+        else:
+            self._events.move_to_end(key)
         while events and events[0] < boundary:
             events.popleft()
 
@@ -48,33 +90,62 @@ class SlidingWindowLimiter:
         events.append(now)
         return True
 
+    def first_excess(self, key: str) -> bool:
+        """Return True once per window for a key that is over its limit.
+
+        Callers use it to audit only the first rejected request of a burst instead
+        of every one (which would let an attacker fill the audit log).
+        """
+        now = time.time()
+        reported_at = self._reported.get(key)
+        if reported_at is not None and now - reported_at < self.window_seconds:
+            return False
+        self._reported[key] = now
+        return True
+
 
 class RequestRateLimiter:
-    """Combined per-IP and per-token request limiter."""
+    """Combined per-IP and per-token request limiter.
 
-    def __init__(self) -> None:
-        """Initialize with current settings."""
-        settings = get_settings()
-        self._audit = get_audit_logger()
-        self._ip_limiter = SlidingWindowLimiter(settings.rate_limit_per_minute, 60)
-        self._token_limiter = SlidingWindowLimiter(settings.rate_limit_per_minute, 60)
+    Args:
+        settings: Application settings (``RATE_LIMIT_PER_MINUTE``).
+    """
+
+    def __init__(self, settings: Settings) -> None:
+        limit = settings.rate_limit_per_minute
+        self._ip_limiter = SlidingWindowLimiter(limit, 60, MAX_IP_KEYS)
+        self._token_limiter = SlidingWindowLimiter(limit, 60, MAX_TOKEN_KEYS)
+
+    @staticmethod
+    def _audit_excess(event_type: str, description: str, severity: str, client_ip: str) -> None:
+        """Record a limit hit; an unavailable audit log must not break request handling."""
+        try:
+            get_audit_logger().log_security_event(
+                event_type=event_type,
+                description=description,
+                severity=severity,
+                metadata={"client_ip": client_ip},
+            )
+        except Exception as exc:
+            logger.error("audit log write failed: %s", type(exc).__name__)
 
     def check(self, client_ip: str, token: str | None) -> RateLimitDecision:
         """Evaluate request against IP and token limits.
 
         Args:
             client_ip: Request source IP.
-            token: Optional authenticated API token.
+            token: Optional Authorization header value.
 
         Returns:
             Rate limit decision.
         """
         if not self._ip_limiter.allow(client_ip):
-            self._audit.log_security_event(
-                event_type="rate_limit_ip_exceeded",
-                description="Per-IP request rate limit exceeded",
-                severity="medium",
-                metadata={"client_ip": client_ip},
+            if self._ip_limiter.first_excess(client_ip):
+                self._audit_excess(
+                    "rate_limit_ip_exceeded",
+                    "Per-IP request rate limit exceeded",
+                    "medium",
+                    client_ip,
                 )
             return RateLimitDecision(False, "Per-IP rate limit exceeded")
 
@@ -82,29 +153,13 @@ class RequestRateLimiter:
             # Hash token before using it as a key to avoid storing secrets in memory maps.
             token_key = hashlib.sha256(token.encode("utf-8")).hexdigest()
             if not self._token_limiter.allow(token_key):
-                self._audit.log_security_event(
-                    event_type="rate_limit_token_exceeded",
-                    description="Per-token request rate limit exceeded",
-                    severity="high",
-                    metadata={"client_ip": client_ip},
+                if self._token_limiter.first_excess(token_key):
+                    self._audit_excess(
+                        "rate_limit_token_exceeded",
+                        "Per-token request rate limit exceeded",
+                        "high",
+                        client_ip,
                     )
                 return RateLimitDecision(False, "Per-token rate limit exceeded")
 
         return RateLimitDecision(True, "within limits")
-
-
-_rate_limiter: RequestRateLimiter | None = None
-
-
-def get_rate_limiter() -> RequestRateLimiter:
-    """Get global request limiter."""
-    global _rate_limiter
-    if _rate_limiter is None:
-        _rate_limiter = RequestRateLimiter()
-    return _rate_limiter
-
-
-def reset_rate_limiter() -> None:
-    """Reset global limiter (primarily for tests)."""
-    global _rate_limiter
-    _rate_limiter = None

src/aegis_gitea_mcp/server.py

@@ -347,6 +347,7 @@ class _Record:
     """Everything one audit entry needs, filled in as the call progresses."""
 
     tool: str
+    request_id: str = ""
     domain: str | None = None
     method: str | None = None
     path: str | None = None
@@ -405,10 +406,24 @@ class _Dispatcher:
     ) -> types.CallToolResult:
         """Handle ``tools/call``; every call produces exactly one audit entry."""
         started = time.perf_counter()
-        set_request_id(uuid.uuid4().hex)
+        request_id = uuid.uuid4().hex
+        set_request_id(request_id)
         arguments = params.arguments or {}
-        record = _Record(tool=params.name[:_MAX_UNKNOWN_NAME_CHARS])
+        record = _Record(tool=params.name[:_MAX_UNKNOWN_NAME_CHARS], request_id=request_id)
         record.argument_names = sorted(_clip(key) for key in arguments)[:_MAX_AUDIT_ARGUMENT_NAMES]
+        # Fail closed: the audit logger is obtained BEFORE anything runs, so a call is
+        # never executed (and its side effects never happen) without a way to record it.
+        try:
+            audit = self._audit_override or get_audit_logger()
+        except Exception as exc:
+            logger.error("audit log unavailable, refusing the call: %s", type(exc).__name__)
+            return _result(
+                _error_payload(
+                    "audit_unavailable",
+                    "The audit log is not writable, so the call was not executed",
+                ),
+                is_error=True,
+            )
         try:
             entry = self._registry.get(params.name)
             if entry is None:
@@ -426,7 +441,7 @@ class _Dispatcher:
                     message += f" ({type(exc).__name__})"
                 return _result(_error_payload("internal_error", message), is_error=True)
         finally:
-            self._audit(record, (time.perf_counter() - started) * 1000)
+            self._audit(audit, record, (time.perf_counter() - started) * 1000)
 
     async def _dispatch(
         self, entry: _Entry, arguments: dict[str, Any], record: _Record
@@ -434,20 +449,19 @@ class _Dispatcher:
         operation = entry.operation
         if operation is not None:
             record.method, record.path = operation.method, operation.path
-        else:
-            method, path = arguments.get("method"), arguments.get("path")
-            if isinstance(method, str):
-                record.method = method.upper()[:10]
-            if isinstance(path, str):
-                record.path = path[:_MAX_AUDIT_PATH_CHARS]
-        sudo = arguments.get("sudo")
-        record.sudo = sudo[:NAME_MAX_LENGTH] if isinstance(sudo, str) else None
 
         invalid = validate_arguments(self._registry.validator(entry.tool.name), arguments)
         if invalid is None and operation is not None:
             invalid = _upload_cap_error(operation, arguments, self._settings.max_binary_bytes)
         if invalid is not None:
             return self._failed(record, "invalid_argument", {"error": invalid})
+        # Only validated values reach the audit record: a rejected call may carry
+        # anything (including a secret) in its raw arguments.
+        sudo = arguments.get("sudo")
+        record.sudo = sudo[:NAME_MAX_LENGTH] if isinstance(sudo, str) else None
+        if operation is None:
+            record.method = arguments["method"]
+            record.path = arguments["path"][:_MAX_AUDIT_PATH_CHARS]
         if operation is None and self._settings.read_only and arguments["method"] != "GET":
             message = "READ_ONLY mode is on: gitea_api_request only accepts GET"
             return self._failed(record, "read_only", _error_payload("read_only", message))
@@ -492,15 +506,23 @@ class _Dispatcher:
             record.outcome = "success"
         return _result(outcome.payload, is_error=outcome.is_error, text=outcome.to_text())
 
-    def _audit(self, record: _Record, duration_ms: float) -> None:
+    @staticmethod
+    def _audit(audit: AuditLogger, record: _Record, duration_ms: float) -> None:
+        """Write the audit entry of a finished call.
+
+        A write failure here is logged and swallowed on purpose: the tool already
+        ran, its side effect (possibly a Gitea mutation) cannot be undone, and the
+        caller still has to learn the result. The fail-closed check happens before
+        execution, in :meth:`call_tool`.
+        """
         params: dict[str, Any] = {"argument_names": record.argument_names}
         if record.http_status is not None:
             params["http_status"] = record.http_status
         try:
-            audit = self._audit_override or get_audit_logger()
             audit.log_tool_invocation(
                 record.tool,
                 params=params,
+                correlation_id=record.request_id,
                 result_status="error" if record.is_error else "success",
                 domain=record.domain,
                 method=record.method,

src/aegis_gitea_mcp/setup_template.py

@@ -31,6 +31,11 @@ MCP_BEARER_TOKEN=
 PUBLIC_BASE_URL=
 # Extra Host header values accepted behind a reverse proxy (comma separated).
 ALLOWED_HOSTS=
+# Proxy addresses whose X-Forwarded-For is trusted (comma separated IPs or
+# networks). The default only trusts a proxy on this host. Behind a Docker proxy
+# on another address list that address; use * only if the port is reachable
+# solely from the proxy.
+FORWARDED_ALLOW_IPS=127.0.0.1
 
 # --- Per-user OAuth (HTTP only) -----------------------------------------------
 OAUTH_ENABLED=false

src/aegis_gitea_mcp/transports.py

@@ -31,9 +31,10 @@ from starlette.routing import Route
 from starlette.types import ASGIApp, Receive, Scope, Send
 
 from aegis_gitea_mcp import __version__
+from aegis_gitea_mcp.audit import get_audit_logger
 from aegis_gitea_mcp.auth import CompositeVerifier, IdentityResolver, StaticBearerVerifier
 from aegis_gitea_mcp.config import Settings
-from aegis_gitea_mcp.rate_limit import get_rate_limiter
+from aegis_gitea_mcp.rate_limit import RequestRateLimiter
 from aegis_gitea_mcp.server import (
     SERVER_NAME,
     AegisServer,
@@ -134,9 +135,10 @@ class GiteaProbe:
 class RateLimitMiddleware:
     """ASGI middleware applying the per-IP and per-token limits to the MCP path."""
 
-    def __init__(self, app: ASGIApp, mcp_path: str) -> None:
+    def __init__(self, app: ASGIApp, mcp_path: str, limiter: RequestRateLimiter) -> None:
         self.app = app
         self._mcp_path = mcp_path
+        self._limiter = limiter
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         if scope["type"] != "http" or scope["path"].rstrip("/") != self._mcp_path:
@@ -149,7 +151,7 @@ class RateLimitMiddleware:
             if name == b"authorization":
                 authorization = value.decode("latin-1")
                 break
-        decision = get_rate_limiter().check(client_ip, authorization)
+        decision = self._limiter.check(client_ip, authorization)
         if decision.allowed:
             await self.app(scope, receive, send)
             return
@@ -307,7 +309,9 @@ def build_http_app(
             await gitea_probe.aclose()
 
     app.router.lifespan_context = lifespan
-    app.add_middleware(RateLimitMiddleware, mcp_path=settings.mcp_path)
+    app.add_middleware(
+        RateLimitMiddleware, mcp_path=settings.mcp_path, limiter=RequestRateLimiter(settings)
+    )
     return app
 
 
@@ -333,10 +337,56 @@ def http_startup_errors(settings: Settings, extras: HttpAuthExtras | None) -> li
     return errors
 
 
+def http_startup_warnings(settings: Settings, extras: HttpAuthExtras | None) -> list[str]:
+    """List non-fatal startup problems of the HTTP server (printed to stderr)."""
+    warnings: list[str] = []
+    if (
+        not settings.is_loopback_bind()
+        and not settings.public_base_url
+        and not settings.allowed_hosts
+    ):
+        warnings.append(
+            f"MCP_HOST={settings.mcp_host} is not loopback and neither PUBLIC_BASE_URL nor "
+            "ALLOWED_HOSTS is set: requests whose Host header is not a loopback name or the "
+            "bind address are rejected with 421. Set PUBLIC_BASE_URL (the public https URL) "
+            "or ALLOWED_HOSTS (comma separated host names clients use)"
+        )
+    if settings.oauth_enabled and extras is None and settings.mcp_bearer_token is not None:
+        warnings.append(
+            "OAUTH_ENABLED=true but per-user OAuth is not available in this build: only "
+            "MCP_BEARER_TOKEN is accepted and every call runs as the GITEA_TOKEN owner"
+        )
+    if settings.forwarded_allow_ips == "*":
+        warnings.append(
+            "FORWARDED_ALLOW_IPS=*: X-Forwarded-For is trusted from any peer, so clients can "
+            "spoof their IP and dodge the per-IP rate limit. List the proxy addresses instead"
+        )
+    return warnings
+
+
+def _audit_startup_error(settings: Settings) -> str | None:
+    """Open the audit log now; return an error message when it is not writable.
+
+    A server that cannot record its calls must not start: the dispatcher refuses
+    calls without an audit log, so every call would fail anyway.
+    """
+    try:
+        get_audit_logger()
+    except OSError as exc:
+        reason = exc.strerror or type(exc).__name__
+        return (
+            f"cannot open the audit log {settings.resolved_audit_log_path} for append "
+            f"({reason}). Fix the permissions or point AUDIT_LOG_PATH (or DATA_DIR) at a "
+            "writable location"
+        )
+    return None
+
+
 def run_http(settings: Settings) -> int:
     """Run the streamable HTTP server; return a process exit code.
 
-    Returns 2 without starting when a startup check fails.
+    Returns 2 without starting when a startup check fails (including an audit
+    log that cannot be opened for append).
     """
     extras = load_oauth_components(settings) if settings.oauth_enabled else None
     errors = http_startup_errors(settings, extras)
@@ -344,6 +394,11 @@ def run_http(settings: Settings) -> int:
         for message in errors:
             _fail(message)
         return EXIT_USAGE
+    audit_error = _audit_startup_error(settings)
+    if audit_error is not None:
+        return _fail(audit_error)
+    for message in http_startup_warnings(settings, extras):
+        print(f"aegis-gitea-mcp: warning: {message}", file=sys.stderr)
     try:
         server = build_server(settings, resolver=extras.resolver if extras else None)
     except ServerConfigError as exc:
@@ -363,7 +418,7 @@ def run_http(settings: Settings) -> int:
         port=settings.mcp_port,
         log_level=settings.log_level.lower(),
         proxy_headers=True,
-        forwarded_allow_ips=getattr(settings, "forwarded_allow_ips", None) or "127.0.0.1",
+        forwarded_allow_ips=settings.forwarded_allow_ips,
         server_header=False,
     )
     return 0
@@ -377,11 +432,14 @@ async def _serve_stdio(server: AegisServer) -> None:
 def run_stdio(settings: Settings) -> int:
     """Run the local stdio server (single identity: the owner of ``GITEA_TOKEN``).
 
-    Returns 2 without starting when ``GITEA_TOKEN`` is missing or the tool
-    filters are invalid. All logging goes to stderr; stdout carries JSON-RPC only.
+    Returns 2 without starting when ``GITEA_TOKEN`` is missing, the audit log is
+    not writable or the tool filters are invalid. All logging goes to stderr; stdout carries JSON-RPC only.
     """
     if settings.gitea_token is None:
         return _fail("stdio needs GITEA_TOKEN; create a Gitea access token and set it in .env")
+    audit_error = _audit_startup_error(settings)
+    if audit_error is not None:
+        return _fail(audit_error)
     try:
         server = build_server(settings)
     except ServerConfigError as exc:

tests/conftest.py

@@ -14,7 +14,6 @@ import pytest
 
 from aegis_gitea_mcp.audit import reset_audit_logger
 from aegis_gitea_mcp.config import Settings, reset_settings
-from aegis_gitea_mcp.rate_limit import reset_rate_limiter
 from aegis_gitea_mcp.request_context import clear_gitea_auth_context
 
 
@@ -22,7 +21,6 @@ def _reset_singletons() -> None:
     """Reset every module-level singleton."""
     reset_settings()
     reset_audit_logger()
-    reset_rate_limiter()
     clear_gitea_auth_context()
 
 

tests/mcp/conftest.py

@@ -0,0 +1,76 @@
+"""Opt-in fixtures for the MCP smoke tests.
+
+Tests in this directory are marked ``mcp_smoke`` and skipped unless
+``AEGIS_INTEGRATION=1``. They reuse the live-Gitea harness from
+``tests/integration`` (one disposable Gitea per session, admin token with the
+``all`` scope) and drive the server through a real MCP client over stdio and
+streamable HTTP, from source and (when ``AEGIS_MCP_IMAGE`` is set) from the
+built image.
+"""
+
+from __future__ import annotations
+
+import os
+import secrets
+from pathlib import Path
+
+import pytest
+
+from tests.integration.conftest import ENABLE_ENV, gitea, integration_enabled
+from tests.integration.harness import GITEA_PORT, GiteaInstance
+
+__all__ = ["ENABLE_ENV", "gitea"]
+
+_HERE = Path(__file__).parent.resolve()
+IMAGE_ENV = "AEGIS_MCP_IMAGE"
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    """Register the ``mcp_smoke`` marker (idempotent with pyproject)."""
+    config.addinivalue_line(
+        "markers", "mcp_smoke: MCP client smoke tests, opt-in via AEGIS_INTEGRATION=1"
+    )
+
+
+def _is_mcp_smoke(item: pytest.Item) -> bool:
+    try:
+        return _HERE in Path(str(item.path)).resolve().parents
+    except (OSError, ValueError):  # pragma: no cover - defensive
+        return False
+
+
+def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
+    """Mark and (unless opted in) skip every test under tests/mcp."""
+    selected = [item for item in items if _is_mcp_smoke(item)]
+    skip = pytest.mark.skip(reason=f"MCP smoke tests need {ENABLE_ENV}=1 and Docker")
+    for item in selected:
+        item.add_marker(pytest.mark.mcp_smoke)
+        if not integration_enabled():
+            item.add_marker(skip)
+    if selected and len(selected) == len(items):
+        # A smoke-only run starts the server in subprocesses, so the in-process
+        # coverage gate does not apply.
+        for namespace in (config.option, config.known_args_namespace):
+            if getattr(namespace, "cov_fail_under", None) is not None:
+                namespace.cov_fail_under = 0
+
+
+@pytest.fixture(scope="session")
+def gitea_internal_url(gitea: GiteaInstance) -> str:
+    """URL of the Gitea container for other containers on the harness network."""
+    return f"http://{gitea.container}:{GITEA_PORT}"
+
+
+@pytest.fixture(scope="session")
+def mcp_image() -> str:
+    """The built image under test; the test is skipped when ``AEGIS_MCP_IMAGE`` is unset."""
+    image = os.environ.get(IMAGE_ENV, "").strip()
+    if not image:
+        pytest.skip(f"{IMAGE_ENV} is not set; image smoke tests need a built image")
+    return image
+
+
+@pytest.fixture
+def bearer_token() -> str:
+    """A random 48 character MCP bearer token."""
+    return secrets.token_urlsafe(36)

tests/mcp/helpers.py

@@ -0,0 +1,425 @@
+"""Shared helpers for the MCP smoke tests: client flows, subprocess and docker plumbing.
+
+Secrets (Gitea token, MCP bearer token) are always passed through environment
+variables, never through argv, and every captured stderr/stdout is checked for
+them by the tests.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import json
+import os
+import secrets
+import socket
+import subprocess
+import sys
+import time
+from collections.abc import Iterator, Mapping, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import httpx
+import mcp.types as types
+from mcp import Client
+
+from aegis_gitea_mcp.catalog import load_catalog
+from tests.integration.harness import NAME_PREFIX, GiteaInstance
+
+# Raw-protocol version used for the hand-driven stdio session (handshake era).
+RAW_PROTOCOL_VERSION = "2025-11-25"
+SOURCE_SERVE_ARGS = [sys.executable, "-m", "aegis_gitea_mcp.cli", "serve"]
+HEALTH_TIMEOUT_SECONDS = 60.0
+RAW_READ_TIMEOUT_SECONDS = 60.0
+# Environment variables a child process needs to find its interpreter and docker.
+_PASSTHROUGH_ENV = (
+    "PATH",
+    "HOME",
+    "LANG",
+    "LC_ALL",
+    "TMPDIR",
+    "DOCKER_HOST",
+    "DOCKER_CONFIG",
+    "DOCKER_CONTEXT",
+    "XDG_RUNTIME_DIR",
+)
+
+
+def expected_tool_count() -> int:
+    """Number of tools a full server exposes: every catalog operation plus the raw tool."""
+    return len(load_catalog().operations) + 1
+
+
+def unique_container_name() -> str:
+    """Return a unique ``aegis-it-mcp-<hex>`` container name."""
+    return f"{NAME_PREFIX}mcp-{secrets.token_hex(4)}"
+
+
+def free_loopback_port() -> int:
+    """Return a free TCP port on 127.0.0.1 (same approach as the Gitea harness)."""
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return int(sock.getsockname()[1])
+
+
+def child_env(extra: Mapping[str, str]) -> dict[str, str]:
+    """Minimal environment for a child process: passthrough variables plus ``extra``."""
+    env = {name: os.environ[name] for name in _PASSTHROUGH_ENV if name in os.environ}
+    env.update(extra)
+    return env
+
+
+def server_env(
+    gitea: GiteaInstance, gitea_url: str, data_dir: Path, **extra: str
+) -> dict[str, str]:
+    """Server settings for a subprocess (``GITEA_TOKEN`` included, so keep it out of argv)."""
+    return child_env(
+        {
+            "GITEA_URL": gitea_url,
+            "GITEA_TOKEN": gitea.admin_token,
+            "DATA_DIR": str(data_dir),
+            "LOG_LEVEL": "DEBUG",
+            **extra,
+        }
+    )
+
+
+# ---------------------------------------------------------------------------
+# Client flows
+# ---------------------------------------------------------------------------
+
+
+def result_text(result: types.CallToolResult) -> str:
+    """Return the text of the first content block of a tool result."""
+    assert result.content, "tool result has no content"
+    block = result.content[0]
+    assert isinstance(block, types.TextContent)
+    return block.text
+
+
+def result_payload(result: types.CallToolResult) -> dict[str, Any]:
+    """Parse the JSON envelope of a tool result."""
+    payload = json.loads(result_text(result))
+    assert isinstance(payload, dict)
+    return payload
+
+
+async def exercise_server(
+    client: Client, gitea: GiteaInstance, secrets_to_hide: Sequence[str]
+) -> None:
+    """List tools, run a read call and a write call (create then delete a repo).
+
+    Asserts the tool count, the identity in the envelope, that the repo really
+    existed on Gitea after creation and was gone after deletion, and that no
+    secret appears in any tool result.
+    """
+    listing = await client.list_tools()
+    names = {tool.name for tool in listing.tools}
+    assert len(listing.tools) == expected_tool_count()
+    assert {"gitea_user_get_current", "gitea_repo_delete", "gitea_api_request"} <= names
+
+    texts: list[str] = []
+
+    current = await client.call_tool("gitea_user_get_current", {})
+    texts.append(result_text(current))
+    assert not current.is_error
+    envelope = result_payload(current)
+    assert envelope["status"] == 200
+    assert envelope["identity"]["login"] == gitea.admin_login
+    assert envelope["identity"]["mode"] == "admin-token"
+    assert envelope["data"]["login"] == gitea.admin_login
+
+    repo = gitea.unique("mcp-smoke")
+    created = False
+    try:
+        create = await client.call_tool(
+            "gitea_user_current_create_repo",
+            {"body": {"name": repo, "auto_init": False, "private": True}},
+        )
+        texts.append(result_text(create))
+        assert not create.is_error, result_text(create)
+        created = True
+        assert result_payload(create)["status"] == 201
+        assert gitea.api("GET", f"/repos/{gitea.admin_login}/{repo}").status_code == 200
+
+        delete = await client.call_tool(
+            "gitea_repo_delete", {"owner": gitea.admin_login, "repo": repo}
+        )
+        texts.append(result_text(delete))
+        assert not delete.is_error, result_text(delete)
+        created = False
+        assert gitea.api("GET", f"/repos/{gitea.admin_login}/{repo}").status_code == 404
+    finally:
+        if created:
+            gitea.api("DELETE", f"/repos/{gitea.admin_login}/{repo}")
+
+    for secret in secrets_to_hide:
+        assert all(secret not in text for text in texts), "a secret leaked into a tool result"
+
+
+def assert_no_secrets(text: str, secrets_to_hide: Sequence[str], where: str) -> None:
+    """Fail when any secret appears in ``text`` (the message never prints the secret)."""
+    for index, secret in enumerate(secrets_to_hide):
+        assert secret not in text, f"secret #{index} found in {where}"
+
+
+# ---------------------------------------------------------------------------
+# Hand-driven stdio session (checks that stdout carries JSON-RPC only)
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class RawSession:
+    """Everything a hand-driven stdio run produced."""
+
+    stdout_lines: list[str]
+    stderr: str
+    returncode: int | None
+
+
+async def _read_response(stdout: asyncio.StreamReader, request_id: int, lines: list[str]) -> None:
+    """Read stdout lines until the response for ``request_id`` arrived (lines are recorded)."""
+    while True:
+        raw = await asyncio.wait_for(stdout.readline(), timeout=RAW_READ_TIMEOUT_SECONDS)
+        if not raw:
+            raise AssertionError("server closed stdout before answering")
+        line = raw.decode("utf-8").rstrip("\r\n")
+        lines.append(line)
+        message = json.loads(line)
+        if (
+            isinstance(message, dict)
+            and message.get("id") == request_id
+            and "method" not in message
+        ):
+            assert "result" in message, f"request {request_id} failed"
+            return
+
+
+async def run_raw_stdio(
+    argv: Sequence[str], env: Mapping[str, str], cwd: Path | None = None
+) -> RawSession:
+    """Start ``argv`` and run initialize + tools/list by hand, then close stdin.
+
+    Returns every stdout line the process wrote (so the caller can assert that
+    each one is a JSON-RPC message), its stderr and the exit code.
+    """
+    proc = await asyncio.create_subprocess_exec(
+        *argv,
+        stdin=asyncio.subprocess.PIPE,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+        env=dict(env),
+        cwd=cwd,
+        limit=64 * 1024 * 1024,
+    )
+    assert proc.stdin is not None
+    assert proc.stdout is not None
+    assert proc.stderr is not None
+    lines: list[str] = []
+    stderr_task = asyncio.create_task(proc.stderr.read())
+    try:
+
+        async def send(message: dict[str, Any]) -> None:
+            proc.stdin.write((json.dumps(message) + "\n").encode("utf-8"))  # type: ignore[union-attr]
+            await proc.stdin.drain()  # type: ignore[union-attr]
+
+        await send(
+            {
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "initialize",
+                "params": {
+                    "protocolVersion": RAW_PROTOCOL_VERSION,
+                    "capabilities": {},
+                    "clientInfo": {"name": "aegis-smoke", "version": "0"},
+                },
+            }
+        )
+        await _read_response(proc.stdout, 1, lines)
+        await send({"jsonrpc": "2.0", "method": "notifications/initialized"})
+        await send({"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
+        await _read_response(proc.stdout, 2, lines)
+        proc.stdin.close()
+        rest = await asyncio.wait_for(proc.stdout.read(), timeout=30.0)
+        lines.extend(rest.decode("utf-8").splitlines())
+        await asyncio.wait_for(proc.wait(), timeout=30.0)
+    finally:
+        if proc.returncode is None:
+            with contextlib.suppress(ProcessLookupError):
+                proc.kill()
+            await proc.wait()
+    stderr = (await stderr_task).decode("utf-8", errors="replace")
+    return RawSession(stdout_lines=lines, stderr=stderr, returncode=proc.returncode)
+
+
+def assert_jsonrpc_only(lines: Sequence[str]) -> None:
+    """Assert that every non-empty stdout line is a JSON-RPC 2.0 message."""
+    assert lines, "no output on stdout"
+    for line in lines:
+        if not line.strip():
+            continue
+        message = json.loads(line)
+        assert isinstance(message, dict) and message.get("jsonrpc") == "2.0"
+
+
+# ---------------------------------------------------------------------------
+# HTTP readiness
+# ---------------------------------------------------------------------------
+
+
+def wait_for_health(
+    url: str, *, alive: Any = None, timeout: float = HEALTH_TIMEOUT_SECONDS
+) -> httpx.Response:
+    """Poll ``url`` until it answers 200 and return the response.
+
+    ``alive`` is an optional callable returning False when the server died, so
+    the wait fails fast instead of running into the timeout.
+    """
+    deadline = time.monotonic() + timeout
+    last = "no response"
+    while time.monotonic() < deadline:
+        if alive is not None and not alive():
+            raise AssertionError("the server process exited before /health answered")
+        try:
+            response = httpx.get(url, timeout=3.0)
+            if response.status_code == 200:
+                return response
+            last = f"HTTP {response.status_code}"
+        except httpx.HTTPError as exc:
+            last = type(exc).__name__
+        time.sleep(0.5)
+    raise AssertionError(f"{url} was not healthy within {timeout:.0f}s ({last})")
+
+
+def assert_rejects_without_valid_token(mcp_url: str, bearer_token: str) -> None:
+    """A missing and a wrong bearer token both get 401 (and the real token never echoes back)."""
+    initialize = {
+        "jsonrpc": "2.0",
+        "id": 1,
+        "method": "initialize",
+        "params": {
+            "protocolVersion": RAW_PROTOCOL_VERSION,
+            "capabilities": {},
+            "clientInfo": {"name": "aegis-smoke", "version": "0"},
+        },
+    }
+    base = {"Accept": "application/json, text/event-stream"}
+    anonymous = httpx.post(mcp_url, json=initialize, headers=base, timeout=10.0)
+    assert anonymous.status_code == 401
+    wrong = httpx.post(
+        mcp_url,
+        json=initialize,
+        headers={**base, "Authorization": f"Bearer {secrets.token_urlsafe(36)}"},
+        timeout=10.0,
+    )
+    assert wrong.status_code == 401
+    assert bearer_token not in anonymous.text + wrong.text
+
+
+# ---------------------------------------------------------------------------
+# Source HTTP server
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class HttpServer:
+    """A source-installed HTTP server subprocess."""
+
+    process: subprocess.Popen[bytes]
+    base_url: str
+    stderr_path: Path
+    stdout_path: Path
+
+    @property
+    def mcp_url(self) -> str:
+        """URL of the MCP endpoint."""
+        return f"{self.base_url}/mcp"
+
+    def alive(self) -> bool:
+        """Return True while the process is running."""
+        return self.process.poll() is None
+
+    def stderr_text(self) -> str:
+        """Return what the server wrote to stderr so far."""
+        return self.stderr_path.read_text(encoding="utf-8", errors="replace")
+
+
+@contextlib.contextmanager
+def run_source_http_server(env: Mapping[str, str], workdir: Path) -> Iterator[HttpServer]:
+    """Run ``aegis-gitea-mcp serve --transport http`` on a free loopback port.
+
+    The working directory is an empty temp dir so no ambient ``.env`` is read.
+    Waits for ``/health`` and always stops the process afterwards.
+    """
+    port = free_loopback_port()
+    stderr_path = workdir / "server.stderr"
+    stdout_path = workdir / "server.stdout"
+    with stderr_path.open("wb") as err, stdout_path.open("wb") as out:
+        process = subprocess.Popen(  # noqa: S603
+            [
+                *SOURCE_SERVE_ARGS,
+                "--transport",
+                "http",
+                "--host",
+                "127.0.0.1",
+                "--port",
+                str(port),
+            ],
+            env=dict(env),
+            cwd=workdir,
+            stdin=subprocess.DEVNULL,
+            stdout=out,
+            stderr=err,
+        )
+        server = HttpServer(
+            process=process,
+            base_url=f"http://127.0.0.1:{port}",
+            stderr_path=stderr_path,
+            stdout_path=stdout_path,
+        )
+        try:
+            wait_for_health(f"{server.base_url}/health", alive=server.alive)
+            yield server
+        finally:
+            process.terminate()
+            try:
+                process.wait(timeout=15)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait(timeout=15)
+
+
+# ---------------------------------------------------------------------------
+# Docker
+# ---------------------------------------------------------------------------
+
+
+def docker(
+    *args: str,
+    env: Mapping[str, str] | None = None,
+    check: bool = True,
+    timeout: float = 120.0,
+) -> subprocess.CompletedProcess[str]:
+    """Run ``docker <args>``; ``env`` adds variables the CLI can forward with ``-e NAME``."""
+    full_env = {**os.environ, **(env or {})}
+    proc = subprocess.run(  # noqa: S603
+        ["docker", *args],  # noqa: S607
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+        check=False,
+        env=full_env,
+    )
+    if check and proc.returncode != 0:
+        # Never echo argv: it may carry non-secret config, and stderr is enough to debug.
+        raise AssertionError(
+            f"docker {' '.join(args[:2])} failed (exit {proc.returncode}): {proc.stderr.strip()}"
+        )
+    return proc
+
+
+def remove_container(name: str) -> None:
+    """Force-remove one container by its exact name (best effort, never prunes)."""
+    docker("rm", "-f", "-v", name, check=False, timeout=60)

tests/mcp/test_http.py

@@ -0,0 +1,46 @@
+"""MCP smoke tests: the streamable HTTP transport from source, against a live Gitea."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import httpx2
+from mcp import Client
+from mcp.client.streamable_http import streamable_http_client
+
+from tests.integration.harness import GiteaInstance
+from tests.mcp.helpers import (
+    assert_no_secrets,
+    assert_rejects_without_valid_token,
+    exercise_server,
+    expected_tool_count,
+    run_source_http_server,
+    server_env,
+)
+
+
+async def test_http_lists_reads_and_writes(
+    gitea: GiteaInstance, tmp_path: Path, bearer_token: str
+) -> None:
+    """A real MCP client over streamable HTTP with a bearer token, plus the 401 negatives."""
+    env = server_env(gitea, gitea.base_url, tmp_path, MCP_BEARER_TOKEN=bearer_token)
+    hidden = [gitea.admin_token, bearer_token]
+    with run_source_http_server(env, tmp_path) as server:
+        health = server.base_url + "/health"
+        body = httpx2.get(health, timeout=10.0).json()
+        assert body["status"] == "ok"
+        assert body["tools"] == expected_tool_count()
+
+        async with httpx2.AsyncClient(
+            headers={"Authorization": f"Bearer {bearer_token}"}, timeout=60.0
+        ) as http_client:
+            transport = streamable_http_client(server.mcp_url, http_client=http_client)
+            async with Client(transport, read_timeout_seconds=60.0) as client:
+                await exercise_server(client, gitea, hidden)
+
+        assert_rejects_without_valid_token(server.mcp_url, bearer_token)
+        assert server.alive()
+
+    assert_no_secrets(server.stderr_text(), hidden, "the server stderr")
+    stdout = server.stdout_path.read_text(encoding="utf-8", errors="replace")
+    assert_no_secrets(stdout, hidden, "the server stdout")

tests/mcp/test_image.py

@@ -0,0 +1,212 @@
+"""MCP smoke tests against the built Docker image (``AEGIS_MCP_IMAGE``).
+
+The MCP container joins the harness's per-run network and reaches Gitea by its
+container name. Secrets go into the container with ``-e NAME`` (the value comes
+from the docker CLI's environment, never from argv). Every container started
+here has a unique ``aegis-it-mcp-<hex>`` name and is removed in a ``finally``.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import os
+from collections.abc import Iterator
+from pathlib import Path
+
+import httpx2
+import pytest
+from mcp import Client, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+from tests.integration.harness import NETWORK_MODE_ENV, GiteaInstance
+from tests.mcp.helpers import (
+    assert_jsonrpc_only,
+    assert_no_secrets,
+    assert_rejects_without_valid_token,
+    child_env,
+    docker,
+    exercise_server,
+    expected_tool_count,
+    free_loopback_port,
+    remove_container,
+    run_raw_stdio,
+    unique_container_name,
+    wait_for_health,
+)
+
+CONTAINER_HTTP_PORT = 8080
+HTTP_TIMEOUT_SECONDS = 60.0
+
+
+@pytest.fixture(scope="module", autouse=True)
+def _image_present(mcp_image: str) -> None:
+    """Fail (not skip) when the configured image does not exist locally."""
+    proc = docker("image", "inspect", mcp_image, check=False, timeout=60)
+    assert proc.returncode == 0, f"AEGIS_MCP_IMAGE={mcp_image!r} is not available locally"
+
+
+def _stdio_argv(name: str, network: str, gitea_url: str, image: str) -> list[str]:
+    return [
+        "run",
+        "--rm",
+        "-i",
+        "--name",
+        name,
+        "--network",
+        network,
+        "--label",
+        "aegis-it=1",
+        "-e",
+        f"GITEA_URL={gitea_url}",
+        "-e",
+        "GITEA_TOKEN",
+        image,
+        "serve",
+        "--transport",
+        "stdio",
+    ]
+
+
+async def test_image_stdio_lists_reads_and_writes(
+    gitea: GiteaInstance, gitea_internal_url: str, mcp_image: str, tmp_path: Path
+) -> None:
+    """The image over stdio (docker run -i): list, read, create and delete a repo."""
+    name = unique_container_name()
+    argv = _stdio_argv(name, gitea.network, gitea_internal_url, mcp_image)
+    params = StdioServerParameters(
+        command="docker", args=argv, env=child_env({"GITEA_TOKEN": gitea.admin_token})
+    )
+    stderr_path = tmp_path / "container.stderr"
+    try:
+        with stderr_path.open("w", encoding="utf-8") as errlog:
+            async with Client(stdio_client(params, errlog=errlog), read_timeout_seconds=60.0) as c:
+                await exercise_server(c, gitea, [gitea.admin_token])
+    finally:
+        remove_container(name)
+    assert_no_secrets(
+        stderr_path.read_text(encoding="utf-8", errors="replace"),
+        [gitea.admin_token],
+        "the container stderr",
+    )
+
+
+async def test_image_stdio_stdout_is_json_rpc_only(
+    gitea: GiteaInstance, gitea_internal_url: str, mcp_image: str
+) -> None:
+    """Nothing but JSON-RPC reaches the image's stdout; stderr carries no secret."""
+    name = unique_container_name()
+    argv = ["docker", *_stdio_argv(name, gitea.network, gitea_internal_url, mcp_image)]
+    try:
+        session = await run_raw_stdio(argv, child_env({"GITEA_TOKEN": gitea.admin_token}))
+    finally:
+        remove_container(name)
+    assert_jsonrpc_only(session.stdout_lines)
+    assert session.returncode == 0
+    assert_no_secrets(session.stderr, [gitea.admin_token], "the container stderr")
+
+
+@contextlib.contextmanager
+def _image_http_server(
+    gitea: GiteaInstance, gitea_internal_url: str, image: str, bearer_token: str
+) -> Iterator[tuple[str, str]]:
+    """Run the image in HTTP mode; yield ``(base_url, container_name)``.
+
+    Local mode publishes the port on 127.0.0.1 only; container mode (the job
+    container shares the per-run network) reaches the MCP container by name.
+    """
+    name = unique_container_name()
+    container_mode = os.environ.get(NETWORK_MODE_ENV, "").lower() == "container"
+    if container_mode:
+        base_url = f"http://{name}:{CONTAINER_HTTP_PORT}"
+        allowed_host = f"{name}:{CONTAINER_HTTP_PORT}"
+        publish: list[str] = []
+    else:
+        port = free_loopback_port()
+        base_url = f"http://127.0.0.1:{port}"
+        allowed_host = f"127.0.0.1:{port}"
+        publish = ["-p", f"127.0.0.1:{port}:{CONTAINER_HTTP_PORT}"]
+    try:
+        docker(
+            "run",
+            "-d",
+            "--name",
+            name,
+            "--network",
+            gitea.network,
+            "--label",
+            "aegis-it=1",
+            *publish,
+            "-e",
+            f"GITEA_URL={gitea_internal_url}",
+            "-e",
+            f"ALLOWED_HOSTS={allowed_host}",
+            "-e",
+            "GITEA_TOKEN",
+            "-e",
+            "MCP_BEARER_TOKEN",
+            image,
+            env={"GITEA_TOKEN": gitea.admin_token, "MCP_BEARER_TOKEN": bearer_token},
+        )
+
+        def alive() -> bool:
+            state = docker("inspect", "--format", "{{.State.Running}}", name, check=False)
+            return state.stdout.strip() == "true"
+
+        wait_for_health(f"{base_url}/health", alive=alive)
+        yield base_url, name
+    finally:
+        remove_container(name)
+
+
+async def test_image_http_lists_reads_and_writes(
+    gitea: GiteaInstance, gitea_internal_url: str, mcp_image: str, bearer_token: str
+) -> None:
+    """The image over streamable HTTP: list, read, write, 401 negatives, no secret in the logs."""
+    from mcp.client.streamable_http import streamable_http_client
+
+    hidden = [gitea.admin_token, bearer_token]
+    with _image_http_server(gitea, gitea_internal_url, mcp_image, bearer_token) as (base, name):
+        health = httpx2.get(f"{base}/health", timeout=10.0).json()
+        assert health["status"] == "ok"
+        assert health["tools"] == expected_tool_count()
+
+        async with httpx2.AsyncClient(
+            headers={"Authorization": f"Bearer {bearer_token}"}, timeout=HTTP_TIMEOUT_SECONDS
+        ) as http_client:
+            transport = streamable_http_client(f"{base}/mcp", http_client=http_client)
+            async with Client(transport, read_timeout_seconds=60.0) as client:
+                await exercise_server(client, gitea, hidden)
+
+        assert_rejects_without_valid_token(f"{base}/mcp", bearer_token)
+        logs = docker("logs", name, check=False, timeout=60)
+        assert_no_secrets(logs.stdout + logs.stderr, hidden, "the container logs")
+
+
+def test_image_refuses_open_bind_without_auth(
+    gitea: GiteaInstance, gitea_internal_url: str, mcp_image: str
+) -> None:
+    """The image binds 0.0.0.0 by default and must exit 2 without MCP_BEARER_TOKEN."""
+    name = unique_container_name()
+    try:
+        proc = docker(
+            "run",
+            "--rm",
+            "--name",
+            name,
+            "--label",
+            "aegis-it=1",
+            "-e",
+            f"GITEA_URL={gitea_internal_url}",
+            "-e",
+            "GITEA_TOKEN",
+            mcp_image,
+            "serve",
+            env={"GITEA_TOKEN": gitea.admin_token},
+            check=False,
+            timeout=90,
+        )
+    finally:
+        remove_container(name)
+    assert proc.returncode == 2
+    assert "MCP_BEARER_TOKEN" in proc.stderr
+    assert_no_secrets(proc.stdout + proc.stderr, [gitea.admin_token], "the container output")

tests/mcp/test_stdio.py

@@ -0,0 +1,53 @@
+"""MCP smoke tests: the stdio transport from source, against a live Gitea."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from mcp import Client, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+from tests.integration.harness import GiteaInstance
+from tests.mcp.helpers import (
+    SOURCE_SERVE_ARGS,
+    assert_jsonrpc_only,
+    assert_no_secrets,
+    exercise_server,
+    server_env,
+)
+
+
+def _stdio_params(gitea: GiteaInstance, data_dir: Path) -> StdioServerParameters:
+    return StdioServerParameters(
+        command=SOURCE_SERVE_ARGS[0],
+        args=[*SOURCE_SERVE_ARGS[1:], "--transport", "stdio"],
+        # The working directory is the empty temp dir so no ambient .env is read.
+        env=server_env(gitea, gitea.base_url, data_dir),
+        cwd=data_dir,
+    )
+
+
+async def test_stdio_lists_reads_and_writes(gitea: GiteaInstance, tmp_path: Path) -> None:
+    """A real MCP client over stdio: initialize, list 538 tools, read, create and delete a repo."""
+    stderr_path = tmp_path / "server.stderr"
+    with stderr_path.open("w", encoding="utf-8") as errlog:
+        transport = stdio_client(_stdio_params(gitea, tmp_path), errlog=errlog)
+        async with Client(transport, read_timeout_seconds=60.0) as client:
+            assert client.server_info is not None
+            assert client.server_info.name == "aegis-gitea-mcp"
+            await exercise_server(client, gitea, [gitea.admin_token])
+
+    stderr = stderr_path.read_text(encoding="utf-8", errors="replace")
+    assert_no_secrets(stderr, [gitea.admin_token], "the server stderr")
+
+
+async def test_stdio_stdout_is_json_rpc_only(gitea: GiteaInstance, tmp_path: Path) -> None:
+    """Every stdout line of the stdio server is JSON-RPC; logs go to stderr without secrets."""
+    from tests.mcp.helpers import run_raw_stdio
+
+    params = _stdio_params(gitea, tmp_path)
+    session = await run_raw_stdio([params.command, *params.args], params.env or {}, tmp_path)
+    assert_jsonrpc_only(session.stdout_lines)
+    assert session.returncode == 0
+    assert_no_secrets("\n".join(session.stdout_lines), [gitea.admin_token], "the server stdout")
+    assert_no_secrets(session.stderr, [gitea.admin_token], "the server stderr")

tests/test_auth.py

@@ -17,8 +17,13 @@ from aegis_gitea_mcp.auth import (
     StaticBearerVerifier,
 )
 from aegis_gitea_mcp.errors import ToolError
-from aegis_gitea_mcp.gitea import GiteaExecutor
-from tests.test_server_helpers import API, BEARER_TOKEN, GITEA_TOKEN, server_settings
+from aegis_gitea_mcp.gitea import GiteaExecutor, ToolOutcome
+from tests.test_server_helpers import (
+    API,
+    BEARER_TOKEN,
+    GITEA_TOKEN,
+    server_settings,
+)
 
 # Static bearer verifier
 
@@ -171,3 +176,23 @@ async def test_missing_gitea_token_is_refused() -> None:
 
     with pytest.raises(ToolError, match="GITEA_TOKEN"):
         await resolver.resolve(None)
+
+
+async def test_concurrent_first_calls_send_one_login_lookup() -> None:
+    import asyncio
+
+    lookups: list[str] = []
+
+    class SlowExecutor(GiteaExecutor):
+        async def request(self, method: str, path: str, *args: Any, **kwargs: Any) -> ToolOutcome:
+            lookups.append(path)
+            await asyncio.sleep(0.01)  # a real network round trip yields to other calls
+            return ToolOutcome(False, {"data": {"login": "admin"}}, 200)
+
+    settings = server_settings()
+    resolver = AdminIdentityResolver(settings, SlowExecutor(settings))
+
+    identities = await asyncio.gather(*(resolver.resolve(None) for _ in range(10)))
+
+    assert {identity.login for identity in identities} == {"admin"}
+    assert lookups == ["/user"]

tests/test_cli.py

@@ -33,3 +33,16 @@ def test_package_version_matches_pyproject() -> None:
 
     data = tomllib.loads((Path(__file__).parent.parent / "pyproject.toml").read_text())
     assert data["project"]["version"] == __version__
+
+
+@pytest.mark.parametrize("value", ["0", "65536", "-1", "http", "80.5"])
+def test_serve_rejects_bad_ports_with_a_short_message(
+    value: str, capsys: pytest.CaptureFixture[str]
+) -> None:
+    with pytest.raises(SystemExit) as exit_info:
+        build_parser().parse_args(["serve", "--port", value])
+
+    err = capsys.readouterr().err
+    assert exit_info.value.code == 2
+    assert "--port" in err and "65535" in err
+    assert len(err) < 500

tests/test_rate_limit.py

@@ -2,21 +2,30 @@
 
 from __future__ import annotations
 
+import json
+
 import pytest
 
-from aegis_gitea_mcp.audit import validate_audit_log_integrity
+from aegis_gitea_mcp.audit import get_audit_logger, validate_audit_log_integrity
 from aegis_gitea_mcp.config import get_settings
-from aegis_gitea_mcp.rate_limit import (
-    RequestRateLimiter,
-    SlidingWindowLimiter,
-    get_rate_limiter,
-    reset_rate_limiter,
-)
+from aegis_gitea_mcp.rate_limit import RequestRateLimiter, SlidingWindowLimiter
+from tests.test_server_helpers import server_settings
 
 
-def test_sliding_window_blocks_after_limit(monkeypatch: pytest.MonkeyPatch) -> None:
+@pytest.fixture
+def clock(monkeypatch: pytest.MonkeyPatch) -> list[float]:
     now = [1000.0]
     monkeypatch.setattr("aegis_gitea_mcp.rate_limit.time.time", lambda: now[0])
+    return now
+
+
+def security_events() -> list[dict[str, object]]:
+    path = get_audit_logger().log_path
+    entries = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
+    return [e["payload"] for e in entries if e["event_type"] == "security_event"]
+
+
+def test_sliding_window_blocks_after_limit(clock: list[float]) -> None:
     limiter = SlidingWindowLimiter(max_requests=2, window_seconds=60)
 
     assert limiter.allow("k")
@@ -24,22 +33,56 @@ def test_sliding_window_blocks_after_limit(monkeypatch: pytest.MonkeyPatch) -> N
     assert not limiter.allow("k")
     assert limiter.allow("other")
 
-    now[0] += 61
+    clock[0] += 61
     assert limiter.allow("k")
 
 
-def test_request_limiter_per_ip_and_token(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setenv("GITEA_URL", "https://g.example.com")
-    monkeypatch.setenv("RATE_LIMIT_PER_MINUTE", "2")
-    limiter = RequestRateLimiter()
+def test_idle_keys_are_evicted(clock: list[float]) -> None:
+    limiter = SlidingWindowLimiter(max_requests=5, window_seconds=60)
+    for index in range(100):
+        limiter.allow(f"client-{index}")
+    assert len(limiter) == 100
+
+    clock[0] += 61
+    limiter.allow("fresh")
+
+    assert len(limiter) == 1
+
+
+def test_key_count_is_capped_and_least_recently_used_keys_go_first(clock: list[float]) -> None:
+    limiter = SlidingWindowLimiter(max_requests=1, window_seconds=60, max_keys=3)
+    for key in ("a", "b", "c"):
+        limiter.allow(key)
+    limiter.allow("a")  # touch: "b" is now the oldest
+    limiter.allow("d")
+
+    assert len(limiter) == 3
+    assert limiter.allow("b")  # evicted earlier, so it starts with a clean window
+    assert not limiter.allow("d")
+
+
+def test_first_excess_reports_once_per_window(clock: list[float]) -> None:
+    limiter = SlidingWindowLimiter(max_requests=1, window_seconds=60)
+    limiter.allow("k")
+
+    assert not limiter.allow("k") and limiter.first_excess("k")
+    assert not limiter.allow("k") and not limiter.first_excess("k")
+    clock[0] += 30
+    assert not limiter.allow("k") and not limiter.first_excess("k")
+    clock[0] += 31
+    assert limiter.allow("k")  # the old event left the window
+    assert not limiter.allow("k") and limiter.first_excess("k")
+
+
+def test_request_limiter_per_ip_and_token(mock_env: None) -> None:
+    limiter = RequestRateLimiter(server_settings(rate_limit_per_minute=2))
 
     assert limiter.check("1.1.1.1", "tok").allowed
     assert limiter.check("1.1.1.1", None).allowed
     assert not limiter.check("1.1.1.1", "tok").allowed
 
-    # A different IP with the same token trips the per-token limiter.
-    other = limiter.check("2.2.2.2", "tok")
-    assert other.allowed or other.reason == "Per-token rate limit exceeded"
+    # The same token from other IPs trips the per-token limiter.
+    assert limiter.check("2.2.2.2", "tok").allowed
     third = limiter.check("3.3.3.3", "tok")
     assert not third.allowed
     assert third.reason == "Per-token rate limit exceeded"
@@ -48,8 +91,30 @@ def test_request_limiter_per_ip_and_token(monkeypatch: pytest.MonkeyPatch) -> No
     assert valid, errors
 
 
-def test_global_limiter_singleton(mock_env: None) -> None:
-    first = get_rate_limiter()
-    assert get_rate_limiter() is first
-    reset_rate_limiter()
-    assert get_rate_limiter() is not first
+def test_only_the_first_excess_per_key_is_audited(mock_env: None, clock: list[float]) -> None:
+    limiter = RequestRateLimiter(server_settings(rate_limit_per_minute=1))
+    limiter.check("1.1.1.1", None)
+    for _ in range(20):
+        assert not limiter.check("1.1.1.1", None).allowed
+
+    events = security_events()
+    assert [e["event_type"] for e in events] == ["rate_limit_ip_exceeded"]
+
+    clock[0] += 61
+    limiter.check("1.1.1.1", None)
+    assert not limiter.check("1.1.1.1", None).allowed
+    assert len(security_events()) == 2
+
+
+def test_unwritable_audit_log_does_not_break_rate_limiting(
+    mock_env: None, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    def broken() -> None:
+        raise OSError("read-only file system")
+
+    monkeypatch.setattr("aegis_gitea_mcp.rate_limit.get_audit_logger", broken)
+    limiter = RequestRateLimiter(server_settings(rate_limit_per_minute=1))
+    limiter.check("1.1.1.1", None)
+
+    assert not limiter.check("1.1.1.1", None).allowed
+    assert "OSError" in caplog.text

tests/test_server.py

@@ -469,7 +469,7 @@ async def test_audit_event_per_call_never_contains_argument_values(
     assert validate_audit_log_integrity(log_path) == (True, [])
 
 
-async def test_audit_failure_does_not_break_the_call(
+async def test_audit_write_failure_after_execution_does_not_break_the_call(
     httpx_mock: HTTPXMock, caplog: pytest.LogCaptureFixture
 ) -> None:
     mock_user(httpx_mock)
@@ -482,11 +482,75 @@ async def test_audit_failure_does_not_break_the_call(
     async with connected(audit=BrokenAudit()) as client:  # type: ignore[arg-type]
         result = await client.call_tool("gitea_repo_get", REPO_ARGS)
 
+    # The call already ran (its side effect cannot be undone), so the result still goes out.
     assert not result.is_error
     assert "OSError" in caplog.text
     assert GITEA_TOKEN not in caplog.text
 
 
+async def test_unavailable_audit_log_refuses_the_call_before_gitea_is_contacted(
+    httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    def broken() -> None:
+        raise OSError("permission denied: " + GITEA_TOKEN)
+
+    monkeypatch.setattr("aegis_gitea_mcp.server.get_audit_logger", broken)
+    async with connected() as client:
+        result = await client.call_tool("gitea_repo_get", REPO_ARGS)
+        written = await client.call_tool("gitea_issue_create_issue", {**REPO_ARGS, "body": {}})
+
+    assert result.is_error and written.is_error
+    error = payload_of(result)["error"]
+    assert error["kind"] == "audit_unavailable"
+    assert GITEA_TOKEN not in json.dumps(payload_of(result)) + caplog.text
+    assert httpx_mock.get_requests() == []
+
+
+async def test_audit_correlation_id_is_the_request_id(httpx_mock: HTTPXMock) -> None:
+    mock_user(httpx_mock)
+    httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={})
+    async with connected() as client:
+        await client.call_tool("gitea_repo_get", REPO_ARGS)
+
+    from aegis_gitea_mcp.audit import get_audit_logger
+
+    line = get_audit_logger().log_path.read_text(encoding="utf-8").splitlines()[-1]
+    entry = json.loads(line)
+    assert entry["payload"]["correlation_id"] == entry["request_id"] != "-"
+
+
+async def test_rejected_raw_call_audits_no_unvalidated_path_or_sudo() -> None:
+    secret = "SECRET-IN-PATH-9f8e7d6c5b"
+    async with connected() as client:
+        bad_path = await client.call_tool(
+            "gitea_api_request", {"method": "GET", "path": secret, "sudo": secret}
+        )
+        bad_sudo = await client.call_tool(
+            "gitea_api_request", {"method": "GET", "path": "/user", "sudo": "bad name!"}
+        )
+
+    assert bad_path.is_error and bad_sudo.is_error
+    first, second = audit_entries()
+    assert first["outcome"] == second["outcome"] == "invalid_argument"
+    for entry in (first, second):
+        assert "path" not in entry and "sudo" not in entry
+    from aegis_gitea_mcp.audit import get_audit_logger
+
+    assert secret not in get_audit_logger().log_path.read_text(encoding="utf-8")
+
+
+async def test_raw_call_audits_the_validated_method_and_path(httpx_mock: HTTPXMock) -> None:
+    mock_user(httpx_mock)
+    httpx_mock.add_response(url=f"{API}/version", json={})
+    async with connected() as client:
+        await client.call_tool(
+            "gitea_api_request", {"method": "GET", "path": "/version", "sudo": "alice"}
+        )
+
+    entry = audit_entries()[-1]
+    assert (entry["method"], entry["path"], entry["sudo"]) == ("GET", "/version", "alice")
+
+
 # Internal errors
 
 

tests/test_transports.py

@@ -25,6 +25,7 @@ from aegis_gitea_mcp.gitea import RAW_TOOL_NAME
 from aegis_gitea_mcp.identity import Identity
 from aegis_gitea_mcp.transports import (
     GiteaProbe,
+    RateLimitMiddleware,
     build_http_app,
     transport_security,
 )
@@ -336,11 +337,7 @@ def test_transport_security_wildcard_bind_adds_no_bind_host() -> None:
 # Rate limiting
 
 
-def test_rate_limit_returns_429_on_the_mcp_path_only(
-    httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    # The limiter reads the process-wide settings.
-    monkeypatch.setenv("RATE_LIMIT_PER_MINUTE", "2")
+def test_rate_limit_returns_429_on_the_mcp_path_only(httpx_mock: HTTPXMock) -> None:
     httpx_mock.add_response(url=f"{API}/version", json={}, is_reusable=True)
     settings = http_settings(rate_limit_per_minute=2)
     with TestClient(build_http_app(settings), base_url=BASE) as c:
@@ -357,22 +354,32 @@ def test_rate_limit_returns_429_on_the_mcp_path_only(
     assert health.status_code == 200
 
 
-def test_rate_limit_is_keyed_per_token(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setenv("RATE_LIMIT_PER_MINUTE", "1")
+def test_rate_limit_is_keyed_per_token_hash() -> None:
     settings = http_settings(rate_limit_per_minute=1)
-    with TestClient(build_http_app(settings), base_url=BASE) as c:
+    app = build_http_app(settings)
+    with TestClient(app, base_url=BASE) as c:
         first = c.post("/mcp", json=INITIALIZE, headers={**MCP_HEADERS, **auth()})
+    limiter = next(m for m in app.user_middleware if m.cls is RateLimitMiddleware).kwargs["limiter"]
 
     assert first.status_code == 200
-    from aegis_gitea_mcp.rate_limit import get_rate_limiter
-
-    limiter = get_rate_limiter()
     assert limiter._token_limiter.max_requests == 1
     # The raw header is never used as a key, only its hash.
     keys = list(limiter._token_limiter._events)
     assert keys and all(BEARER_TOKEN not in key for key in keys)
 
 
+def test_rate_limit_uses_the_settings_passed_to_the_app(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("RATE_LIMIT_PER_MINUTE", "500")  # the process-wide value must not matter
+    app = build_http_app(http_settings(rate_limit_per_minute=1))
+    with TestClient(app, base_url=BASE) as c:
+        statuses = [
+            c.post("/mcp", json=INITIALIZE, headers={**MCP_HEADERS, **auth()}).status_code
+            for _ in range(2)
+        ]
+
+    assert statuses == [200, 429]
+
+
 # Auth wiring
 
 

tests/test_transports_startup.py

@@ -95,14 +95,107 @@ def test_loopback_without_auth_runs(uvicorn_spy: UvicornSpy) -> None:
     assert uvicorn_spy.calls[0][1]["log_level"] == "debug"
 
 
-def test_forwarded_allow_ips_setting_is_honoured_when_present(
-    uvicorn_spy: UvicornSpy, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    monkeypatch.setattr(Settings, "forwarded_allow_ips", "10.0.0.0/8", raising=False)
+def test_forwarded_allow_ips_setting_is_passed_to_uvicorn(uvicorn_spy: UvicornSpy) -> None:
+    assert run_http(server_settings(forwarded_allow_ips="10.0.0.0/8, 172.18.0.2")) == 0
 
+    assert uvicorn_spy.calls[0][1]["forwarded_allow_ips"] == "10.0.0.0/8,172.18.0.2"
+
+
+def test_forwarded_allow_ips_wildcard_needs_an_explicit_value_and_warns(
+    uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str]
+) -> None:
+    assert server_settings().forwarded_allow_ips == "127.0.0.1"
+
+    assert run_http(server_settings(forwarded_allow_ips="*")) == 0
+
+    assert uvicorn_spy.calls[0][1]["forwarded_allow_ips"] == "*"
+    assert "FORWARDED_ALLOW_IPS=*" in capsys.readouterr().err
+
+
+def test_non_loopback_bind_without_public_host_warns_which_variables_to_set(
+    uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str]
+) -> None:
+    assert run_http(server_settings(mcp_host=ANY_HOST, mcp_bearer_token=BEARER_TOKEN)) == 0
+
+    err = capsys.readouterr().err
+    assert "PUBLIC_BASE_URL" in err and "ALLOWED_HOSTS" in err and "421" in err
+
+
+@pytest.mark.parametrize(
+    "extra",
+    [{"public_base_url": "https://mcp.example.com"}, {"allowed_hosts": ["mcp.example.com"]}],
+)
+def test_no_host_warning_when_a_public_host_is_configured(
+    uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str], extra: dict[str, Any]
+) -> None:
+    settings = server_settings(mcp_host=ANY_HOST, mcp_bearer_token=BEARER_TOKEN, **extra)
+
+    assert run_http(settings) == 0
+
+    assert "421" not in capsys.readouterr().err
+
+
+def test_no_host_warning_on_loopback(
+    uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str]
+) -> None:
     assert run_http(server_settings()) == 0
 
-    assert uvicorn_spy.calls[0][1]["forwarded_allow_ips"] == "10.0.0.0/8"
+    assert capsys.readouterr().err == ""
+
+
+def test_oauth_enabled_with_a_bearer_warns_that_oauth_is_unavailable(
+    uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str]
+) -> None:
+    assert run_http(oauth_settings(mcp_bearer_token=BEARER_TOKEN)) == 0
+
+    assert "per-user OAuth is not available" in capsys.readouterr().err
+
+
+@pytest.mark.parametrize("value", ["*,10.0.0.1", "not-an-ip", "10.0.0.1/99"])
+def test_invalid_forwarded_allow_ips_is_rejected(value: str) -> None:
+    with pytest.raises(ValueError, match="FORWARDED_ALLOW_IPS"):
+        server_settings(forwarded_allow_ips=value)
+
+
+def test_blank_forwarded_allow_ips_means_the_default() -> None:
+    assert server_settings(forwarded_allow_ips="  ").forwarded_allow_ips == "127.0.0.1"
+
+
+# Audit log at startup
+
+
+@pytest.fixture
+def unwritable_audit(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    """An audit path whose parent is a regular file, so it can never be opened."""
+    blocker = tmp_path / "blocker"
+    blocker.write_text("x", encoding="utf-8")
+    path = blocker / "audit.log"
+    monkeypatch.setenv("AUDIT_LOG_PATH", str(path))
+    return path
+
+
+def test_http_exits_2_when_the_audit_log_cannot_be_opened(
+    uvicorn_spy: UvicornSpy, unwritable_audit: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    assert run_http(server_settings(audit_log_path=unwritable_audit)) == 2
+
+    captured = capsys.readouterr()
+    assert uvicorn_spy.calls == []
+    assert "audit log" in captured.err and "AUDIT_LOG_PATH" in captured.err
+    assert str(unwritable_audit) in captured.err
+    assert captured.out == ""
+
+
+def test_stdio_exits_2_when_the_audit_log_cannot_be_opened(
+    unwritable_audit: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    monkeypatch.setattr(transports.anyio, "run", lambda *a: pytest.fail("must not serve"))
+
+    assert run_stdio(server_settings(audit_log_path=unwritable_audit)) == 2
+
+    captured = capsys.readouterr()
+    assert "AUDIT_LOG_PATH" in captured.err
+    assert captured.out == ""
 
 
 def test_oauth_misconfiguration_lists_every_missing_setting(