Release v2.5.0 (#22)

* fix: settings parser type error

* feat: handle unhandled exceptions in token decode method

* Remove cognito-jwt and move to joserfc

* Update exception message and bump version

* Update README.md

* Change error message formatting syntax

* Update pyproject.toml

README.md

@@ -1,7 +1,7 @@
 # FastAPI - Cognito
-FastAPI package that ease usage of AWS Cognito Auth.
-This package provides basic functions/tools which helps developers to use
-Cognito JWT.
+FastAPI library that ease usage of AWS Cognito Auth.
+This library provides basic functionalities for decoding, validation and parsing
+Cognito JWT tokens and for now it does not support sign up and sign in features.
 
 ## Requirements
 
@@ -27,13 +27,12 @@ app = FastAPI()
 All mandatory fields are added in CognitoSettings
 BaseSettings object. Settings can be added in different ways.
 You can provide all required settings in **.yaml** or **.json** files,
-or your global BaseSettings file. Note that userpools field is Dict,
+or your global BaseSettings object. Note that `userpools` field is Dict and
 **FIRST** user pool in a dict will be set as default automatically if
-userpool_name is not provided in CognitoAuth object.
+`userpool_name` is not provided in CognitoAuth object.
 All fields shown in example below, are also required in .json or .yaml file
 (with syntax matching those files.)
 
-You should also import BaseSettings from pydantic if you are going to use global BaseSettings object.
 * Provide settings that are mandatory for CognitoAuth to work. You can provide
 one or more userpools.
   * `app_client_id` field for userpool besides string, can contain multiple string values provided within 
@@ -66,8 +65,8 @@ settings = Settings()
 
 This example below shows how global BaseSettings object can be mapped to
 CognitoSettings and passed as param to CognitoAuth.
-If we were using .yaml or .json, we should call **.from_yaml(_filename_)** or
-**.from_json(_filename_)** methods on CognitoSettings object.
+If we were using .yaml or .json, we should call **.from_yaml(_path_)** or
+**.from_json(_path_)** methods on CognitoSettings object.
 
 * Instantiate CognitoAuth and pass previously created settings as settings param.
 
@@ -83,10 +82,8 @@ cognito_us = CognitoAuth(
 )
 ```
 
-* This is a simple endpoint that is protected by Cognito, it uses FastAPI 
+* This is a simple endpoint that requires authentication, it uses FastAPI 
 dependency injection to resolve all required operations and get Cognito JWT.
-It can be used later to add more security to endpoints and to get required
-data about user which token belongs to.
 
 ```python
 from fastapi_cognito import CognitoToken
@@ -112,11 +109,13 @@ def hello_world(auth: CognitoToken = Depends(cognito_eu.auth_optional)):
 ```
 
 ### Custom Token Model
+This feature adds possiblity to use any token type for authentication(e.g. parsing ID token).
 
 In case your token payload contains additional values, you can provide custom
 token model instead of `CognitoToken`. If there is no custom token model
 provided, `CognitoToken` will be set as a default model. Custom model should
-be provided to `CognitoAuth` object.
+be provided to `CognitoAuth` object, and should be set as type of `auth` 
+variable for endpoint dependency.
 
 Example:
 ```python
@@ -131,6 +130,41 @@ cognito = CognitoAuth(
 )
 
 @app.get("/")
+# Type of `auth` should be custom token Class
 def hello_world(auth: CustomTokenModel = Depends(cognito.auth_required)):
     return {"message": f"Hello {auth.custom_value}"}
 ```
+
+#### Custom Cognito attributes
+Custom attributes in Cognito starts with `custom:`, which is the issue for 
+parsing this variable with pydantic because of the colon. To parse custom 
+attributes, add the full name of Cognito attribute to Pydantic Field alias.
+
+```python
+class CustomTokenModel(CognitoToken):
+    custom_value: Optional[str] = Field(alias="custom:custom_attr")
+```
+Pydantic will automatically parse value by alias if specified. Make sure that
+you have default value set if attribute is optional.
+
+### OpenAPI docs authentication 
+To use tokens to authenticate requests using OpenAPI docs, you can
+create wrapper class. 
+```python
+from fastapi.security import HTTPBearer
+from starlette.requests import Request
+from fastapi_cognito import CognitoToken
+
+class CognitoAuth(HTTPBearer):
+    async def __call__(self, request: Request) -> CognitoToken:
+        return await cognito.auth_required(request=request)
+
+cognito_auth = CognitoAuth()
+
+@router.get("/")
+async def test_endpoint(auth: CognitoToken = Depends(cognito_auth)):
+    return JSONResponse(
+        status_code=200, content={"detail": "Success"}
+    )
+```
+This will show button for adding authentication token to the request.

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "fastapi-cognito"
-version = "2.4.2"
-description = "Basic AWS cognito authentication package for FastAPI"
+version = "2.5.0"
+description = "AWS Cognito JWT authentication library for FastAPI"
 authors = ["Marko Mirosavljev <mirosavljevm023@gmail.com>"]
 maintainers = ["Marko Mirosavljev <mirosavljevm023@gmail.com>"]
 license = "MIT"
@@ -28,7 +28,10 @@ fastapi = "^0.111.0"
 pydantic = "^2.7.1"
 pydantic-settings = "^2.2.1"
 pyyaml = "^6.0.1"
-cognitojwt = { version = "1.4.1", extras = ["async"] }
+aiohttp = "^3.9.5"
+aiofile = "^3.8.8"
+async-lru = "^2.0.4"
+joserfc = "^0.9.0"
 
 
 [build-system]

src/fastapi_cognito/cognito_jwt/constants.py

@@ -0,0 +1 @@
+PUBLIC_KEYS_URL_TEMPLATE = 'https://cognito-idp.{}.amazonaws.com/{}/.well-known/jwks.json'

src/fastapi_cognito/cognito_jwt/decode.py

@@ -0,0 +1,90 @@
+import json
+import os
+from typing import Dict, Container, Optional, Union, List, Mapping
+
+import aiohttp
+from aiofile import AIOFile
+from async_lru import alru_cache
+from joserfc import jwk, jwt
+from joserfc.errors import BadSignatureError
+
+from fastapi_cognito.cognito_jwt.constants import PUBLIC_KEYS_URL_TEMPLATE
+from fastapi_cognito.cognito_jwt.exceptions import CognitoJWTException
+from fastapi_cognito.cognito_jwt.utils import check_expired, check_client_id, \
+    get_unverified_token_header
+
+
+@alru_cache(maxsize=1)
+async def __get_keys_async(keys_url: str) -> List[dict]:
+    """
+    Retrieves public keys from AWS Cognito or read from file
+
+    :return: List of public keys
+    """
+    if keys_url.startswith("http"):
+        async with aiohttp.ClientSession() as session:
+            async with session.get(keys_url) as resp:
+                response = await resp.json()
+    else:
+        async with AIOFile(keys_url, 'r') as afp:
+            f = await afp.read()
+            response = json.loads(f)
+    return response.get('keys')
+
+
+async def __get_public_key_async(token: str, region: str, userpool_id: str):
+    """
+    Get public key, verify that `kid` value matches value from token headers
+    and generate `joserfc._keys.Key` object
+
+    :return: `joserfc._keys.Key`
+    """
+    keys_url: str = (
+            os.environ.get("AWS_COGNITO_KEYS_URL") or
+            PUBLIC_KEYS_URL_TEMPLATE.format(region, userpool_id)
+    )
+    keys: list = await __get_keys_async(keys_url)
+
+    headers: Mapping[str, str] = get_unverified_token_header(token)
+    kid: str = headers["kid"]
+
+    key = list(filter(lambda k: k["kid"] == kid, keys))
+    if not key:
+        raise CognitoJWTException(
+            "Public key not found, check userpool configuration."
+        )
+    else:
+        key = key[0]
+
+    return jwk.JWKRegistry.import_key(key)
+
+
+async def decode_cognito_jwt(
+        token: str,
+        region: str,
+        userpool_id: str,
+        app_client_id: Optional[Union[str, Container[str]]] = None,
+        testmode: bool = False
+) -> Dict:
+    """
+    Retrieve public key, decode and validate JWT. Check if token is issued
+     for provided `app_client_id` and if it's expired.
+
+    :return: Dict with token claims.
+    """
+    public_key = await __get_public_key_async(token, region, userpool_id)
+
+    try:
+        decoded_claims = jwt.decode(token, public_key)
+    except BadSignatureError:
+        raise CognitoJWTException(
+            "Token signature verification failed."
+        )
+
+    claims = decoded_claims.claims
+    check_expired(claims["exp"], testmode=testmode)
+
+    if app_client_id:
+        check_client_id(claims, app_client_id)
+
+    return claims

src/fastapi_cognito/cognito_jwt/exceptions.py

@@ -0,0 +1,2 @@
+class CognitoJWTException(Exception):
+    pass