improve: granular optional auth

Author: samedii

example/main.py

@@ -4,6 +4,7 @@
 from fastapi import Depends
 from fastapi import FastAPI
 from fastapi import HTTPException
+from fastapi import Security
 from fastapi import status
 from fastapi.middleware.cors import CORSMiddleware
 from starlette.responses import RedirectResponse
@@ -14,12 +15,15 @@
 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
-    audience="my-audience",  # optional, verification only
-    auto_error=False,  # optional
+    # audience="my-audience",  # optional, verification only
     idtoken_model=KeycloakIDToken,  # optional
 )
 
-app = FastAPI(title="Example", version="dev")
+app = FastAPI(
+    title="Example",
+    version="dev",
+    dependencies=[Depends(auth.oidc_scheme)],
+)
 
 # CORS errors instead of seeing internal exceptions
 # https://stackoverflow.com/questions/63606055/why-do-i-get-cors-error-reason-cors-request-did-not-succeed
@@ -37,16 +41,21 @@ def redirect_to_docs():
     return RedirectResponse(url="/docs")
 
 
-@app.get("/protected", dependencies=[Depends(auth.oidc_scheme)])
-def protected(id_token: Optional[KeycloakIDToken] = Depends(auth.authenticate_user)):
+@app.get("/protected")
+def protected(id_token: Optional[KeycloakIDToken] = Security(auth.authenticate_user())):
+    print(id_token)
     if id_token is None:
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
 
     return dict(message=f"You are {id_token.email}")
 
 
-@app.get("/mixed", dependencies=[Depends(auth.oidc_scheme)])
-def mixed(id_token: Optional[KeycloakIDToken] = Depends(auth.authenticate_user)):
+@app.get("/mixed")
+def mixed(
+    id_token: Optional[KeycloakIDToken] = Security(
+        auth.authenticate_user(auto_error=False)
+    ),
+):
     if id_token is None:
         return dict(message="You are not authenticated")
     else:

fastapi_oidc/auth.py

