diff --git a/src/memos/api/middleware/auth.py b/src/memos/api/middleware/auth.py index 15b217651..125d17278 100644 --- a/src/memos/api/middleware/auth.py +++ b/src/memos/api/middleware/auth.py @@ -6,6 +6,7 @@ """ import hashlib +import hmac import os import time @@ -25,6 +26,7 @@ # Environment configuration AUTH_ENABLED = os.getenv("AUTH_ENABLED", "false").lower() == "true" MASTER_KEY_HASH = os.getenv("MASTER_KEY_HASH") # SHA-256 hash of master key +INTERNAL_SERVICE_SECRET = os.getenv("INTERNAL_SERVICE_SECRET") # shared secret for X-Internal-Service header INTERNAL_SERVICE_IPS = {"127.0.0.1", "::1", "memos-mcp", "moltbot", "clawdbot"} # Connection pool for auth queries (lazy init) @@ -142,16 +144,35 @@ async def lookup_api_key(key_hash: str) -> dict[str, Any] | None: def is_internal_request(request: Request) -> bool: - """Check if request is from internal service.""" + """Check if request is from internal service. + + Two authorised paths: + + 1. The client host address is a well-known trusted host in + ``INTERNAL_SERVICE_IPS``. + 2. The request carries an ``X-Internal-Service`` header whose value matches + the operator-configured ``INTERNAL_SERVICE_SECRET`` env var. Comparison + is constant-time (``hmac.compare_digest``) to avoid timing side channels. + + The header branch is **disabled** whenever either the environment secret or + the request header is missing or empty. This closes GHSA-9pw6-vmgx-qgwx / + issue #2259, where an unset secret combined with an absent header caused + ``None == None`` to be treated as a match and grant the ``internal`` + principal to unauthenticated remote callers. + """ client_host = request.client.host if request.client else None # Check internal IPs if client_host in INTERNAL_SERVICE_IPS: return True - # Check internal header (for container-to-container) + # Check internal header (for container-to-container). Treat an unset / + # empty secret or missing / empty header as "disabled" and fail closed. + secret = INTERNAL_SERVICE_SECRET internal_header = request.headers.get("X-Internal-Service") - return internal_header == os.getenv("INTERNAL_SERVICE_SECRET") + if not secret or not internal_header: + return False + return hmac.compare_digest(internal_header, secret) async def verify_api_key( @@ -180,7 +201,8 @@ async def verify_api_key( # Allow internal services if is_internal_request(request): - logger.debug(f"Internal request from {request.client.host}") + client_host = request.client.host if request.client else "" + logger.debug(f"Internal request from {client_host}") return { "user_name": "internal", "scopes": ["all"], diff --git a/tests/api/test_auth_internal_request.py b/tests/api/test_auth_internal_request.py new file mode 100644 index 000000000..9a38b3820 --- /dev/null +++ b/tests/api/test_auth_internal_request.py @@ -0,0 +1,195 @@ +"""Tests for `memos.api.middleware.auth.is_internal_request` fail-open regression. + +Regression test suite for GHSA-9pw6-vmgx-qgwx / issue #2259. + +The bug: when `INTERNAL_SERVICE_SECRET` is unset (its default across every shipped +Dockerfile / compose / Helm config) and an external client does not send the +`X-Internal-Service` header, `is_internal_request()` compared +``request.headers.get("X-Internal-Service")`` to ``os.getenv("INTERNAL_SERVICE_SECRET")`` +directly, so ``None == None`` returned True and the request was granted the +`internal` principal with ``scopes: ["all"]``. + +These tests lock in the fixed behaviour: unset secret OR missing header → not internal; +matching non-empty secret + header → internal (constant-time compare). +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from fastapi import HTTPException +from starlette.requests import Request + +from memos.api.middleware import auth as auth_module +from memos.api.middleware.auth import is_internal_request, verify_api_key + + +def _make_request( + *, + client: tuple[str, int] | None = ("203.0.113.9", 53124), + headers: dict[str, str] | None = None, +) -> Request: + """Build a bare-bones Starlette Request for the dependency under test.""" + scope: dict[str, Any] = { + "type": "http", + "method": "GET", + "path": "/admin/keys", + "headers": [(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()], + } + if client is not None: + scope["client"] = client + return Request(scope) + + +class TestIsInternalRequest: + def test_returns_false_when_secret_unset_and_header_missing( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Regression: None == None must not authenticate.""" + monkeypatch.setattr(auth_module, "INTERNAL_SERVICE_SECRET", None) + request = _make_request(headers={}) + + assert is_internal_request(request) is False + + def test_returns_false_when_secret_unset_and_header_present( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Header path is disabled when secret is not configured.""" + monkeypatch.setattr(auth_module, "INTERNAL_SERVICE_SECRET", None) + request = _make_request(headers={"X-Internal-Service": "guessed-value"}) + + assert is_internal_request(request) is False + + def test_returns_false_when_secret_empty_string(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Empty string secret is treated as unset (defence in depth).""" + monkeypatch.setattr(auth_module, "INTERNAL_SERVICE_SECRET", "") + request = _make_request(headers={"X-Internal-Service": ""}) + + assert is_internal_request(request) is False + + def test_returns_false_when_secret_set_but_header_missing( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(auth_module, "INTERNAL_SERVICE_SECRET", "super-secret-value") + request = _make_request(headers={}) + + assert is_internal_request(request) is False + + def test_returns_false_when_secret_set_but_header_wrong( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(auth_module, "INTERNAL_SERVICE_SECRET", "super-secret-value") + request = _make_request(headers={"X-Internal-Service": "wrong-value"}) + + assert is_internal_request(request) is False + + def test_returns_true_when_secret_matches_header(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(auth_module, "INTERNAL_SERVICE_SECRET", "super-secret-value") + request = _make_request(headers={"X-Internal-Service": "super-secret-value"}) + + assert is_internal_request(request) is True + + def test_returns_true_when_client_ip_is_internal(self, monkeypatch: pytest.MonkeyPatch) -> None: + """The internal-IP branch remains untouched by the fix.""" + monkeypatch.setattr(auth_module, "INTERNAL_SERVICE_SECRET", None) + request = _make_request(client=("127.0.0.1", 12345), headers={}) + + assert is_internal_request(request) is True + + def test_uses_constant_time_compare(self, monkeypatch: pytest.MonkeyPatch) -> None: + """The fix must call `hmac.compare_digest` (not plain ==) to avoid timing leaks.""" + monkeypatch.setattr(auth_module, "INTERNAL_SERVICE_SECRET", "super-secret-value") + calls: list[tuple[str, str]] = [] + + real_compare = auth_module.hmac.compare_digest + + def spy(a: str, b: str) -> bool: + calls.append((a, b)) + return real_compare(a, b) + + monkeypatch.setattr(auth_module.hmac, "compare_digest", spy) + request = _make_request(headers={"X-Internal-Service": "super-secret-value"}) + + assert is_internal_request(request) is True + assert calls, "expected hmac.compare_digest to be invoked for header comparison" + + def test_no_client_and_no_header_and_no_secret(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Missing request.client + no header + no secret must NOT authenticate.""" + monkeypatch.setattr(auth_module, "INTERNAL_SERVICE_SECRET", None) + request = _make_request(client=None, headers={}) + + assert is_internal_request(request) is False + + +class TestVerifyApiKeyEndToEnd: + """End-to-end reproduction of the advisory PoC against `verify_api_key`.""" + + @pytest.mark.asyncio + async def test_external_request_no_key_is_rejected_when_secret_unset( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The exact PoC from GHSA-9pw6-vmgx-qgwx must now be rejected with 401.""" + monkeypatch.setattr(auth_module, "AUTH_ENABLED", True) + monkeypatch.setattr(auth_module, "INTERNAL_SERVICE_SECRET", None) + request = _make_request(headers={}) + + with pytest.raises(HTTPException) as exc_info: + await verify_api_key(request, api_key=None) + + assert exc_info.value.status_code == 401 + assert "Missing API key" in exc_info.value.detail + + @pytest.mark.asyncio + async def test_external_request_with_wrong_header_still_rejected( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(auth_module, "AUTH_ENABLED", True) + monkeypatch.setattr(auth_module, "INTERNAL_SERVICE_SECRET", None) + request = _make_request(headers={"X-Internal-Service": "guess"}) + + with pytest.raises(HTTPException) as exc_info: + await verify_api_key(request, api_key=None) + + assert exc_info.value.status_code == 401 + + @pytest.mark.asyncio + async def test_internal_request_with_matching_secret_is_authorised( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(auth_module, "AUTH_ENABLED", True) + monkeypatch.setattr(auth_module, "INTERNAL_SERVICE_SECRET", "shared-secret") + request = _make_request( + headers={"X-Internal-Service": "shared-secret"}, + client=("10.0.0.42", 40000), + ) + + result = await verify_api_key(request, api_key=None) + assert result == { + "user_name": "internal", + "scopes": ["all"], + "is_master_key": False, + "is_internal": True, + } + + @pytest.mark.asyncio + async def test_internal_request_via_header_with_no_client_does_not_crash( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Regression for finding #1: ``request.client`` may be None on the header path. + + `verify_api_key` must not raise ``AttributeError`` when a request without a + ``client`` tuple is granted the internal principal via a matching + ``X-Internal-Service`` header. + """ + monkeypatch.setattr(auth_module, "AUTH_ENABLED", True) + monkeypatch.setattr(auth_module, "INTERNAL_SERVICE_SECRET", "shared-secret") + request = _make_request( + client=None, + headers={"X-Internal-Service": "shared-secret"}, + ) + + result = await verify_api_key(request, api_key=None) + assert result["is_internal"] is True + assert result["user_name"] == "internal"