improve: verify scopes

Author: samedii

.pre-commit-config.yaml

@@ -42,14 +42,3 @@ repos:
     hooks:
       - id: bandit
         entry: bandit -c .bandit.yml
-
-  - repo: local
-    hooks:
-      - id: pytest
-        name: pytest
-        entry: .venv/bin/pytest
-        language: script
-        pass_filenames: false
-        # alternatively you could `types: [python]` so it only runs when python files change
-        # though tests might be invalidated if you were to say change a data file
-        always_run: true

README.md

@@ -65,16 +65,16 @@ auth = Auth(
     openid_connect_url="http://localhost:8080/auth/realms/my-realm/.well-known/openid-configuration",
     issuer="http://localhost:8080/auth/realms/my-realm",  # optional, verification only
     client_id="my-client",  # optional, verification only
-    idtoken_model=KeycloakIDToken,  # optional
+    scopes=["email"],  # optional, verification only
+    idtoken_model=KeycloakIDToken,  # optional, verification only
 )
 
 app = FastAPI(
     title="Example",
     version="dev",
-    dependencies=[Depends(auth.implicit_scheme)],
+    dependencies=[Depends(auth.implicit_scheme)],  # multiple schemes available
 )
 
-
 @app.get("/protected")
 def protected(id_token: KeycloakIDToken = Security(auth.required)):
     return dict(message=f"You are {id_token.email}")

example/main.py

@@ -15,7 +15,8 @@
     openid_connect_url="http://localhost:8080/auth/realms/my-realm/.well-known/openid-configuration",
     issuer="http://localhost:8080/auth/realms/my-realm",  # optional, verification only
     client_id="my-client",  # optional, verification only
