refactor!: new features and big refactor of library

chose between schemes, optional auth, reintroduce
idtoken_model, define scopes, and verify scope

BREAKING CHANGE

Author: samedii

.bandit.yml

@@ -82,4 +82,4 @@
 tests:
 
 # (optional) list skipped test IDs here, eg '[B101, B406]':
-skips: [B101]
+skips: [B101, B104]

example/__init__.py

@@ -0,0 +1,59 @@
+from typing import Optional
+
+import uvicorn
+from fastapi import Depends
+from fastapi import FastAPI
+from fastapi import HTTPException
+from fastapi import status
+from fastapi.middleware.cors import CORSMiddleware
+from starlette.responses import RedirectResponse
+
+from fastapi_oidc import Auth
+from fastapi_oidc import KeycloakIDToken
+
+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
+    idtoken_model=KeycloakIDToken,  # optional
+)
+
+app = FastAPI(title="Example", version="dev")
+
+# CORS errors instead of seeing internal exceptions
+# https://stackoverflow.com/questions/63606055/why-do-i-get-cors-error-reason-cors-request-did-not-succeed
+cors = CORSMiddleware(
+    app=app,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.get("/", status_code=status.HTTP_303_SEE_OTHER)
+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)):
+    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)):
+    if id_token is None:
+        return dict(message="You are not authenticated")
+    else:
+        return dict(message=f"You are {id_token.email}")
+
+
+if __name__ == "__main__":
+    uvicorn.run(
+        "example.main:cors", host="0.0.0.0", port=8000, loop="asyncio", reload=True
+    )

example/main.py

@@ -1,3 +1,4 @@
-from fastapi_oidc.auth import get_auth  # noqa
+from fastapi_oidc.auth import Auth  # noqa
 from fastapi_oidc.types import IDToken  # noqa
+from fastapi_oidc.types import KeycloakIDToken  # noqa
 from fastapi_oidc.types import OktaIDToken  # noqa

fastapi_oidc/__init__.py

