fix(auth): deny CIMD host mismatch, require pinned IP for outbound peer, remove duplicate helpers

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")

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
 
 

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
 

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")