Merge pull request #1237 from Aiven-Open/mbasani-add-test-user-not-found

Improvements to normalization

Author: jclarysse

src/karapace/core/instrumentation/path_normalization.py

@@ -5,24 +5,28 @@
 See LICENSE for details
 """
 
-from __future__ import annotations
-
 import re
+from enum import Enum
+
+
+class PathPattern(Enum):
+    """Regex patterns and their replacements for normalizing dynamic path segments.
+
+    Each member is a (pattern, replacement) pair. Patterns are applied in
+    declaration order -- put broader matches (e.g. UUIDs) before narrower ones.
+    """
+
+    UUID = (r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}", "{uuid}")
+    TOPIC = (r"(/topics/)([^/]+)", r"\1{topic}")
+    SCHEMA_ID = (r"(/schemas/ids/)(\d+)", r"\1{id}")
+    SUBJECT = (r"(/subjects/)([^/]+)", r"\1{subject}")
+    VERSION = (r"(/versions/)(\d+)", r"\1{version}")
+    CONFIG_SUBJECT = (r"(/config/)([^/]+)", r"\1{subject}")
+    MODE_SUBJECT = (r"(/mode/)([^/]+)", r"\1{subject}")
 
-# UUIDs: 8-4-4-4-12 hex format
-UUID_PATTERN = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}")
-# Topic names after /topics/ - match until next slash or end
-TOPIC_PATTERN = re.compile(r"(/topics/)([^/]+)")
-# Schema IDs (numeric) after /schemas/ids/
-SCHEMA_ID_PATTERN = re.compile(r"(/schemas/ids/)(\d+)")
-# Subject names after /subjects/
-SUBJECT_PATTERN = re.compile(r"(/subjects/)([^/]+)")
-# Version numbers after /versions/
-VERSION_PATTERN = re.compile(r"(/versions/)(\d+)")
-# Subject names after /config/
-CONFIG_SUBJECT_PATTERN = re.compile(r"(/config/)([^/]+)")
-# Subject names after /mode/
-MODE_SUBJECT_PATTERN = re.compile(r"(/mode/)([^/]+)")
+    def __init__(self, pattern: str, replacement: str) -> None:
+        self.regex = re.compile(pattern)
+        self.replacement = replacement
 
 
 def normalize_path(path: str) -> str:
@@ -33,17 +37,13 @@ def normalize_path(path: str) -> str:
 
     Examples:
         /topics/my-topic -> /topics/{topic}
-        /consumers/abc-123-def/instances/xyz-456 -> /consumers/{uuid}/instances/{uuid}
+        /consumers/3f410583-cfa3-48e8-bd9f-22eff1881c00/instances/0333f5a1-d242-42bc-ba35-d7c5d67fc121 -> /consumers/{uuid}/instances/{uuid}
         /schemas/ids/42 -> /schemas/ids/{id}
         /subjects/my-subject/versions/3 -> /subjects/{subject}/versions/{version}
         /config/my-subject -> /config/{subject}
         /mode/my-subject -> /mode/{subject}
     """
-    normalized = UUID_PATTERN.sub("{uuid}", path)
-    normalized = TOPIC_PATTERN.sub(r"\1{topic}", normalized)
-    normalized = SCHEMA_ID_PATTERN.sub(r"\1{id}", normalized)
-    normalized = SUBJECT_PATTERN.sub(r"\1{subject}", normalized)
-    normalized = VERSION_PATTERN.sub(r"\1{version}", normalized)
-    normalized = CONFIG_SUBJECT_PATTERN.sub(r"\1{subject}", normalized)
-    normalized = MODE_SUBJECT_PATTERN.sub(r"\1{subject}", normalized)
+    normalized = path
+    for p in PathPattern:
+        normalized = p.regex.sub(p.replacement, normalized)
     return normalized

tests/integration/test_schema_coordinator.py

@@ -167,6 +167,7 @@ async def test_coordinator_workflow(
         session_timeout_ms=10000,
         heartbeat_interval_ms=500,
         retry_backoff_ms=100,
+        waiting_time_before_acting_as_master_ms=waiting_time_before_acting_as_master_sec * 1000,
     )
     coordinator2.start()
     assert coordinator2.coordinator_id is None