-    idtoken_model=KeycloakIDToken,  # optional
+    scopes=["email"],  # optional, verification only
+    idtoken_model=KeycloakIDToken,  # optional, verification only
 )
 
 app = FastAPI(
@@ -46,9 +47,7 @@ def protected(id_token: KeycloakIDToken = Security(auth.required)):
 
 
 @app.get("/mixed")
-def mixed(
-    id_token: Optional[KeycloakIDToken] = Security(auth.optional),
-):
+def mixed(id_token: Optional[KeycloakIDToken] = Security(auth.optional)):
     if id_token is None:
         return dict(message="You are not authenticated")
     else:

fastapi_oidc/auth.py

@@ -16,7 +16,7 @@ def test_auth(authenticated_user: AuthenticatedUser = Depends(authenticate_user)
         return f"Hello {name}"
 """
 
-from typing import Dict
+from typing import List
 from typing import Optional
 from typing import Type
 
@@ -52,11 +52,11 @@ def __init__(
         openid_connect_url: str,
         issuer: Optional[str] = None,
         client_id: Optional[str] = None,
-        scopes: Dict[str, str] = dict(),
+        scopes: List[str] = list(),
         signature_cache_ttl: int = 3600,
         idtoken_model: Type = IDToken,
     ):
-        """Configure authentication and use method :func:`authenticate_user`
+        """Configure authentication and use method :func:`require` or :func:`optional`
         to check user credentials.
 
         Args:
@@ -77,19 +77,23 @@ def __init__(
         self.issuer = issuer
         self.client_id = client_id
         self.idtoken_model = idtoken_model
+        self.scopes = scopes
 
         self.discover = discovery.configure(cache_ttl=signature_cache_ttl)
         oidc_discoveries = self.discover.auth_server(
             openid_connect_url=self.openid_connect_url
         )
+        scopes_dict = {
+            scope: "" for scope in self.discover.supported_scopes(oidc_discoveries)
+        }
 
         self.oidc_scheme = OpenIdConnect(
             openIdConnectUrl=openid_connect_url,
             auto_error=False,
         )
         self.password_scheme = OAuth2PasswordBearer(
             tokenUrl=self.discover.token_url(oidc_discoveries),
-            scopes=scopes,
+            scopes=scopes_dict,
             auto_error=False,
         )
         self.implicit_scheme = OAuth2Facade(
@@ -98,7 +102,7 @@ def __init__(
                     "authorizationUrl": self.discover.authorization_url(
                         oidc_discoveries
                     ),
-                    "scopes": scopes,
+                    "scopes": scopes_dict,
                 }
             ),
             scheme_name="OAuth2ImplicitBearer",
@@ -108,7 +112,7 @@ def __init__(
             authorizationUrl=self.discover.authorization_url(oidc_discoveries),
             tokenUrl=self.discover.token_url(oidc_discoveries),
             # refreshUrl=self.discover.refresh_url(oidc_discoveries),
-            scopes=scopes,
+            scopes=scopes_dict,
             auto_error=False,
         )
 
@@ -118,7 +122,7 @@ def required(
         authorization_credentials: Optional[HTTPAuthorizationCredentials] = Depends(
             HTTPBearer()
         ),
-    ) -> Optional[IDToken]:
+    ) -> IDToken:
         """Validate and parse OIDC ID token against issuer in config.
         Note this function caches the signatures and algorithms of the issuing
         server for signature_cache_ttl seconds.
@@ -136,11 +140,15 @@ def required(
             IDToken validation errors
         """
 
-        return self.authenticate_user(
+        id_token = self.authenticate_user(
             security_scopes,
             authorization_credentials,
             auto_error=True,
         )
+        if id_token is None:
+            raise HTTPException(status.HTTP_401_UNAUTHORIZED)
+        else:
+            return id_token
 
     def optional(
         self,
@@ -230,21 +238,21 @@ def authenticate_user(
                         raise JWTError(
                             f"""Invalid authorized party "azp": {id_token["azp"]}"""
                         )
-                    elif type(token_audience) == list and len(token_audience) >= 1:
-                        raise JWTError('Missing authorized party "azp" in IDToken')
+                elif type(token_audience) == list and len(token_audience) >= 1:
+                    raise JWTError('Missing authorized party "azp" in IDToken')
 
         except (ExpiredSignatureError, JWTError, JWTClaimsError) as error:
             raise HTTPException(status_code=401, detail=f"Unauthorized: {error}")
 
-        if not set(security_scopes.scopes).issubset(
-            id_token.get("scope", "").split(" ")
-        ):
-            if auto_error:
-                raise HTTPException(
-                    status.HTTP_401_UNAUTHORIZED,
-                    detail=f"""Missing scope token, only have {id_token["scopes"]}""",
-                )
-            else:
-                return None
+        expected_scopes = set(self.scopes + security_scopes.scopes)
+        token_scopes = id_token.get("scope", "").split(" ")
+        if not expected_scopes.issubset(token_scopes):
+            raise HTTPException(
+                status.HTTP_401_UNAUTHORIZED,
+                detail=(
+                    f"Missing scope token, expected {expected_scopes} to be a "
+                    f"subset of received {token_scopes}",
+                ),
+            )
 
         return self.idtoken_model(**id_token)

fastapi_oidc/discovery.py

@@ -35,11 +35,15 @@ def get_authorization_url(OIDC_spec: Dict) -> str:
     def get_token_url(OIDC_spec: Dict) -> str:
         return OIDC_spec["token_endpoint"]
 
+    def get_supported_scopes(OIDC_spec: Dict) -> str:
+        return OIDC_spec["scopes_supported"]
+
     class functions:
         auth_server = discover_auth_server
         public_keys = get_authentication_server_public_keys
         signing_algos = get_signing_algos
         authorization_url = get_authorization_url
         token_url = get_token_url
+        supported_scopes = get_supported_scopes
 
     return functions

fastapi_oidc/types.py

@@ -51,7 +51,6 @@ class OktaIDToken(IDToken):
 class KeycloakIDToken(IDToken):
     """Pydantic Model for the IDToken returned by Keycloak's OIDC implementation."""
 
-    auth_time: int
     jti: str
     name: str
     email: str

tests/conftest.py

@@ -139,5 +139,6 @@ class functions:
         signing_algos = lambda x: x["id_token_signing_alg_values_supported"]
         authorization_url = lambda x: x["authorization_endpoint"]
         token_url = lambda x: x["token_endpoint"]
+        supported_scopes = lambda x: x["scopes_supported"]
 
     return lambda *args, **kwargs: functions