Merge pull request #1240 from Aiven-Open/mbasani-fix-metrics-headers

Skip schema request headers for few endpoints

Author: tvainika

src/karapace/api/content_type.py

@@ -24,11 +24,16 @@
     "application/vnd.schemaregistry+json",
     JSON_CONTENT_TYPE,
 ]
+SCHEMA_RESPONSE_DEFAULT_CONTENT_TYPE = "application/vnd.schemaregistry.v1+json"
 
 
-def check_schema_headers(request: Request) -> str:
+def negotiate_schema_content_type(request: Request) -> str:
+    """Validate Accept and Content-Type headers for schema-registry endpoints.
+
+    Returns the negotiated response content type on success.
+    Raises HTTPException 406 or 415 on invalid headers.
+    """
     method = request.method
-    response_default_content_type = "application/vnd.schemaregistry.v1+json"
 
     message = Message()
     message["Content-Type"] = request.headers.get("Content-Type", JSON_CONTENT_TYPE)
@@ -42,14 +47,11 @@ def check_schema_headers(request: Request) -> str:
             detail={
                 "message": "HTTP 415 Unsupported Media Type",
             },
-            headers={
-                "Content-Type": response_default_content_type,
-            },
         )
     accept_val = request.headers.get("Accept")
     if accept_val:
         if accept_val in ("*/*", "*") or accept_val.startswith("*/"):
-            return response_default_content_type
+            return SCHEMA_RESPONSE_DEFAULT_CONTENT_TYPE
         content_type_match = get_best_match(accept_val, SCHEMA_ACCEPT_VALUES)
         if not content_type_match:
             LOG.debug("Unexpected Accept value: %r", accept_val)
@@ -58,9 +60,6 @@ def check_schema_headers(request: Request) -> str:
                 detail={
                     "message": "HTTP 406 Not Acceptable",
                 },
-                headers={
-                    "Content-Type": response_default_content_type,
-                },
             )
         return content_type_match
-    return response_default_content_type
+    return SCHEMA_RESPONSE_DEFAULT_CONTENT_TYPE

src/karapace/api/middlewares/__init__.py

@@ -6,7 +6,6 @@
 from collections.abc import Awaitable, Callable
 from fastapi import FastAPI, HTTPException, Request, Response
 from fastapi.responses import JSONResponse
-from karapace.api.content_type import check_schema_headers
 from karapace.api.telemetry.middleware import setup_telemetry_middleware
 
 from karapace.api.oidc.middleware import OIDCMiddleware
