VYPR
High severityNVD Advisory· Published Sep 8, 2025· Updated Sep 9, 2025

Fides Webserver API is Vulnerable to OAuth Client Privilege Escalation

CVE-2025-57817

Description

Fides is an open-source privacy engineering platform. Prior to version 2.69.1, the OAuth client creation and update endpoints of the Fides Webserver API do not properly authorize scope assignment. This allows highly privileged users with client:create or client:update permissions to escalate their privileges to owner-level. Version 2.69.1 fixes the issue. No known workarounds are available.

Affected packages

Versions sourced from the GitHub Security Advisory.

PackageAffected versionsPatched versions
ethyca-fidesPyPI
< 2.69.12.69.1

Affected products

1

Patches

1
2ffd125e1089

Merge commit from fork

https://github.com/ethyca/fidesThabo FletcherSep 2, 2025via ghsa
4 files changed · +558 20
  • CHANGELOG.md+5 4 modified
    @@ -21,6 +21,11 @@ Changes can also be flagged with a GitHub label for tracking purposes. The URL o
     
     ## [Unreleased](https://github.com/ethyca/fides/compare/2.69.0...main)
     
    +### Security
    +- Fixed OAuth scope privilege escalation vulnerability that allowed clients to create or update other OAuth clients with unauthorized scopes [CVE-2025-57817](https://github.com/ethyca/fides/security/advisories/GHSA-hjfh-p8f5-24wr)
    +- Added stricter rate limiting to authentication endpoints to mitigate against brute force attacks. [CVE-2025-57815](https://github.com/ethyca/fides/security/advisories/GHSA-7q62-r88r-j5gw)
    +- Adds Redis-driven rate limiting across all endpoints [CVE-2025-57816](https://github.com/ethyca/fides/security/advisories/GHSA-fq34-xw6c-fphf)
    +
     ## [2.69.0](https://github.com/ethyca/fides/compare/2.68.0...2.69.0)
     
     ### Added
    @@ -62,10 +67,6 @@ Changes can also be flagged with a GitHub label for tracking purposes. The URL o
     - Handle missing GVL in TCF experience by displaying an error message instead of infinite spinners. [#6472](https://github.com/ethyca/fides/pull/6472)
     - Prevent edits for assets that have been ignored in the Action Center [#6485](https://github.com/ethyca/fides/pull/6485)
     
    -### Security
    -- Added stricter rate limiting to authentication endpoints to mitigate against brute force attacks. [CVE-2025-57815](https://github.com/ethyca/fides/security/advisories/GHSA-7q62-r88r-j5gw)
    -- Adds Redis-driven rate limiting across all endpoints [CVE-2025-57816](https://github.com/ethyca/fides/security/advisories/GHSA-fq34-xw6c-fphf)
    -
     ## [2.68.0](https://github.com/ethyca/fides/compare/2.67.2...2.68.0)
     
     ### Added
    
  • src/fides/api/api/v1/endpoints/oauth_endpoints.py+18 5 modified
    @@ -8,6 +8,7 @@
     from sqlalchemy.orm import Session
     from starlette.status import (
         HTTP_400_BAD_REQUEST,
    +    HTTP_403_FORBIDDEN,
         HTTP_404_NOT_FOUND,
         HTTP_422_UNPROCESSABLE_ENTITY,
     )
    @@ -26,7 +27,7 @@
     from fides.api.models.connectionconfig import ConnectionConfig, ConnectionTestStatus
     from fides.api.models.fides_user import FidesUser
     from fides.api.oauth.roles import ROLES_TO_SCOPES_MAPPING
    -from fides.api.oauth.utils import verify_oauth_client
    +from fides.api.oauth.utils import verify_client_can_assign_scopes, verify_oauth_client
     from fides.api.schemas.client import ClientCreatedResponse
     from fides.api.schemas.oauth import AccessToken, OAuth2ClientCredentialsRequestForm
     from fides.api.service.authentication.authentication_strategy import (
    @@ -123,22 +124,28 @@ async def acquire_access_token(
     
     @router.post(
         CLIENT,
    -    dependencies=[Security(verify_oauth_client, scopes=[CLIENT_CREATE])],
         response_model=ClientCreatedResponse,
     )
     def create_client(
         *,
    +    request: Request,
         db: Session = Depends(get_db),
         scopes: List[str] = Body([]),
    +    requesting_client: ClientDetail = Security(
    +        verify_oauth_client, scopes=[CLIENT_CREATE]
    +    ),
     ) -> ClientCreatedResponse:
         """Creates a new client and returns the credentials. Only direct scopes can be added to the client via this endpoint."""
         logger.info("Creating new client")
         if not all(scope in SCOPE_REGISTRY for scope in scopes):
             raise HTTPException(
    -            status_code=HTTP_422_UNPROCESSABLE_ENTITY,
    +            status_code=HTTP_403_FORBIDDEN,
                 detail=f"Invalid Scope. Scopes must be one of {SCOPE_REGISTRY}.",
             )
     
    +    # Security check: Verify that the requesting client has all the scopes they're trying to assign
    +    verify_client_can_assign_scopes(request, requesting_client, scopes, db)
    +
         client, secret = ClientDetail.create_client_and_secret(
             db,
             CONFIG.security.oauth_client_id_length_bytes,
    @@ -180,13 +187,16 @@ def get_client_scopes(client_id: str, db: Session = Depends(get_db)) -> List[str
     
     @router.put(
         CLIENT_SCOPE,
    -    dependencies=[Security(verify_oauth_client, scopes=[CLIENT_UPDATE])],
         response_model=None,
     )
     def set_client_scopes(
         client_id: str,
         scopes: List[str],
    +    request: Request,
         db: Session = Depends(get_db),
    +    requesting_client: ClientDetail = Security(
    +        verify_oauth_client, scopes=[CLIENT_UPDATE]
    +    ),
     ) -> None:
         """Overwrites the client's directly-assigned scopes with those provided.
         Roles cannot be edited via this endpoint.
    @@ -197,10 +207,13 @@ def set_client_scopes(
     
         if not all(elem in SCOPE_REGISTRY for elem in scopes):
             raise HTTPException(
    -            status_code=HTTP_422_UNPROCESSABLE_ENTITY,
    +            status_code=HTTP_403_FORBIDDEN,
                 detail=f"Invalid Scope. Scopes must be one of {SCOPE_REGISTRY}.",
             )
     
    +    # Security check: Verify that the requesting client has all the scopes they're trying to assign
    +    verify_client_can_assign_scopes(request, requesting_client, scopes, db)
    +
         logger.info("Updating client scopes")
         client.update(db, data={"scopes": scopes})
     
    
  • src/fides/api/oauth/utils.py+84 3 modified
    @@ -6,14 +6,14 @@
     from types import FunctionType
     from typing import Any, Callable, Dict, List, Optional, Tuple
     
    -from fastapi import Depends, HTTPException, Security
    +from fastapi import Depends, HTTPException, Request, Security
     from fastapi.security import SecurityScopes
     from jose import exceptions, jwe
     from jose.constants import ALGORITHMS
     from loguru import logger
     from pydantic import ValidationError
     from sqlalchemy.orm import Session
    -from starlette.status import HTTP_404_NOT_FOUND
    +from starlette.status import HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND
     
     from fides.api.api.deps import get_db
     from fides.api.common_exceptions import AuthenticationError, AuthorizationError
    @@ -30,7 +30,7 @@
     from fides.api.models.policy import PolicyPreWebhook
     from fides.api.models.pre_approval_webhook import PreApprovalWebhook
     from fides.api.models.privacy_request import RequestTask
    -from fides.api.oauth.roles import get_scopes_from_roles
    +from fides.api.oauth.roles import ROLES_TO_SCOPES_MAPPING, get_scopes_from_roles
     from fides.api.request_context import set_user_id
     from fides.api.schemas.external_https import (
         DownloadTokenJWE,
    @@ -476,6 +476,87 @@ def has_scope_subset(user_scopes: List[str], endpoint_scopes: SecurityScopes) ->
         return set(endpoint_scopes.scopes).issubset(user_scopes)
     
     
    +def get_client_effective_scopes(client: "ClientDetail") -> List[str]:
    +    """
    +    Get all scopes available to a client, including both direct scopes and role-derived scopes.
    +
    +    Args:
    +        client: The ClientDetail instance
    +
    +    Returns:
    +        List of scope strings that the client has access to
    +    """
    +    effective_scopes = set()
    +
    +    # Add direct scopes
    +    if client.scopes:
    +        effective_scopes.update(client.scopes)
    +
    +    # Add role-derived scopes
    +    if client.roles:
    +        for role in client.roles:
    +            role_scopes = ROLES_TO_SCOPES_MAPPING.get(role, [])
    +            effective_scopes.update(role_scopes)
    +
    +    # Add user permission scopes if client is associated with a user
    +    # Note: client.user is available via SQLAlchemy backref from FidesUser.client relationship
    +    user = getattr(client, "user", None)  # Use getattr to avoid mypy attr-defined error
    +    if user and hasattr(user, "permissions") and user.permissions:
    +        effective_scopes.update(user.permissions.total_scopes)
    +
    +    return sorted(list(effective_scopes))
    +
    +
    +def verify_client_can_assign_scopes(
    +    request: "Request",
    +    requesting_client: "ClientDetail",
    +    scopes: List[str],
    +    db: "Session",
    +) -> None:
    +    """
    +    Verify that a requesting client has permission to assign the given scopes.
    +
    +    Raises HTTPException if the client lacks permission.
    +    Root client is exempt from this check.
    +
    +    Args:
    +        request: FastAPI request object containing Authorization header
    +        requesting_client: The client making the request
    +        scopes: List of scopes to be assigned
    +        db: Database session
    +
    +    Raises:
    +        HTTPException: If the client lacks permission to assign the scopes
    +    """
    +    # Root client can assign any scope
    +    if requesting_client.id == CONFIG.security.oauth_root_client_id:
    +        return
    +
    +    # Get the actual token scopes (not the client's database scopes)
    +    authorization = request.headers.get("Authorization", "").replace("Bearer ", "")
    +    token_data, _ = extract_token_and_load_client(authorization, db)
    +
    +    # Get token's effective scopes
    +    token_scopes = token_data.get("scopes", [])
    +
    +    # Check if user has the scopes via roles as well
    +    has_scope_via_role = has_permissions(
    +        token_data=token_data,
    +        client=requesting_client,
    +        endpoint_scopes=SecurityScopes(scopes),
    +    )
    +
    +    # If they don't have all scopes via direct assignment or roles, check individual scopes
    +    if not has_scope_via_role:
    +        unauthorized_scopes = set(scopes) - set(token_scopes)
    +
    +        if unauthorized_scopes:
    +            raise HTTPException(
    +                status_code=HTTP_403_FORBIDDEN,
    +                detail=f"Cannot assign scopes that you do not have. Missing scopes: {sorted(unauthorized_scopes)}",
    +            )
    +
    +
     def create_temporary_user_for_login_flow(config: FidesConfig) -> FidesUser:
         """
         Create a temporary FidesUser in-memory with an attached in-memory ClientDetail
    
  • tests/ops/api/v1/endpoints/test_oauth_endpoints.py+451 8 modified
    @@ -117,13 +117,14 @@ def test_create_client_with_scopes(
             url,
             generate_auth_header,
         ) -> None:
    -        auth_header = generate_auth_header([CLIENT_CREATE])
    -
    +        # Note: requesting client must have all scopes they want to assign (security fix)
             scopes = [
                 CLIENT_CREATE,
                 CLIENT_DELETE,
                 CLIENT_READ,
             ]
    +        auth_header = generate_auth_header(scopes)  # Give requester all needed scopes
    +
             response = api_client.post(
                 url,
                 headers=auth_header,
    @@ -154,7 +155,7 @@ def test_create_client_with_invalid_scopes(
                 json=["invalid-scope"],
             )
     
    -        assert 422 == response.status_code
    +        assert 403 == response.status_code
             assert response.json()["detail"].startswith("Invalid Scope.")
     
         def test_create_client_with_root_client(self, url, api_client):
    @@ -266,15 +267,15 @@ def test_set_invalid_scope(
             response = api_client.put(
                 url, headers=auth_header, json=["this-is-not-a-valid-scope"]
             )
    -        assert 422 == response.status_code
    +        assert 403 == response.status_code
     
         def test_set_scopes_invalid_client(
             self, api_client: TestClient, oauth_client: ClientDetail, generate_auth_header
         ) -> None:
             url = V1_URL_PREFIX + CLIENT_SCOPE.format(client_id="bad_client")
     
             auth_header = generate_auth_header([CLIENT_UPDATE])
    -        response = api_client.put(url, headers=auth_header, json=["storage:read"])
    +        response = api_client.put(url, headers=auth_header, json=[STORAGE_READ])
             response_body = json.loads(response.text)
     
             assert 200 == response.status_code
    @@ -288,16 +289,17 @@ def test_set_scopes(
             url,
             generate_auth_header,
         ) -> None:
    -        auth_header = generate_auth_header([CLIENT_UPDATE])
    +        # Note: requesting client must have all scopes they want to assign (security fix)
    +        auth_header = generate_auth_header([CLIENT_UPDATE, STORAGE_READ])
     
    -        response = api_client.put(url, headers=auth_header, json=["storage:read"])
    +        response = api_client.put(url, headers=auth_header, json=[STORAGE_READ])
             response_body = json.loads(response.text)
     
             assert 200 == response.status_code
             assert response_body is None
     
             db.refresh(oauth_client)
    -        assert oauth_client.scopes == ["storage:read"]
    +        assert oauth_client.scopes == [STORAGE_READ]
     
     
     class TestReadScopes:
    @@ -688,3 +690,444 @@ def test_get_role_scope_mapping(
                 "viewer",
                 "viewer_and_approver",
             }
    +
    +
    +class TestOAuthPrivilegeEscalationSecurity:
    +    """Security tests to verify the OAuth privilege escalation vulnerability has been fixed."""
    +
    +    @pytest.fixture(scope="function")
    +    def create_client_url(self, oauth_client) -> str:
    +        return V1_URL_PREFIX + CLIENT
    +
    +    @pytest.fixture(scope="function")
    +    def set_scopes_url(self, oauth_client) -> str:
    +        return V1_URL_PREFIX + CLIENT_SCOPE.format(client_id=oauth_client.id)
    +
    +    @pytest.fixture(scope="function")
    +    def contributor_client(self, db) -> ClientDetail:
    +        """Create a client with contributor-level scopes for testing."""
    +        client, _ = ClientDetail.create_client_and_secret(
    +            db,
    +            CONFIG.security.oauth_client_id_length_bytes,
    +            CONFIG.security.oauth_client_secret_length_bytes,
    +            scopes=[CLIENT_CREATE, CLIENT_UPDATE, CLIENT_READ],  # Contributor scopes
    +        )
    +        yield client
    +        client.delete(db)
    +
    +    def test_create_client_privilege_escalation_blocked(
    +        self,
    +        db,
    +        api_client: TestClient,
    +        create_client_url,
    +        generate_auth_header,
    +    ) -> None:
    +        """Test that a client cannot create another client with scopes they don't have."""
    +        # Client only has CLIENT_CREATE scope
    +        auth_header = generate_auth_header([CLIENT_CREATE])
    +
    +        # Try to create a client with owner-level scopes that the requester doesn't have
    +        privileged_scopes = [
    +            "storage:create_or_update",
    +            "messaging:create_or_update",
    +            "user-permission:assign_owners",  # Note: hyphen, not underscore
    +        ]
    +
    +        response = api_client.post(
    +            create_client_url,
    +            headers=auth_header,
    +            json=privileged_scopes,
    +        )
    +
    +        assert response.status_code == 403
    +        response_data = response.json()
    +        assert "Cannot assign scopes that you do not have" in response_data["detail"]
    +        assert "storage:create_or_update" in response_data["detail"]
    +
    +    def test_create_client_with_insecure_scopes(
    +        self,
    +        db,
    +        api_client: TestClient,
    +        create_client_url,
    +        generate_auth_header,
    +    ) -> None:
    +        """Test that creating a client with unauthorized scopes is blocked (security vulnerability fix)."""
    +        # Attacker only has basic client management scope
    +        auth_header = generate_auth_header([CLIENT_CREATE])
    +
    +        # Try to escalate privileges by creating a client with admin scopes they don't have
    +        insecure_scopes = [
    +            "storage:delete",  # Can delete storage backends
    +            "messaging:delete",  # Can control messaging systems
    +            "user-permission:assign_owners",  # Can create owner accounts
    +            "user:delete",  # Can delete users
    +        ]
    +
    +        response = api_client.post(
    +            create_client_url,
    +            headers=auth_header,
    +            json=insecure_scopes,
    +        )
    +
    +        # Should be rejected with 403 Forbidden
    +        assert response.status_code == 403
    +        response_data = response.json()
    +        assert "Cannot assign scopes that you do not have" in response_data["detail"]
    +
    +        # Verify all unauthorized scopes are listed
    +        for scope in insecure_scopes:
    +            assert scope in response_data["detail"]
    +
    +        # Verify no new client was created (the request should have been blocked)
    +        # Since we expect a 422 status, no client should have been created by this request
    +        # (This is already implicitly tested by the 422 status check above, but let's be explicit)
    +
    +    def test_create_client_with_role_derived_scopes_blocked(
    +        self,
    +        db,
    +        api_client: TestClient,
    +        create_client_url,
    +    ) -> None:
    +        """Test that role-derived scopes are properly considered in authorization checks."""
    +        from fides.api.models.client import ClientDetail
    +        from fides.api.models.fides_user import FidesUser
    +        from fides.api.oauth.roles import CONTRIBUTOR
    +        from tests.conftest import generate_role_header_for_user
    +
    +        # Create a user and client with only the CONTRIBUTOR role (no direct scopes)
    +        user = FidesUser.create(
    +            db=db,
    +            data={
    +                "username": "test_role_user_blocked",
    +                "first_name": "Test",
    +                "last_name": "User",
    +            },
    +        )
    +
    +        requesting_client, _ = ClientDetail.create_client_and_secret(
    +            db,
    +            CONFIG.security.oauth_client_id_length_bytes,
    +            CONFIG.security.oauth_client_secret_length_bytes,
    +            roles=[
    +                CONTRIBUTOR
    +            ],  # Contributor role has CLIENT_CREATE but not administrative scopes
    +            scopes=[],  # No direct scopes, only role-derived
    +            user_id=user.id,
    +        )
    +
    +        # Create auth header for this role-based client
    +        auth_header = generate_role_header_for_user(user, [CONTRIBUTOR])
    +
    +        # Try to create a client with admin scopes that CONTRIBUTOR role doesn't have
    +        admin_scopes = [
    +            "storage:delete",  # Admin scope not in CONTRIBUTOR role
    +            "messaging:delete",  # Admin scope not in CONTRIBUTOR role
    +        ]
    +
    +        response = api_client.post(
    +            create_client_url,
    +            headers=auth_header,
    +            json=admin_scopes,
    +        )
    +
    +        # Should be blocked because CONTRIBUTOR role doesn't include these admin scopes
    +        assert response.status_code == 403
    +        response_data = response.json()
    +        assert "Cannot assign scopes that you do not have" in response_data["detail"]
    +
    +        # Clean up
    +        requesting_client.delete(db)
    +        user.delete(db)
    +
    +    def test_create_client_with_role_derived_scopes_allowed(
    +        self,
    +        db,
    +        api_client: TestClient,
    +        create_client_url,
    +    ) -> None:
    +        """Test that clients with roles can assign scopes they have via role inheritance."""
    +        from fides.api.models.client import ClientDetail
    +        from fides.api.models.fides_user import FidesUser
    +        from fides.api.oauth.roles import CONTRIBUTOR, ROLES_TO_SCOPES_MAPPING
    +        from tests.conftest import generate_role_header_for_user
    +
    +        # Create a user and client with CONTRIBUTOR role
    +        user = FidesUser.create(
    +            db=db,
    +            data={
    +                "username": "test_role_user",
    +                "first_name": "Test",
    +                "last_name": "User",
    +            },
    +        )
    +
    +        requesting_client, _ = ClientDetail.create_client_and_secret(
    +            db,
    +            CONFIG.security.oauth_client_id_length_bytes,
    +            CONFIG.security.oauth_client_secret_length_bytes,
    +            roles=[CONTRIBUTOR],
    +            scopes=[],  # No direct scopes, only role-derived
    +            user_id=user.id,  # Associate client with user
    +        )
    +
    +        # Get a scope that CONTRIBUTOR role has
    +        contributor_scopes = ROLES_TO_SCOPES_MAPPING[CONTRIBUTOR]
    +        # Pick a safe scope that CONTRIBUTOR has
    +        allowed_scope = CLIENT_READ  # CONTRIBUTOR should have this
    +        assert (
    +            allowed_scope in contributor_scopes
    +        ), f"{allowed_scope} should be in CONTRIBUTOR scopes"
    +
    +        # Create auth header using role-based authentication
    +        auth_header = generate_role_header_for_user(user, [CONTRIBUTOR])
    +
    +        # Try to create a client with a scope that CONTRIBUTOR role DOES have
    +        response = api_client.post(
    +            create_client_url,
    +            headers=auth_header,
    +            json=[allowed_scope],
    +        )
    +
    +        # Should succeed because CONTRIBUTOR role includes this scope
    +        assert response.status_code == 200
    +        response_data = response.json()
    +        assert "client_id" in response_data
    +        assert "client_secret" in response_data
    +
    +        # Verify the new client was created with the correct scope
    +        new_client = ClientDetail.get(
    +            db, object_id=response_data["client_id"], config=CONFIG
    +        )
    +        assert new_client.scopes == [allowed_scope]
    +
    +        # Clean up
    +        requesting_client.delete(db)
    +        new_client.delete(db)
    +        user.delete(db)
    +
    +    def test_get_client_effective_scopes_utility(
    +        self,
    +        db,
    +    ) -> None:
    +        """Test the get_client_effective_scopes utility function directly."""
    +        from fides.api.models.client import ClientDetail
    +        from fides.api.oauth.roles import CONTRIBUTOR, ROLES_TO_SCOPES_MAPPING
    +        from fides.api.oauth.utils import get_client_effective_scopes
    +
    +        # Test 1: Client with only direct scopes
    +        client_direct_only, _ = ClientDetail.create_client_and_secret(
    +            db,
    +            CONFIG.security.oauth_client_id_length_bytes,
    +            CONFIG.security.oauth_client_secret_length_bytes,
    +            scopes=[CLIENT_READ, CLIENT_UPDATE],
    +            roles=[],
    +        )
    +
    +        effective_scopes = get_client_effective_scopes(client_direct_only)
    +        assert set(effective_scopes) == {CLIENT_READ, CLIENT_UPDATE}
    +
    +        # Test 2: Client with only roles
    +        client_roles_only, _ = ClientDetail.create_client_and_secret(
    +            db,
    +            CONFIG.security.oauth_client_id_length_bytes,
    +            CONFIG.security.oauth_client_secret_length_bytes,
    +            scopes=[],
    +            roles=[CONTRIBUTOR],
    +        )
    +
    +        effective_scopes = get_client_effective_scopes(client_roles_only)
    +        expected_scopes = ROLES_TO_SCOPES_MAPPING[CONTRIBUTOR]
    +        assert set(effective_scopes) == set(expected_scopes)
    +
    +        # Test 3: Client with both direct scopes and roles
    +        client_mixed, _ = ClientDetail.create_client_and_secret(
    +            db,
    +            CONFIG.security.oauth_client_id_length_bytes,
    +            CONFIG.security.oauth_client_secret_length_bytes,
    +            scopes=[CLIENT_DELETE],  # Direct scope
    +            roles=[CONTRIBUTOR],  # Role-derived scopes
    +        )
    +
    +        effective_scopes = get_client_effective_scopes(client_mixed)
    +        expected_scopes = set([CLIENT_DELETE] + ROLES_TO_SCOPES_MAPPING[CONTRIBUTOR])
    +        assert set(effective_scopes) == expected_scopes
    +
    +        # Clean up
    +        client_direct_only.delete(db)
    +        client_roles_only.delete(db)
    +        client_mixed.delete(db)
    +
    +    def test_create_client_with_valid_scopes_allowed(
    +        self,
    +        db,
    +        api_client: TestClient,
    +        create_client_url,
    +        generate_auth_header,
    +    ) -> None:
    +        """Test that a client can create another client with scopes they do have."""
    +        # Client has these scopes
    +        requester_scopes = [CLIENT_CREATE, CLIENT_READ, CLIENT_DELETE]
    +        auth_header = generate_auth_header(requester_scopes)
    +
    +        # Create client with subset of requester's scopes (should work)
    +        allowed_scopes = [CLIENT_READ]
    +
    +        response = api_client.post(
    +            create_client_url,
    +            headers=auth_header,
    +            json=allowed_scopes,
    +        )
    +
    +        assert response.status_code == 200
    +        response_data = response.json()
    +        assert "client_id" in response_data
    +        assert "client_secret" in response_data
    +
    +        # Verify the client was created with correct scopes
    +        new_client = ClientDetail.get(
    +            db, object_id=response_data["client_id"], config=CONFIG
    +        )
    +        assert new_client.scopes == allowed_scopes
    +
    +        # Cleanup
    +        new_client.delete(db)
    +
    +    def test_set_client_scopes_privilege_escalation_blocked(
    +        self,
    +        db,
    +        api_client: TestClient,
    +        oauth_client,
    +        generate_auth_header,
    +    ) -> None:
    +        """Test that a client cannot set scopes they don't have on another client."""
    +        # Client only has CLIENT_UPDATE scope
    +        auth_header = generate_auth_header([CLIENT_UPDATE])
    +
    +        url = V1_URL_PREFIX + CLIENT_SCOPE.format(client_id=oauth_client.id)
    +
    +        # Try to set owner-level scopes that the requester doesn't have
    +        privileged_scopes = [
    +            "storage:create_or_update",
    +            "messaging:create_or_update",
    +            "user-permission:assign_owners",  # Note: hyphen, not underscore
    +        ]
    +
    +        response = api_client.put(
    +            url,
    +            headers=auth_header,
    +            json=privileged_scopes,
    +        )
    +
    +        assert response.status_code == 403
    +        response_data = response.json()
    +        assert "Cannot assign scopes that you do not have" in response_data["detail"]
    +
    +    def test_set_client_scopes_with_valid_scopes_allowed(
    +        self,
    +        db,
    +        api_client: TestClient,
    +        generate_auth_header,
    +    ) -> None:
    +        """Test that a client can set scopes they do have on another client."""
    +        # Create a fresh client with limited scopes for this test
    +        target_client, _ = ClientDetail.create_client_and_secret(
    +            db,
    +            CONFIG.security.oauth_client_id_length_bytes,
    +            CONFIG.security.oauth_client_secret_length_bytes,
    +            scopes=[CLIENT_READ, CLIENT_DELETE],  # Initial scopes
    +        )
    +
    +        # Client has these scopes
    +        requester_scopes = [CLIENT_UPDATE, CLIENT_READ, CLIENT_DELETE]
    +        auth_header = generate_auth_header(requester_scopes)
    +
    +        url = V1_URL_PREFIX + CLIENT_SCOPE.format(client_id=target_client.id)
    +
    +        # Set scopes that are subset of requester's scopes
    +        allowed_scopes = [CLIENT_READ]
    +
    +        response = api_client.put(
    +            url,
    +            headers=auth_header,
    +            json=allowed_scopes,
    +        )
    +
    +        assert response.status_code == 200
    +
    +        # Verify the scopes were updated
    +        db.refresh(target_client)  # Refresh to get updated data from database
    +        assert target_client.scopes == allowed_scopes
    +
    +        # Cleanup
    +        target_client.delete(db)
    +
    +    def test_root_client_can_assign_any_scope(
    +        self,
    +        db,
    +        api_client: TestClient,
    +        create_client_url,
    +    ) -> None:
    +        """Test that the root client can still assign any scope (maintains functionality)."""
    +        # Use root client credentials
    +        data = {
    +            "client_id": CONFIG.security.oauth_root_client_id,
    +            "client_secret": CONFIG.security.oauth_root_client_secret,
    +        }
    +
    +        token_url = V1_URL_PREFIX + TOKEN
    +        token_response = api_client.post(token_url, data=data)
    +        jwt = token_response.json().get("access_token")
    +        auth_header = {"Authorization": "Bearer " + jwt}
    +
    +        # Try to assign any scope (should work for root client)
    +        privileged_scopes = [
    +            "storage:create_or_update",
    +            "messaging:create_or_update",
    +            "user-permission:assign_owners",  # Note: hyphen, not underscore
    +        ]
    +
    +        response = api_client.post(
    +            create_client_url,
    +            headers=auth_header,
    +            json=privileged_scopes,
    +        )
    +
    +        assert response.status_code == 200
    +        response_data = response.json()
    +
    +        # Verify the client was created with privileged scopes
    +        new_client = ClientDetail.get(
    +            db, object_id=response_data["client_id"], config=CONFIG
    +        )
    +        assert set(new_client.scopes) == set(privileged_scopes)
    +
    +        # Cleanup
    +        new_client.delete(db)
    +
    +    def test_empty_scopes_allowed(
    +        self,
    +        db,
    +        api_client: TestClient,
    +        create_client_url,
    +        generate_auth_header,
    +    ) -> None:
    +        """Test that creating a client with no scopes is allowed."""
    +        auth_header = generate_auth_header([CLIENT_CREATE])
    +
    +        response = api_client.post(
    +            create_client_url,
    +            headers=auth_header,
    +            json=[],  # Empty scopes
    +        )
    +
    +        assert response.status_code == 200
    +        response_data = response.json()
    +
    +        # Verify the client was created with no scopes
    +        new_client = ClientDetail.get(
    +            db, object_id=response_data["client_id"], config=CONFIG
    +        )
    +        assert new_client.scopes == []
    +
    +        # Cleanup
    +        new_client.delete(db)
    

Vulnerability mechanics

Generated by null/stub on May 9, 2026. Inputs: CWE entries + fix-commit diffs from this CVE's patches. Citations validated against bundle.

References

5

News mentions

0

No linked articles in our index yet.