@@ -16,52 +16,114 @@ def test_auth(authenticated_user: AuthenticatedUser = Depends(authenticate_user)
         return f"Hello {name}"
 """
 
-from typing import Callable
 from typing import Dict
 from typing import Optional
+from typing import Type
 
 from fastapi import Depends
 from fastapi import HTTPException
+from fastapi import Request
+from fastapi import status
+from fastapi.openapi.models import OAuthFlows
+from fastapi.security import HTTPAuthorizationCredentials
+from fastapi.security import HTTPBearer
+from fastapi.security import OAuth2
+from fastapi.security import OAuth2AuthorizationCodeBearer
+from fastapi.security import OAuth2PasswordBearer
 from fastapi.security import OpenIdConnect
+from fastapi.security import SecurityScopes
 from jose import ExpiredSignatureError
 from jose import JWTError
 from jose import jwt
 from jose.exceptions import JWTClaimsError
 
 from fastapi_oidc import discovery
+from fastapi_oidc.types import IDToken
 
 
-def get_auth(
-    openid_connect_url: str,
-    issuer: Optional[str] = None,
-    audience: Optional[str] = None,
-    signature_cache_ttl: int = 3600,
-) -> Callable[[str], Dict]:
-    """Take configurations and returns the :func:`authenticate_user` function.
+class AuthBearer(HTTPBearer):
+    async def __call__(self, request: Request):
+        return await super().__call__(request)
 
-    This function should only be invoked once at the beggining of your
-    server code. The function it returns should be used to check user credentials.
 
-    Args:
-        openid_connect_url (URL): URL to the "well known" openid connect config
-            e.g. https://dev-123456.okta.com/.well-known/openid-configuration
-        issuer (URL): (Optional) The issuer URL from your auth server.
-        audience (str): (Optional) The audience string configured by your auth server.
-        signature_cache_ttl (int): How many seconds your app should cache the
-            authorization server's public signatures.
+class EmptyOAuth2(OAuth2):
+    async def __call__(self, request: Request) -> Optional[str]:
+        return None
 
-    Returns:
-        func: authenticate_user(auth_header: str) -> Dict
 
-    Raises:
-        Nothing intentional
-    """
+class Auth:
+    def __init__(
+        self,
+        openid_connect_url: str,
+        issuer: Optional[str] = None,
+        audience: Optional[str] = None,
+        scopes: Dict[str, str] = dict(),
+        auto_error: bool = True,
+        signature_cache_ttl: int = 3600,
+        idtoken_model: Type = IDToken,
+    ):
+        """Configure authentication and use method :func:`authenticate_user`
+        to check user credentials.
 
-    oauth2_scheme = OpenIdConnect(openIdConnectUrl=openid_connect_url)
-
-    discover = discovery.configure(cache_ttl=signature_cache_ttl)
+        Args:
+            openid_connect_url (URL): URL to the "well known" openid connect config
+                e.g. https://dev-123456.okta.com/.well-known/openid-configuration
+            issuer (URL): (Optional) The issuer URL from your auth server.
+            audience (str): (Optional) The audience string configured by your auth server.
+            scopes (Dict[str, str]): (Optional) A dictionary of scopes and their descriptions.
+            auto_error (bool): (Optional) If True, raise an HTTPException if the token is invalid.
+            signature_cache_ttl (int): How many seconds your app should cache the
+                authorization server's public signatures.
+            idtoken_model (Type): (Optional) The model to use for validating the ID Token.
+
+        Raises:
+            Nothing intentional
+        """
 
-    def authenticate_user(auth_header: str = Depends(oauth2_scheme)) -> Dict:
+        self.openid_connect_url = openid_connect_url
+        self.issuer = issuer
+        self.audience = audience
+        self.auto_error = auto_error
+        self.idtoken_model = idtoken_model
+
+        self.discover = discovery.configure(cache_ttl=signature_cache_ttl)
+        oidc_discoveries = self.discover.auth_server(
+            openid_connect_url=self.openid_connect_url
+        )
+
+        self.oidc_scheme = OpenIdConnect(
+            openIdConnectUrl=openid_connect_url, auto_error=auto_error
+        )
+        self.password_scheme = OAuth2PasswordBearer(
+            tokenUrl=self.discover.token_url(oidc_discoveries),
+            scopes=scopes,
+        )
+        self.implicit_scheme = EmptyOAuth2(
+            flows=OAuthFlows(
+                implicit={
+                    "authorizationUrl": self.discover.authorization_url(
+                        oidc_discoveries
+                    ),
+                    "scopes": scopes,
+                }
+            ),
+            scheme_name="OAuth2ImplicitBearer",
+            auto_error=auto_error,
+        )
+        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,
+        )
+
+    def authenticate_user(
+        self,
+        security_scopes: SecurityScopes,
+        authorization_credentials: Optional[HTTPAuthorizationCredentials] = Depends(
+            AuthBearer(auto_error=False)
+        ),
+    ) -> Optional[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.
@@ -76,27 +138,48 @@ def authenticate_user(auth_header: str = Depends(oauth2_scheme)) -> Dict:
         raises:
             HTTPException(status_code=401, detail=f"Unauthorized: {err}")
         """
-        id_token = auth_header.split(" ")[-1]
-        OIDC_discoveries = discover.auth_server(openid_connect_url=openid_connect_url)
-        key = discover.public_keys(OIDC_discoveries)
-        algorithms = discover.signing_algos(OIDC_discoveries)
+
+        if authorization_credentials is None:
+            if self.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
+        )
+        key = self.discover.public_keys(oidc_discoveries)
+        algorithms = self.discover.signing_algos(oidc_discoveries)
 
         try:
-            return jwt.decode(
-                id_token,
+            id_token = jwt.decode(
+                authorization_credentials.credentials,
                 key,
                 algorithms,
-                audience=audience,
-                issuer=issuer,
+                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": issuer is not None,
-                    "verify_aud": audience is not None,
+                    "verify_iss": self.issuer is not None,
+                    "verify_aud": self.audience is not None,
                 },
             )
-
         except (ExpiredSignatureError, JWTError, JWTClaimsError) as err:
-            raise HTTPException(status_code=401, detail=f"Unauthorized: {err}")
-
-    return authenticate_user
+            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"]}""",
+                )
+            else:
+                return None
+
+        return self.idtoken_model(**id_token)

fastapi_oidc/auth.py

@@ -29,9 +29,17 @@ def discover_auth_server(*_, openid_connect_url: str) -> Dict:
         configuration = r.json()
         return configuration
 
+    def get_authorization_url(OIDC_spec: Dict) -> str:
+        return OIDC_spec["authorization_endpoint"]
+
+    def get_token_url(OIDC_spec: Dict) -> str:
+        return OIDC_spec["token_endpoint"]
+
     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
 
     return functions

fastapi_oidc/discovery.py

@@ -24,7 +24,7 @@ class IDToken(BaseModel):
 
     iss: str
     sub: str
-    aud: str
+    aud: List[str]
     exp: int
     iat: int
 
@@ -45,3 +45,18 @@ class OktaIDToken(IDToken):
     name: str
     email: str
     preferred_username: str
+
+
+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
+    preferred_username: str