@@ -26,23 +25,6 @@ async def set_content_types(request: Request, call_next: Callable[[Request], Awa
         if request.url.path in {"/docs", "/docs/oauth2-redirect", "/redoc", "/openapi.json"}:
             return await call_next(request)
 
-        try:
-            response_content_type = check_schema_headers(request)
-        except HTTPException as exc:
-            return JSONResponse(
-                status_code=exc.status_code,
-                headers=exc.headers,
-                content=exc.detail,
-            )
-
-        # Schema registry supports application/octet-stream, assumption is JSON object body.
-        # Force internally to use application/json in this case for compatibility.
-        if request.headers.get("Content-Type") == "application/octet-stream":
-            new_headers = request.headers.mutablecopy()
-            new_headers["Content-Type"] = "application/json"
-            request._headers = new_headers
-            request.scope.update(headers=request.headers.raw)
-
         # Check for skip paths like /_health and /metrics and bypass
         if request.url.path in config.sasl_oauthbearer_skip_auth_paths:
             return await call_next(request)
@@ -79,7 +61,11 @@ async def set_content_types(request: Request, call_next: Callable[[Request], Awa
                 )
 
         response = await call_next(request)
-        response.headers["Content-Type"] = response_content_type
+
+        content_type = getattr(request.state, "schema_response_content_type", None)
+        if content_type:
+            response.headers["Content-Type"] = content_type
+
         return response
 
     setup_telemetry_middleware(app=app)

src/karapace/api/routers/compatibility.py

@@ -8,7 +8,7 @@
 from karapace.api.container import SchemaRegistryContainer
 from karapace.api.controller import KarapaceSchemaRegistryController
 from karapace.api.routers.errors import unauthorized
-from karapace.api.routers.raw_path_router import RawPathRoute
+from karapace.api.routers.raw_path_router import SchemaRegistryRoute
 from karapace.api.routers.requests import CompatibilityCheckResponse, SchemaRequest
 from karapace.api.user import get_current_user
 from karapace.core.auth import AuthenticatorAndAuthorizer, Operation, User
@@ -21,7 +21,7 @@
     prefix="/compatibility",
     tags=["compatibility"],
     responses={404: {"description": "Not found"}},
-    route_class=RawPathRoute,
+    route_class=SchemaRegistryRoute,
 )
 
 

src/karapace/api/routers/config.py

@@ -9,7 +9,7 @@
 from karapace.api.controller import KarapaceSchemaRegistryController
 from karapace.api.forward_client import ForwardClient
 from karapace.api.routers.errors import no_primary_url_error, unauthorized
-from karapace.api.routers.raw_path_router import RawPathRoute
+from karapace.api.routers.raw_path_router import SchemaRegistryRoute
 from karapace.api.routers.requests import CompatibilityLevelResponse, CompatibilityRequest, CompatibilityResponse
 from karapace.api.user import get_current_user
 from karapace.core.auth import AuthenticatorAndAuthorizer, Operation, User
@@ -23,7 +23,7 @@
     prefix="/config",
     tags=["config"],
     responses={404: {"description": "Not found"}},
-    route_class=RawPathRoute,
+    route_class=SchemaRegistryRoute,
 )
 
 

src/karapace/api/routers/mode.py

@@ -8,7 +8,7 @@
 from karapace.api.container import SchemaRegistryContainer
 from karapace.api.controller import KarapaceSchemaRegistryController
 from karapace.api.routers.errors import unauthorized
-from karapace.api.routers.raw_path_router import RawPathRoute
+from karapace.api.routers.raw_path_router import SchemaRegistryRoute
 from karapace.api.routers.requests import ModeResponse
 from karapace.api.user import get_current_user
 from karapace.core.auth import AuthenticatorAndAuthorizer, Operation, User
@@ -21,7 +21,7 @@
     prefix="/mode",
     tags=["mode"],
     responses={404: {"description": "Not found"}},
-    route_class=RawPathRoute,
+    route_class=SchemaRegistryRoute,
 )
 
 

src/karapace/api/routers/raw_path_router.py

