From 7731a50da005e8dc20e2aa58248ab777b9e932c6 Mon Sep 17 00:00:00 2001 From: Aadarsh Mahesh K Date: Thu, 22 Oct 2026 23:49:21 +0530 Subject: [PATCH] fix(auth): deny CIMD host mismatch, require pinned IP for outbound peer, remove duplicate helpers --- nitrostack/auth/cimd.py | 31 +-------------- nitrostack/auth/oauth_module.py | 26 +++++++++---- nitrostack/auth/request.py | 11 +++--- tests/test_mcp20_oauth_cimd.py | 68 ++++++++++++++++++++++++++++----- 4 files changed, 84 insertions(+), 52 deletions(-) diff --git a/nitrostack/auth/cimd.py b/nitrostack/auth/cimd.py index f9aa7a3..5ff4366 100644 --- a/nitrostack/auth/cimd.py +++ b/nitrostack/auth/cimd.py @@ -125,16 +125,10 @@ def _pick_pinned_ip(resolved_ips: set[str]) -> str: def cimd_peer_is_acceptable(peer: str, pinned_ip: str) -> bool: - """Require the direct socket peer to match the pinned IP. - - A configured trusted proxy hop is the only exception; untrusted peers - cannot stand in for the destination. - """ + """Require the direct socket peer to match the pinned IP.""" if peer == pinned_ip: return not is_blocked_ip(peer) - from nitrostack.transports.proxy import peer_is_trusted - - return peer_is_trusted(peer) + return False def request_host_for_cimd( @@ -369,24 +363,3 @@ def resolve_cimd_sync( resolve_cimd(client_id_url, allow_loopback=allow_loopback, timeout_sec=timeout_sec) ) raise CimdFetchError("CIMD resolution cannot run nested on a running event loop") - - -def looks_like_cimd_url(value: Any) -> bool: - """Return True when ``value`` is an http(s) client identifier URL.""" - return isinstance(value, str) and value.startswith(("https://", "http://")) - - -def resolve_cimd_sync( - client_id_url: str, - *, - allow_loopback: bool = False, - timeout_sec: float = CIMD_FETCH_TIMEOUT_SEC, -) -> dict[str, Any]: - """Synchronous wrapper for registration / other non-async callers.""" - try: - asyncio.get_running_loop() - except RuntimeError: - return asyncio.run( - resolve_cimd(client_id_url, allow_loopback=allow_loopback, timeout_sec=timeout_sec) - ) - raise CimdFetchError("CIMD resolution cannot run nested on a running event loop") diff --git a/nitrostack/auth/oauth_module.py b/nitrostack/auth/oauth_module.py index b2585c4..bfc59d1 100644 --- a/nitrostack/auth/oauth_module.py +++ b/nitrostack/auth/oauth_module.py @@ -112,22 +112,32 @@ def apply_cimd_to_registration_body( honored only when the peer is trusted, so an untrusted ``X-Forwarded-Host`` cannot rebind the CIMD host comparison. """ - from nitrostack.auth.cimd import cimd_host_matches_request, request_host_for_cimd + from nitrostack.auth.cimd import ( + CimdValidationError, + cimd_host_matches_request, + request_host_for_cimd, + ) payload = dict(body or {}) client_id = payload.get("client_id") if not looks_like_cimd_url(client_id): return payload document = resolve_cimd_sync(str(client_id)) - payload["client_id"] = document["client_id"] - payload["_cimd"] = document - if headers is not None: - payload["_cimd_request_host"] = request_host_for_cimd(headers, peer) - payload["_cimd_host_matches_request"] = cimd_host_matches_request( + if headers is not None or peer is not None: + req_headers = headers if headers is not None else {} + if not cimd_host_matches_request( str(document["client_id"]), - headers, + req_headers, peer, - ) + ): + req_host = request_host_for_cimd(req_headers, peer) + raise CimdValidationError( + f"CIMD client_id URL host does not match request host: {req_host!r}" + ) + payload["_cimd_request_host"] = request_host_for_cimd(req_headers, peer) + payload["_cimd_host_matches_request"] = True + payload["client_id"] = document["client_id"] + payload["_cimd"] = document return payload diff --git a/nitrostack/auth/request.py b/nitrostack/auth/request.py index b3c22c4..6804951 100644 --- a/nitrostack/auth/request.py +++ b/nitrostack/auth/request.py @@ -108,12 +108,13 @@ def authorization_token_from_request(rc: Any) -> Optional[str]: def verify_bearer_payload(token: str) -> Optional[dict[str, Any]]: - try: - from nitrostack.auth.jwt import JWTService - from nitrostack.core.di import DIContainer + from nitrostack.auth.jwt import JWTService + from nitrostack.core.di import DIContainer - payload = DIContainer.get_instance().resolve(JWTService).verify_token(token) - except Exception: + service = DIContainer.get_instance().resolve(JWTService) + try: + payload = service.verify_token(token) + except ValueError: return None return payload if isinstance(payload, dict) else None diff --git a/tests/test_mcp20_oauth_cimd.py b/tests/test_mcp20_oauth_cimd.py index b269faa..fc72ce0 100644 --- a/tests/test_mcp20_oauth_cimd.py +++ b/tests/test_mcp20_oauth_cimd.py @@ -176,9 +176,10 @@ def test_direct_peer_must_match_pin(self, monkeypatch): assert cimd_peer_is_acceptable("8.8.8.8", "8.8.8.8") is True assert cimd_peer_is_acceptable("1.2.3.4", "8.8.8.8") is False - def test_trusted_proxy_peer_is_acceptable(self, monkeypatch): + def test_trusted_proxy_peer_is_not_acceptable_for_outbound(self, monkeypatch): monkeypatch.setenv("TRUSTED_PROXIES", "10.0.0.5") - assert cimd_peer_is_acceptable("10.0.0.5", "8.8.8.8") is True + # Trusted proxies apply to inbound headers, NOT outbound socket fetch pins. + assert cimd_peer_is_acceptable("10.0.0.5", "8.8.8.8") is False assert cimd_peer_is_acceptable("10.0.0.9", "8.8.8.8") is False def test_untrusted_forwarded_host_cannot_rebind_cimd(self, monkeypatch): @@ -191,7 +192,7 @@ def test_untrusted_forwarded_host_cannot_rebind_cimd(self, monkeypatch): assert request_host_for_cimd(headers, peer="8.8.8.8") == "mcp.nitrostack.io" assert cimd_host_matches_request(url, headers, peer="8.8.8.8") is False - def test_apply_cimd_records_untrusted_request_host(self, monkeypatch): + def test_apply_cimd_denies_untrusted_request_host_mismatch(self, monkeypatch): monkeypatch.delenv("TRUSTED_PROXIES", raising=False) url = "https://app.nitrostack.io/oauth/client-metadata.json" headers = { @@ -202,13 +203,13 @@ def test_apply_cimd_records_untrusted_request_host(self, monkeypatch): "nitrostack.auth.oauth_module.resolve_cimd_sync", return_value={"client_id": url, "client_name": "Studio"}, ): - body = apply_cimd_to_registration_body( - {"client_id": url}, - headers=headers, - peer="8.8.8.8", - ) - assert body["_cimd_request_host"] == "mcp.nitrostack.io" - assert body["_cimd_host_matches_request"] is False + with pytest.raises(CimdValidationError) as exc_info: + apply_cimd_to_registration_body( + {"client_id": url}, + headers=headers, + peer="8.8.8.8", + ) + assert "host" in str(exc_info.value).lower() def test_apply_cimd_honors_trusted_forwarded_host(self, monkeypatch): monkeypatch.setenv("TRUSTED_PROXIES", "10.0.0.5") @@ -229,6 +230,12 @@ def test_apply_cimd_honors_trusted_forwarded_host(self, monkeypatch): assert body["_cimd_request_host"] == "app.nitrostack.io" assert body["_cimd_host_matches_request"] is True + def test_cimd_helpers_defined_once(self): + from pathlib import Path + source = Path("nitrostack/auth/cimd.py").read_text(encoding="utf-8") + assert source.count("def looks_like_cimd_url") == 1 + assert source.count("def resolve_cimd_sync") == 1 + class TestCimdRegistrationWiring: def test_applies_cimd_document_to_registration_body(self): @@ -266,3 +273,44 @@ def test_rejects_missing_iss(self): def test_rejects_mismatched_iss(self): with pytest.raises(AuthorizationIssuerMismatchError, match="mismatch"): validate_authorization_iss("https://evil.example", "https://auth.nitrostack.io") + + +class TestVerifyBearerPayload: + def test_returns_payload_for_valid_token(self): + from nitrostack.auth.jwt import JWTService + from nitrostack.auth.request import verify_bearer_payload + from nitrostack.core.di import DIContainer + + DIContainer.reset() + service = JWTService() + DIContainer.get_instance().register_value(JWTService, service) + token = service.create_token({"sub": "user_123", "tenant_id": "org_abc"}) + payload = verify_bearer_payload(token) + assert payload is not None + assert payload["sub"] == "user_123" + assert payload["tenant_id"] == "org_abc" + + def test_returns_none_for_invalid_token(self): + from nitrostack.auth.jwt import JWTService + from nitrostack.auth.request import verify_bearer_payload + from nitrostack.core.di import DIContainer + + DIContainer.reset() + service = JWTService() + DIContainer.get_instance().register_value(JWTService, service) + assert verify_bearer_payload("invalid.token.structure") is None + assert verify_bearer_payload("not-a-jwt") is None + + def test_raises_on_di_misconfig(self): + from nitrostack.auth.request import verify_bearer_payload + from nitrostack.core.di import DIContainer + from nitrostack.core.errors import DependencyResolutionError + + DIContainer.reset() + with patch.object( + DIContainer.get_instance(), + "resolve", + side_effect=DependencyResolutionError("JWTService missing"), + ): + with pytest.raises(DependencyResolutionError): + verify_bearer_payload("header.payload.signature")