@@ -205,8 +206,8 @@ async def test_coordinator_workflow(
     await primary.close()
     await primary_client.close()
 
-    now = time.time()
-    while time.time() - now < waiting_time_before_acting_as_master_sec:
+    now = time.monotonic()
+    while time.monotonic() - now < waiting_time_before_acting_as_master_sec:
         assert not secondary.are_we_master(), (
             f"Cannot become master before {waiting_time_before_acting_as_master_sec} seconds "
             f"for the late records that can arrive from the previous master"

tests/unit/instrumentation/test_prometheus.py

@@ -233,3 +233,95 @@ async def handler(req):
 
         # Total should NOT be incremented when handler raises a non-HTTP exception
         total_metric.labels.assert_not_called()
+
+    @patch("karapace.core.instrumentation.prometheus.time")
+    async def test_http_request_metrics_middleware_normalizes_path(
+        self,
+        mock_time: MagicMock,
+        prometheus: PrometheusInstrumentation,
+    ) -> None:
+        """Verify that the middleware normalizes dynamic path segments to prevent unbounded metric cardinality."""
+        mock_time.time.return_value = 10
+
+        (
+            app_metrics,
+            in_progress_metric,
+            in_progress_instance,
+            duration_metric,
+            duration_instance,
+            total_metric,
+            total_instance,
+        ) = _make_metric_mocks(prometheus)
+
+        request = DummyRequest(path="/schemas/ids/878916964", method="GET", app=app_metrics)
+
+        response = MagicMock(status=200)
+
+        async def handler(req):
+            return response
+
+        await prometheus.http_request_metrics_middleware(request=request, handler=handler)
+
+        in_progress_metric.labels.assert_called_with("GET", "/schemas/ids/{id}")
+        total_metric.labels.assert_called_with("GET", "/schemas/ids/{id}", response.status)
+        duration_metric.labels.assert_called_with("GET", "/schemas/ids/{id}")
+
+    @patch("karapace.core.instrumentation.prometheus.time")
+    async def test_http_request_metrics_middleware_normalizes_subject_version_path(
+        self,
+        mock_time: MagicMock,
+        prometheus: PrometheusInstrumentation,
+    ) -> None:
+        mock_time.time.return_value = 10
+
+        (
+            app_metrics,
+            in_progress_metric,
+            in_progress_instance,
+            duration_metric,
+            duration_instance,
+            total_metric,
+            total_instance,
+        ) = _make_metric_mocks(prometheus)
+
+        request = DummyRequest(path="/subjects/my-subject/versions/3", method="GET", app=app_metrics)
+
+        response = MagicMock(status=200)
+
+        async def handler(req):
+            return response
+
+        await prometheus.http_request_metrics_middleware(request=request, handler=handler)
+
+        in_progress_metric.labels.assert_called_with("GET", "/subjects/{subject}/versions/{version}")
+        total_metric.labels.assert_called_with("GET", "/subjects/{subject}/versions/{version}", response.status)
+
+    @patch("karapace.core.instrumentation.prometheus.time")
+    async def test_http_request_metrics_middleware_normalizes_config_subject_path(
+        self,
+        mock_time: MagicMock,
+        prometheus: PrometheusInstrumentation,
+    ) -> None:
+        mock_time.time.return_value = 10
+
+        (
+            app_metrics,
+            in_progress_metric,
+            in_progress_instance,
+            duration_metric,
+            duration_instance,
+            total_metric,
+            total_instance,
+        ) = _make_metric_mocks(prometheus)
+
+        request = DummyRequest(path="/config/journeys_v1-dbx", method="GET", app=app_metrics)
+
+        response = MagicMock(status=200)
+
+        async def handler(req):
+            return response
+
+        await prometheus.http_request_metrics_middleware(request=request, handler=handler)
+
+        in_progress_metric.labels.assert_called_with("GET", "/config/{subject}")
+        total_metric.labels.assert_called_with("GET", "/config/{subject}", response.status)

tests/unit/test_auth.py

@@ -5,7 +5,18 @@
 
 import re
 
-from karapace.core.auth import ACLAuthorizer, ACLEntry, HashAlgorithm, Operation, User, hash_password
+import pytest
+
+from karapace.core.auth import (
+    ACLAuthorizer,
+    ACLEntry,
+    AuthenticationError,
+    HashAlgorithm,
+    HTTPAuthorizer,
+    Operation,
+    User,
+    hash_password,
+)
 
 
 def test_empty_acl_authorizer() -> None:
@@ -223,3 +234,53 @@ def test_acl_authorizer() -> None:
             "Subject:readwrite_subject",
         ],
     )
+
+
+def test_get_user_returns_none_for_nonexistent_user() -> None:
+    """get_user must return None (not raise) for unknown usernames.
+
+    Regression test: a previous implementation raised ValueError here,
+    which bypassed the AuthenticationError handling in authenticate()
+    and surfaced as an unhandled 500 to clients.
+    """
+    admin_password_hash = hash_password(algorithm=HashAlgorithm.SHA256, salt="salt", plaintext_password="password")
+    authorizer = ACLAuthorizer(
+        user_db={
+            "admin": User(username="admin", algorithm=HashAlgorithm.SHA256, salt="salt", password_hash=admin_password_hash),
+        },
+    )
+
+    assert authorizer.get_user("admin") is not None
+    assert authorizer.get_user("nonexistent") is None
+
+
+def test_authenticate_raises_authentication_error_for_nonexistent_user() -> None:
+    """authenticate() must raise AuthenticationError -- not ValueError --
+    when the user does not exist, so the caller can return a proper 401."""
+    admin_password_hash = hash_password(algorithm=HashAlgorithm.SHA256, salt="salt", plaintext_password="password")
+    authorizer = ACLAuthorizer(
+        user_db={
+            "admin": User(username="admin", algorithm=HashAlgorithm.SHA256, salt="salt", password_hash=admin_password_hash),
+        },
+    )
+    http_authorizer = HTTPAuthorizer.__new__(HTTPAuthorizer)
+    http_authorizer.user_db = authorizer.user_db
+    http_authorizer.permissions = authorizer.permissions
+
+    with pytest.raises(AuthenticationError):
+        http_authorizer.authenticate(username="nonexistent", password="any")
+
+
+def test_authenticate_raises_authentication_error_for_wrong_password() -> None:
+    admin_password_hash = hash_password(algorithm=HashAlgorithm.SHA256, salt="salt", plaintext_password="password")
+    authorizer = ACLAuthorizer(
+        user_db={
+            "admin": User(username="admin", algorithm=HashAlgorithm.SHA256, salt="salt", password_hash=admin_password_hash),
+        },
+    )
+    http_authorizer = HTTPAuthorizer.__new__(HTTPAuthorizer)
+    http_authorizer.user_db = authorizer.user_db
+    http_authorizer.permissions = authorizer.permissions
+
+    with pytest.raises(AuthenticationError):
+        http_authorizer.authenticate(username="admin", password="wrong")