@@ -9,8 +9,12 @@
 Copyright (c) 2018 Sebastián Ramírez
 """
 
-from fastapi import HTTPException  # noqa: E402
+from collections.abc import Coroutine, Callable  # noqa: E402
+from fastapi import HTTPException, Request, Response  # noqa: E402
+from typing import Any  # noqa: E402
+from fastapi.responses import JSONResponse  # noqa: E402
 from fastapi.routing import APIRoute  # noqa: E402
+from karapace.api.content_type import negotiate_schema_content_type, SCHEMA_RESPONSE_DEFAULT_CONTENT_TYPE  # noqa: E402
 from starlette.routing import Match  # noqa: E402
 from starlette.types import Scope  # noqa: E402
 
@@ -42,3 +46,39 @@ def matches(self, scope: Scope) -> tuple[Match, Scope]:
         new_path = re.sub(r"\?.*", "", raw_path)
         scope["path"] = new_path
         return super().matches(scope)
+
+
+class SchemaRegistryRoute(RawPathRoute):
+    """Route class for schema-registry endpoints that require content negotiation.
+
+    Validates Accept and Content-Type headers before any dependency injection or
+    body parsing occurs. Returns 406/415 errors immediately for invalid headers,
+    and sets the negotiated Content-Type on successful responses.
+    """
+
+    def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
+        original_handler = super().get_route_handler()
+
+        async def schema_content_handler(request: Request) -> Response:
+            try:
+                response_content_type = negotiate_schema_content_type(request)
+            except HTTPException as exc:
+                return JSONResponse(
+                    status_code=exc.status_code,
+                    content=exc.detail,
+                    headers={"Content-Type": SCHEMA_RESPONSE_DEFAULT_CONTENT_TYPE},
+                )
+
+            if request.headers.get("Content-Type") == "application/octet-stream":
+                new_headers = request.headers.mutablecopy()
+                new_headers["Content-Type"] = "application/json"
+                request._headers = new_headers
+                request.scope.update(headers=request.headers.raw)
+
+            request.state.schema_response_content_type = response_content_type
+
+            response = await original_handler(request)
+            response.headers["Content-Type"] = response_content_type
+            return response
+
+        return schema_content_handler

src/karapace/api/routers/schemas.py

@@ -7,6 +7,7 @@
 from fastapi import APIRouter, Depends, Query
 from karapace.api.container import SchemaRegistryContainer
 from karapace.api.controller import KarapaceSchemaRegistryController
+from karapace.api.routers.raw_path_router import SchemaRegistryRoute
 from karapace.api.routers.requests import SchemaListingItem, SchemasResponse, SubjectVersion
 from karapace.api.user import get_current_user
 from karapace.core.auth import AuthenticatorAndAuthorizer, User
@@ -18,6 +19,7 @@
     prefix="/schemas",
     tags=["schemas"],
     responses={404: {"description": "Not found"}},
+    route_class=SchemaRegistryRoute,
 )
 
 

src/karapace/api/routers/subjects.py

@@ -9,7 +9,7 @@
 from karapace.api.controller import KarapaceSchemaRegistryController
 from karapace.api.forward_client import ForwardClient
 from karapace.api.routers.errors import no_primary_url_error, unauthorized
-from karapace.api.routers.raw_path_router import RawPathRoute
+from karapace.api.routers.raw_path_router import SchemaRegistryRoute
 from karapace.api.routers.requests import SchemaIdResponse, SchemaRequest, SchemaResponse, SubjectSchemaVersionResponse
 from karapace.api.user import get_current_user
 from karapace.core.auth import AuthenticatorAndAuthorizer, Operation, User
@@ -28,7 +28,7 @@
     prefix="/subjects",
     tags=["subjects"],
     responses={404: {"description": "Not found"}},
-    route_class=RawPathRoute,
+    route_class=SchemaRegistryRoute,
 )
 
 

tests/e2e/instrumentation/test_prometheus.py

@@ -21,6 +21,24 @@ async def test_metrics_endpoint(registry_async_client: Client) -> None:
     assert result.status_code == HTTPStatus.OK.value
 
 
+async def test_metrics_endpoint_with_prometheus_accept_header(registry_async_client: Client) -> None:
+    """Regression test: Prometheus sends Accept headers that don't match schema-registry content types.
+
+    The /metrics endpoint must be excluded from the schema-registry content negotiation
+    middleware, otherwise it returns HTTP 406 Not Acceptable.
+    """
+    prometheus_accept = (
+        "application/openmetrics-text;version=1.0.0,application/openmetrics-text;version=0.0.1;q=0.75,"
+        "text/plain;version=0.0.4;q=0.5,*/*;q=0.1"
+    )
+    result: Result = await registry_async_client.get(
+        PrometheusInstrumentation.METRICS_ENDPOINT_PATH,
+        headers={"Accept": prometheus_accept},
+        json_response=False,
+    )
+    assert result.status_code == HTTPStatus.OK.value
+
+
 async def test_metrics_endpoint_parsed_response(registry_async_client: Client) -> None:
     result: Result = await registry_async_client.get(
         PrometheusInstrumentation.METRICS_ENDPOINT_PATH,

tests/integration/test_health_check.py

@@ -13,6 +13,31 @@
 from tests.integration.utils.cluster import RegistryDescription
 
 
+async def test_health_check_with_non_schema_accept_header(registry_async_client: Client) -> None:
+    """Regression test: /_health must be exempt from schema-registry content negotiation.
+
+    Monitoring tools (e.g. Kubernetes probes) may send Accept headers that don't match
+    schema-registry content types. This must not result in HTTP 406.
+    """
+    res = await registry_async_client.get("/_health", headers={"Accept": "text/plain, */*;q=0.1"})
+    assert res.status_code in (
+        http.HTTPStatus.OK,
+        http.HTTPStatus.SERVICE_UNAVAILABLE,
+    ), f"Expected 200 or 503, got {res.status_code}"
+
+
+async def test_root_with_non_schema_accept_header(registry_async_client: Client) -> None:
+    """Regression test: / must be exempt from schema-registry content negotiation."""
+    res = await registry_async_client.get("/", headers={"Accept": "text/html"})
+    assert res.ok, f"Expected 200, got {res.status_code}"
+
+
+async def test_master_available_with_non_schema_accept_header(registry_async_client: Client) -> None:
+    """Regression test: /master_available must be exempt from schema-registry content negotiation."""
+    res = await registry_async_client.get("/master_available", headers={"Accept": "text/plain"})
+    assert res.ok, f"Expected 200, got {res.status_code}"
+
+
 async def test_health_check(
     registry_cluster: RegistryDescription, registry_async_client: Client, admin_client: KafkaAdminClient
 ) -> None: