[security] fix(mcp): harden signed upload URL authority

openviking/server/mcp_endpoint.py

@@ -18,6 +18,7 @@
 
 import asyncio
 import contextvars
+import ipaddress
 import os
 from contextlib import asynccontextmanager
 from datetime import datetime, timezone
@@ -378,27 +379,75 @@ async def remember(messages: list[StoreMessage]) -> str:
 _DEFAULT_UPLOAD_TTL_SECONDS = 600
 
 
+def _first_header_value(value: Optional[str]) -> Optional[str]:
+    """Return the first value from a potentially comma-separated HTTP header."""
+    if not value:
+        return None
+    head = value.split(",", 1)[0].strip()
+    return head or None
+
+
+def _is_loopback_authority(authority: str) -> bool:
+    """Return True for syntactically safe localhost/loopback Host authorities.
+
+    ``authority`` may include a numeric port (``localhost:1933``) or a bracketed
+    IPv6 literal (``[::1]:1933``). Non-loopback request authorities are not
+    trusted for signed upload URLs because they can be supplied by the client.
+    """
+    host = authority.strip()
+    if not host:
+        return False
+    # Reject characters that can change URL authority interpretation when the
+    # value is interpolated into ``http://{host}`` (userinfo, paths, fragments,
+    # queries, whitespace/control chars, or scheme-relative forms).
+    if any(ch in host for ch in "@/?#\\") or any(ch.isspace() for ch in host):
+        return False
+
+    if host.startswith("["):
+        end = host.find("]")
+        if end == -1:
+            return False
+        addr = host[1:end]
+        rest = host[end + 1 :]
+        if rest:
+            if not rest.startswith(":") or not rest[1:].isdigit():
+                return False
+        host = addr
+    elif ":" in host:
+        # Host headers with a single colon are host:port. Multiple colons without
+        # brackets are IPv6 literals; leave them intact for ipaddress parsing.
+        if host.count(":") == 1:
+            name, port = host.rsplit(":", 1)
+            if not port.isdigit():
+                return False
+            host = name
+
+    host = host.rstrip(".").lower()
+    if host == "localhost" or host.endswith(".localhost"):
+        return True
+    try:
+        return ipaddress.ip_address(host).is_loopback
+    except ValueError:
+        return False
+
+
 def _resolve_public_base_url() -> tuple[str, str]:
     """Pick the URL the agent should POST uploads to. Returns ``(base_url, source)``.
 
     Resolution order (first match wins):
-
     1. ``env`` — ``OPENVIKING_PUBLIC_BASE_URL`` environment variable. Operator-set,
        always wins.
     2. ``config`` — ``ServerConfig.public_base_url``. Operator-set baseline in ov.conf.
-    3. ``forwarded`` — ``X-Forwarded-Host`` (+ ``X-Forwarded-Proto``) from the request.
-       Set by reverse proxies (nginx, ALB, ingress controllers, MCP proxies). Reliable
-       when the proxy chain forwards these headers, which is the standard default.
-    4. ``host`` — the raw ``Host`` header from a direct connection. Reliable for
-       same-host MCP clients (e.g. local Claude Code talking to localhost server).
-    5. ``listen`` — ``http://{listen_host}:{listen_port}`` last-resort fallback.
-       Only produces an agent-reachable URL when the server is bound to a routable
-       address; commonly wrong behind reverse proxies.
-
-    Sources 1 and 2 are "explicit" — operator vouched for the URL. Sources 3-5 are
-    inferred and may be wrong when the proxy chain doesn't forward request headers.
-    Callers should append a "set OPENVIKING_PUBLIC_BASE_URL if upload fails" hint
-    in that case.
+    3. ``host`` — a raw ``Host`` header only when it names localhost/loopback.
+       This preserves direct local MCP clients without letting arbitrary request
+       authorities control signed upload destinations.
+    4. ``listen`` — ``http://{listen_host}:{listen_port}`` last-resort fallback.
+       Reverse-proxy deployments that need an external upload URL should set
+       ``OPENVIKING_PUBLIC_BASE_URL`` or ``server.public_base_url`` explicitly.
+
+    Sources 1 and 2 are "explicit" — operator vouched for the URL. Sources 3-4 are
+    inferred and may be wrong behind reverse proxies. Callers should append a
+    "set OPENVIKING_PUBLIC_BASE_URL if upload fails" hint in that case.
     """
     env_url = os.environ.get("OPENVIKING_PUBLIC_BASE_URL")
     if env_url:
