fix: check private endpoint scopes when default_public=False

Author: alukach

src/stac_auth_proxy/utils/requests.py

@@ -26,6 +26,23 @@ def dict_to_bytes(d: dict) -> bytes:
     return json.dumps(d, separators=(",", ":")).encode("utf-8")
 
 
+def _check_endpoint_match(
+    path: str,
+    method: str,
+    endpoints: EndpointMethods,
+) -> tuple[bool, Sequence[str]]:
+    """Check if the path and method match any endpoint in the given endpoints map."""
+    for pattern, endpoint_methods in endpoints.items():
+        if re.match(pattern, path):
+            for endpoint_method in endpoint_methods:
+                required_scopes: Sequence[str] = []
+                if isinstance(endpoint_method, tuple):
+                    endpoint_method, required_scopes = endpoint_method
+                if method.casefold() == endpoint_method.casefold():
+                    return True, required_scopes
+    return False, []
+
+
 def find_match(
     path: str,
     method: str,
@@ -34,22 +51,25 @@ def find_match(
     default_public: bool,
 ) -> "MatchResult":
     """Check if the given path and method match any of the regex patterns and methods in the endpoints."""
-    endpoints = private_endpoints if default_public else public_endpoints
-    for pattern, endpoint_methods in endpoints.items():
-        if not re.match(pattern, path):
-            continue
-        for endpoint_method in endpoint_methods:
-            required_scopes: Sequence[str] = []
-            if isinstance(endpoint_method, tuple):
-                endpoint_method, required_scopes = endpoint_method
-            if method.casefold() == endpoint_method.casefold():
-                # If default_public, we're looking for a private endpoint.
-                # If not default_public, we're looking for a public endpoint.
-                return MatchResult(
-                    is_private=default_public,
-                    required_scopes=required_scopes,
-                )
-    return MatchResult(is_private=not default_public)
+    primary_endpoints = private_endpoints if default_public else public_endpoints
+    matched, required_scopes = _check_endpoint_match(path, method, primary_endpoints)
+    if matched:
+        return MatchResult(
+            is_private=default_public,
+            required_scopes=required_scopes,
+        )
+
+    # If default_public and no match found in private_endpoints, it's public
+    if default_public:
+        return MatchResult(is_private=False)
+
+    # If not default_public, check private_endpoints for required scopes
+    matched, required_scopes = _check_endpoint_match(path, method, private_endpoints)
+    if matched:
+        return MatchResult(is_private=True, required_scopes=required_scopes)
+
+    # Default case: if not default_public and no explicit match, it's private
+    return MatchResult(is_private=True)
 
 
 @dataclass

tests/test_authn.py

@@ -50,27 +50,56 @@ def test_default_public_false(source_api_server, path, method, token_builder):
     assert response.status_code == 200
 
 
+@pytest.mark.parametrize(
+    "token,permitted",
+    [
+        [{"scope": "collection:create"}, True],
+        [{"scope": ""}, False],
+        [{"scope": "openid"}, False],
+        [{"scope": "openid collection:create"}, True],
+    ],
+)
+def test_default_public_false_with_scopes(
+    source_api_server, token, permitted, token_builder
+):
+    """Private endpoints permit access with a valid token."""
+    test_app = app_factory(
+        upstream_url=source_api_server,
+        default_public=False,
+        private_endpoints={r"^/collections$": [("POST", ["collection:create"])]},
+    )
+    valid_auth_token = token_builder(token)
+
+    client = TestClient(test_app)
+    response = client.request(
+        method="POST",
+        url="/collections",
+        headers={"Authorization": f"Bearer {valid_auth_token}"},
+    )
+    assert response.status_code == (200 if permitted else 401)
+
+
 @pytest.mark.parametrize(
     "token_scopes, private_endpoints, path, method, expected_permitted",
     [
         pytest.param(
             "",
-            {r"^/*": [("POST", ["collections:create"])]},
+            {r"^/*": [("POST", ["collection:create"])]},
             "/collections",
             "POST",
             False,
             id="empty scopes + private endpoint",
         ),
         pytest.param(
-            "openid profile collections:createbutnotcreate",
-            {r"^/*": [("POST", ["collections:create"])]},
+            "openid profile collection:createbutnotcreate",
+            {r"^/*": [("POST", ["collection:create"])]},
             "/collections",
             "POST",
             False,
             id="invalid scopes + private endpoint",
         ),
         pytest.param(
-            "openid profile collections:create somethingelse",
+            "openid profile collection:create somethingelse",
             {r"^/*": [("POST", [])]},
             "/collections",
             "POST",
@@ -79,7 +108,7 @@ def test_default_public_false(source_api_server, path, method, token_builder):
         ),
         pytest.param(
             "openid",
-            {r"^/collections/.*/items$": [("POST", ["collections:create"])]},
+            {r"^/collections/.*/items$": [("POST", ["collection:create"])]},
             "/collections",
             "GET",
             True,