User Endpoints Implementation

.github/workflows/build-ecr.yml

@@ -7,7 +7,7 @@ on:
 
 permissions:
   contents: read
-  id-token: write  # Required for OIDC
+  id-token: write # Required for OIDC
   actions: read
 
 env:

.github/workflows/deploy.yml

@@ -58,4 +58,4 @@ jobs:
           EOF
 
       - name: CDK Deploy
-        run: cdk deploy --require-approval never
+        run: cdk deploy --require-approval never

README.md

@@ -53,8 +53,8 @@ uv run pytest
 
 # Deployment
 
-Currently the service is deployed to AWS via the CDK scripts in `deploy/`, 
+Currently the service is deployed to AWS via the CDK scripts in `deploy/`,
 and updated on each commit to `main`.
 
 Secrets/configuration variables for the deployment are stored in the
-GitHub Secrets for the repository.
+GitHub Secrets for the repository.

auth/config.py

@@ -14,4 +14,4 @@ class Settings(BaseSettings):
 
 @lru_cache()
 def get_settings():
-    return Settings()
+    return Settings()

auth/management.py

@@ -5,13 +5,13 @@
 
 def get_management_token():
     settings = get_settings()
-    url = f'https://{settings.auth0_domain}/oauth/token'
+    url = f"https://{settings.auth0_domain}/oauth/token"
     payload = {
-        'grant_type': 'client_credentials',
-        'client_id': settings.auth0_management_id,
-        'client_secret': settings.auth0_management_secret,
-        'audience': f'https://{settings.auth0_domain}/api/v2/'
+        "grant_type": "client_credentials",
+        "client_id": settings.auth0_management_id,
+        "client_secret": settings.auth0_management_secret,
+        "audience": f"https://{settings.auth0_domain}/api/v2/",
     }
     response = httpx.post(url, json=payload)
     response.raise_for_status()
-    return response.json()['access_token']
+    return response.json()["access_token"]

auth/validator.py

@@ -1,25 +1,14 @@
-from typing import Dict
-
 import httpx
-from fastapi import HTTPException
-from jose import jwt, jwk
-from jose.exceptions import JWTError
 
 from auth.config import get_settings
+from fastapi import Depends, HTTPException
+from fastapi.security import OAuth2PasswordBearer
+from jose import jwt, jwk
+from jose.exceptions import JWTError
 from schemas.tokens import AccessTokenPayload
+from schemas.user import User
 
-
-def get_rsa_key(token: str) -> jwk.RSAKey | None:
-    settings = get_settings()
-    jwks_url = f"https://{settings.auth0_domain}/.well-known/jwks.json"
-    response = httpx.get(jwks_url)
-    jwks = response.json()
-    unverified_header = jwt.get_unverified_header(token)
-
-    for key in jwks["keys"]:
-        if key["kid"] == unverified_header["kid"]:
-            return jwk.construct(key)
-    return None
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
 
 def verify_jwt(token: str) -> AccessTokenPayload:
@@ -28,26 +17,54 @@ def verify_jwt(token: str) -> AccessTokenPayload:
         rsa_key = get_rsa_key(token)
     except JWTError as e:
         raise HTTPException(status_code=401, detail=f"Invalid token: {e}")
+
     if rsa_key is None:
-        raise HTTPException(status_code=401, detail="Couldn't find a matching signing key.")
+        raise HTTPException(
+            status_code=401, detail="Couldn't find a matching signing key."
+        )
 
     try:
         payload = jwt.decode(
             token,
             rsa_key,
             algorithms=settings.auth0_algorithms,
             audience=settings.auth0_audience,
-            issuer=f"https://{settings.auth0_domain}/"
+            issuer=f"https://{settings.auth0_domain}/",
         )
     except JWTError as e:
         raise HTTPException(status_code=401, detail=f"Invalid token: {e}")
 
     roles_claim = "biocommons.org.au/roles"
     if roles_claim not in payload:
-        raise HTTPException(status_code=403, detail=f"Missing required claim: {roles_claim}")
+        raise HTTPException(
+            status_code=403, detail=f"Missing required claim: {roles_claim}"
+        )
 
     roles = payload[roles_claim]
-    if not isinstance(roles, list) or not any("admin" in role.lower() for role in roles):
-        raise HTTPException(status_code=403, detail=f"Access denied: Insufficient permissions")
+    if not isinstance(roles, list) or not any(
+        "admin" in role.lower() for role in roles
+    ):
+        raise HTTPException(
+            status_code=403, detail="Access denied: Insufficient permissions"
+        )
 
     return AccessTokenPayload(**payload)
+
+
+def get_rsa_key(token: str) -> jwk.RSAKey | None:  # type: ignore
+    settings = get_settings()
+    jwks_url = f"https://{settings.auth0_domain}/.well-known/jwks.json"
+    response = httpx.get(jwks_url)
+    jwks = response.json()
+    unverified_header = jwt.get_unverified_header(token)
+
+    for key in jwks["keys"]:
+        if key["kid"] == unverified_header["kid"]:
+            return jwk.construct(key)
+
+    return None
+
+
+def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
+    access_token = verify_jwt(token)
+    return User(access_token=access_token)

main.py

@@ -1,22 +1,12 @@
-from fastapi import Depends, FastAPI
-from fastapi.security import OAuth2PasswordBearer
-
-from auth.management import get_management_token
-from auth.validator import verify_jwt
-from schemas.user import User
+from fastapi import FastAPI
+from routers import user
 
 app = FastAPI()
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
-
 
-def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
-    access_token = verify_jwt(token)
-    return User(access_token=access_token)
 
 @app.get("/")
 def public_route():
-    return {"message": "Public route"}
+    return {"message": "AAI Backend API"}
+
 
-@app.get("/private")
-def private_route(user=Depends(get_current_user)):
-    return {"message": "Private route", "user_claims": user, "management_token": get_management_token()}
+app.include_router(user.router)