@@ -409,22 +458,12 @@ def _resolve_public_base_url() -> tuple[str, str]:
 
     url_info = _request_url_ctx.get()
     if url_info:
-        # X-Forwarded-Host / -Proto can be comma-separated lists when the request
-        # crosses multiple proxy hops. Take the first (left-most original-client)
-        # value, matching the normalization in openviking.server.oauth.router.
-        def _first(value: Optional[str]) -> Optional[str]:
-            if not value:
-                return None
-            head = value.split(",", 1)[0].strip()
-            return head or None
-
-        xfh = _first(url_info.get("x_forwarded_host"))
-        xfp = _first(url_info.get("x_forwarded_proto"))
-        host_hdr = _first(url_info.get("host"))
-        if xfh:
-            proto = xfp or "https"
-            return f"{proto}://{xfh}", "forwarded"
-        if host_hdr:
+        # Do not infer signed upload destinations from X-Forwarded-* by default:
+        # untrusted clients can often supply these headers unless a deployment
+        # explicitly normalizes them. Operators behind proxies should configure a
+        # public base URL instead.
+        host_hdr = _first_header_value(url_info.get("host"))
+        if host_hdr and _is_loopback_authority(host_hdr):
             return f"http://{host_hdr}", "host"
 
     if config is not None:

tests/server/test_mcp_endpoint.py