@@ -40,13 +40,12 @@ def test_auth(authenticated_user: AuthenticatedUser = Depends(authenticate_user)
 from fastapi_oidc import discovery
 from fastapi_oidc.types import IDToken
 
+# class AuthBearer(HTTPBearer):
+#     async def __call__(self, request: Request):
+#         return await super().__call__(request)
 
-class AuthBearer(HTTPBearer):
-    async def __call__(self, request: Request):
-        return await super().__call__(request)
 
-
-class EmptyOAuth2(OAuth2):
+class OAuth2Facade(OAuth2):
     async def __call__(self, request: Request) -> Optional[str]:
         return None
 
@@ -92,13 +91,15 @@ def __init__(
         )
 
         self.oidc_scheme = OpenIdConnect(
-            openIdConnectUrl=openid_connect_url, auto_error=auto_error
+            openIdConnectUrl=openid_connect_url,
+            auto_error=False,
         )
         self.password_scheme = OAuth2PasswordBearer(
             tokenUrl=self.discover.token_url(oidc_discoveries),
             scopes=scopes,
+            auto_error=False,
         )
-        self.implicit_scheme = EmptyOAuth2(
+        self.implicit_scheme = OAuth2Facade(
             flows=OAuthFlows(
                 implicit={
                     "authorizationUrl": self.discover.authorization_url(
@@ -108,22 +109,17 @@ def __init__(
                 }
             ),
             scheme_name="OAuth2ImplicitBearer",
-            auto_error=auto_error,
+            auto_error=False,
         )
         self.authcode_scheme = OAuth2AuthorizationCodeBearer(
             authorizationUrl=self.discover.authorization_url(oidc_discoveries),
             tokenUrl=self.discover.token_url(oidc_discoveries),
             # refreshUrl=self.discover.refresh_url(oidc_discoveries),
             scopes=scopes,
+            auto_error=False,
         )
 
-    def authenticate_user(
-        self,
-        security_scopes: SecurityScopes,
-        authorization_credentials: Optional[HTTPAuthorizationCredentials] = Depends(
-            AuthBearer(auto_error=False)
-        ),
-    ) -> Optional[IDToken]:
+    def authenticate_user(self, auto_error=None):
         """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.
@@ -133,53 +129,64 @@ def authenticate_user(
                 scenes by Depends.
 
         Return:
-            Dict: Dictionary with IDToken information
+            IDToken: Dictionary with IDToken information
 
         raises:
             HTTPException(status_code=401, detail=f"Unauthorized: {err}")
         """
 
-        if authorization_credentials is None:
-            if self.auto_error:
-                raise HTTPException(
-                    status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token"
-                )
-            else:
-                return None
+        if auto_error is None:
+            auto_error = self.auto_error
 
-        oidc_discoveries = self.discover.auth_server(
-            openid_connect_url=self.openid_connect_url
-        )
-        key = self.discover.public_keys(oidc_discoveries)
-        algorithms = self.discover.signing_algos(oidc_discoveries)
-
-        try:
-            id_token = jwt.decode(
-                authorization_credentials.credentials,
-                key,
-                algorithms,
-                audience=self.audience,
-                issuer=self.issuer,
-                options={
-                    # Disabled at_hash check since we aren't using the access token
-                    "verify_at_hash": False,
-                    "verify_iss": self.issuer is not None,
-                    "verify_aud": self.audience is not None,
-                },
+        def authenticate_user_(
+            security_scopes: SecurityScopes,
+            authorization_credentials: Optional[HTTPAuthorizationCredentials] = Depends(
+                HTTPBearer(auto_error=auto_error)
+            ),
+        ) -> Optional[IDToken]:
+            if authorization_credentials is None:
+                if auto_error:
+                    raise HTTPException(
+                        status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token"
+                    )
+                else:
+                    return None
+
+            oidc_discoveries = self.discover.auth_server(
+                openid_connect_url=self.openid_connect_url
             )
-        except (ExpiredSignatureError, JWTError, JWTClaimsError) as err:
-            if self.auto_error:
-                raise HTTPException(status_code=401, detail=f"Unauthorized: {err}")
-            else:
-                return None
-
-        if not set(security_scopes.scopes).issubset(id_token["scope"].split(" ")):
-            if self.auto_error:
-                raise HTTPException(
-                    status.HTTP_401_UNAUTHORIZED,
-                    detail=f"""Missing scope token, only have {id_token["scopes"]}""",
+            key = self.discover.public_keys(oidc_discoveries)
+            algorithms = self.discover.signing_algos(oidc_discoveries)
+
+            try:
+                id_token = jwt.decode(
+                    authorization_credentials.credentials,
+                    key,
+                    algorithms,
+                    audience=self.audience,
+                    issuer=self.issuer,
+                    options={
+                        # Disabled at_hash check since we aren't using the access token
+                        "verify_at_hash": False,
+                        "verify_iss": self.issuer is not None,
+                        "verify_aud": self.audience is not None,
+                    },
                 )
-            else:
-                return None
-
-        return self.idtoken_model(**id_token)
+            except (ExpiredSignatureError, JWTError, JWTClaimsError) as err:
+                if auto_error:
+                    raise HTTPException(status_code=401, detail=f"Unauthorized: {err}")
+                else:
+                    return None
+
+            if not set(security_scopes.scopes).issubset(id_token["scope"].split(" ")):
+                if auto_error:
+                    raise HTTPException(
+                        status.HTTP_401_UNAUTHORIZED,
+                        detail=f"""Missing scope token, only have {id_token["scopes"]}""",
+                    )
+                else:
+                    return None
+
+            return self.idtoken_model(**id_token)
+
+        return authenticate_user_

fastapi_oidc/types.py

@@ -51,12 +51,8 @@ class KeycloakIDToken(IDToken):
     """Pydantic Model for the IDToken returned by Keycloak's OIDC implementation."""
 
     auth_time: int
-    ver: int
     jti: str
-    amr: List[str]
-    idp: str
-    nonce: str
-    at_hash: str
     name: str
     email: str
+    email_verified: bool
     preferred_username: str