@@ -361,7 +361,7 @@ async def test_add_resource_local_path_uses_config_when_env_unset(service, monke
     upload_token_store.clear()
 
 
-async def test_add_resource_local_path_infers_from_x_forwarded_headers(service, monkeypatch):
+async def test_add_resource_local_path_ignores_untrusted_forwarded_headers(service, monkeypatch):
     from openviking.server.mcp_endpoint import _request_url_ctx
     from openviking.server.upload_token_store import upload_token_store
 
@@ -371,7 +371,7 @@ async def test_add_resource_local_path_infers_from_x_forwarded_headers(service,
     token = _request_url_ctx.set(
         {
             "x_forwarded_proto": "https",
-            "x_forwarded_host": "ov.public.example.com",
+            "x_forwarded_host": "attacker.example.com",
             "host": "internal:1933",
         }
     )
@@ -380,12 +380,63 @@ async def test_add_resource_local_path_infers_from_x_forwarded_headers(service,
     finally:
         _request_url_ctx.reset(token)
 
-    assert "https://ov.public.example.com/api/v1/resources/temp_upload_signed" in result
+    assert "attacker.example.com" not in result
+    assert "http://127.0.0.1:1933/api/v1/resources/temp_upload_signed" in result
     # Inferred → hint must appear
     assert "OPENVIKING_PUBLIC_BASE_URL" in result
     upload_token_store.clear()
 
 
+async def test_add_resource_local_path_ignores_untrusted_host_header(service, monkeypatch):
+    from openviking.server.mcp_endpoint import _request_url_ctx
+    from openviking.server.upload_token_store import upload_token_store
+
+    upload_token_store.clear()
+    monkeypatch.delenv("OPENVIKING_PUBLIC_BASE_URL", raising=False)
+
+    token = _request_url_ctx.set(
+        {
+            "x_forwarded_proto": None,
+            "x_forwarded_host": None,
+            "host": "attacker.example.com",
+        }
+    )
+    try:
+        result = await add_resource(path="/tmp/x.pdf")
+    finally:
+        _request_url_ctx.reset(token)
+
+    assert "attacker.example.com" not in result
+    assert "http://127.0.0.1:1933/api/v1/resources/temp_upload_signed" in result
+    assert "OPENVIKING_PUBLIC_BASE_URL" in result
+    upload_token_store.clear()
+
+
+async def test_add_resource_local_path_allows_loopback_host_for_local_clients(service, monkeypatch):
+    from openviking.server.mcp_endpoint import _request_url_ctx
+    from openviking.server.upload_token_store import upload_token_store
+
+    upload_token_store.clear()
+    monkeypatch.delenv("OPENVIKING_PUBLIC_BASE_URL", raising=False)
+
+    token = _request_url_ctx.set(
+        {
+            "x_forwarded_proto": "https",
+            "x_forwarded_host": "attacker.example.com",
+            "host": "localhost:1933",
+        }
+    )
+    try:
+        result = await add_resource(path="/tmp/x.pdf")
+    finally:
+        _request_url_ctx.reset(token)
+
+    assert "attacker.example.com" not in result
+    assert "http://localhost:1933/api/v1/resources/temp_upload_signed" in result
+    assert "OPENVIKING_PUBLIC_BASE_URL" in result
+    upload_token_store.clear()
+
+
 async def test_add_resource_temp_file_id_lookalike_in_path_is_rejected(service):
     result = await add_resource(path="upload_abc123.pdf")
     assert "looks like a temp_file_id" in result.lower()

tests/unit/test_mcp_public_base_url_trust.py

@@ -0,0 +1,107 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
+# SPDX-License-Identifier: AGPL-3.0
+"""Regression tests for MCP upload URL host trust."""
+
+from __future__ import annotations
+
+from openviking.server import dependencies
+from openviking.server.config import ServerConfig
+from openviking.server.mcp_endpoint import (
+    _is_loopback_authority,
+    _request_url_ctx,
+    _resolve_public_base_url,
+)
+
+
+def _set_config(monkeypatch, **kwargs) -> ServerConfig:
+    config = ServerConfig(**kwargs)
+    monkeypatch.setattr(dependencies, "_server_config", config)
+    return config
+
+
+def test_mcp_upload_url_does_not_trust_forwarded_host_without_public_base_url(monkeypatch):
+    monkeypatch.delenv("OPENVIKING_PUBLIC_BASE_URL", raising=False)
+    _set_config(monkeypatch, host="127.0.0.1", port=1933, public_base_url=None)
+    token = _request_url_ctx.set(
+        {
+            "x_forwarded_proto": "https",
+            "x_forwarded_host": "attacker.example",
+            "host": "openviking.example",
+        }
+    )
+    try:
+        base_url, source = _resolve_public_base_url()
+    finally:
+        _request_url_ctx.reset(token)
+
+    assert base_url == "http://127.0.0.1:1933"
+    assert source == "listen"
+
+
+def test_mcp_upload_url_does_not_trust_external_host_header_without_public_base_url(monkeypatch):
+    monkeypatch.delenv("OPENVIKING_PUBLIC_BASE_URL", raising=False)
+    _set_config(monkeypatch, host="0.0.0.0", port=1933, public_base_url=None)
+    token = _request_url_ctx.set(
+        {
+            "x_forwarded_proto": None,
+            "x_forwarded_host": None,
+            "host": "attacker.example",
+        }
+    )
+    try:
+        base_url, source = _resolve_public_base_url()
+    finally:
+        _request_url_ctx.reset(token)
+
+    assert base_url == "http://0.0.0.0:1933"
+    assert source == "listen"
+
+
+def test_mcp_upload_url_keeps_explicit_public_base_url_authoritative(monkeypatch):
+    monkeypatch.delenv("OPENVIKING_PUBLIC_BASE_URL", raising=False)
+    _set_config(monkeypatch, public_base_url="https://openviking.example")
+    token = _request_url_ctx.set(
+        {
+            "x_forwarded_proto": "https",
+            "x_forwarded_host": "attacker.example",
+            "host": "attacker.example",
+        }
+    )
+    try:
+        base_url, source = _resolve_public_base_url()
+    finally:
+        _request_url_ctx.reset(token)
+
+    assert base_url == "https://openviking.example"
+    assert source == "config"
+
+
+def test_mcp_upload_url_allows_loopback_host_for_local_clients(monkeypatch):
+    monkeypatch.delenv("OPENVIKING_PUBLIC_BASE_URL", raising=False)
+    _set_config(monkeypatch, host="127.0.0.1", port=1933, public_base_url=None)
+    token = _request_url_ctx.set(
+        {
+            "x_forwarded_proto": None,
+            "x_forwarded_host": None,
+            "host": "localhost:1933",
+        }
+    )
+    try:
+        base_url, source = _resolve_public_base_url()
+    finally:
+        _request_url_ctx.reset(token)
+
+    assert base_url == "http://localhost:1933"
+    assert source == "host"
+
+
+def test_loopback_authority_rejects_url_authority_confusion():
+    assert _is_loopback_authority("localhost:1933")
+    assert _is_loopback_authority("127.0.0.1:1933")
+    assert _is_loopback_authority("[::1]:1933")
+    assert not _is_loopback_authority("localhost:1933@attacker.example")
+    assert not _is_loopback_authority("localhost/path")
+    assert not _is_loopback_authority("localhost?next=attacker")
+    assert not _is_loopback_authority("localhost#attacker")
+    assert not _is_loopback_authority("localhost:not-a-port")
+    assert not _is_loopback_authority("[::1]extra")