midway branch on oauth refactoring

.env.example

@@ -57,6 +57,10 @@ DISALLOW_OLD_CLIENTS=True
 
 DISCORD_AUDIT_LOG_WEBHOOK=
 
+JWT_PUBLIC_KEY=
+JWT_PRIVATE_KEY=
+ROTATION_JWT_PRIVATE_KEY=
+
 # automatically share information with the primary
 # developer of bancho.py (https://github.com/cmyui)
 # for debugging & development purposes.

.github/workflows/test.yaml

@@ -31,6 +31,9 @@ env:
   DISALLOWED_PASSWORDS: ${{ vars.DISALLOWED_PASSWORDS }}
   DISALLOW_OLD_CLIENTS: ${{ vars.DISALLOW_OLD_CLIENTS }}
   DISCORD_AUDIT_LOG_WEBHOOK: ${{ vars.DISCORD_AUDIT_LOG_WEBHOOK }}
+  JWT_PUBLIC_KEY: ${{ vars.JWT_PUBLIC_KEY }}
+  JWT_PRIVATE_KEY: ${{ vars.JWT_PRIVATE_KEY }}
+  ROTATION_JWT_PRIVATE_KEY: ${{ vars.ROTATION_JWT_PRIVATE_KEY }}
   AUTOMATICALLY_REPORT_PROBLEMS: ${{ vars.AUTOMATICALLY_REPORT_PROBLEMS }}
   SSL_CERT_PATH: ${{ vars.SSL_CERT_PATH }}
   SSL_KEY_PATH: ${{ vars.SSL_KEY_PATH }}

app/api/v2/__init__.py

@@ -1,14 +1,16 @@
 # isort: dont-add-imports
 
 from typing import Any
+from typing import TypedDict
 
+import jwt
 from fastapi import APIRouter
 from fastapi import Depends
 from fastapi import HTTPException
 from fastapi import status
 
+from app import settings
 from app.api.v2.common.oauth import OAuth2Scheme
-from app.repositories import access_tokens as access_tokens_repo
 
 oauth2_scheme = OAuth2Scheme(
     authorizationUrl="/v2/oauth/authorize",
@@ -23,16 +25,45 @@
 )
 
 
-async def get_current_client(token: str = Depends(oauth2_scheme)) -> dict[str, Any]:
-    """Look up the token in the Redis-based token store"""
-    access_token = await access_tokens_repo.fetch_one(token)
-    if not access_token:
+class AuthorizationContext(TypedDict):
+    verified_claims: dict[str, Any]
+
+
+async def authenticate_api_request(
+    token: str = Depends(oauth2_scheme),
+) -> AuthorizationContext:
+    verified_claims: dict[str, Any] | None = None
+    try:
+        verified_claims = jwt.decode(
+            token,
+            settings.JWT_PRIVATE_KEY,
+            algorithms=["HS256"],
+            options={"require": ["exp", "nbf", "iss", "aud", "iat"]},
+        )
+    except jwt.InvalidTokenError:
+        pass
+
+    if verified_claims is None:
+        try:
+            verified_claims = jwt.decode(
+                token,
+                settings.ROTATION_JWT_PRIVATE_KEY,
+                algorithms=["HS256"],
+                options={"require": ["exp", "nbf", "iss", "aud", "iat"]},
+            )
+        except jwt.InvalidTokenError:
+            pass
+
+    if verified_claims is None:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Not authenticated",
             headers={"WWW-Authenticate": "Bearer"},
         )
-    return access_token
+
+    return AuthorizationContext(
+        verified_claims=verified_claims,
+    )
 
 
 from . import clans

app/api/v2/common/oauth.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import base64
+from typing import TypedDict
 
 from fastapi import Request
 from fastapi import status
@@ -60,10 +61,15 @@ async def __call__(self, request: Request) -> str | None:
         return param
 
 
+class BasicAuthCredentials(TypedDict):
+    client_id: str
+    client_secret: str
+
+
 # https://developer.zendesk.com/api-reference/sales-crm/authentication/requests/#client-authentication
 def get_credentials_from_basic_auth(
     request: Request,
-) -> dict[str, str | int] | None:
+) -> BasicAuthCredentials | None:
     authorization = request.headers.get("Authorization")
     scheme, param = get_authorization_scheme_param(authorization)
     if not authorization or scheme.lower() != "basic":

app/api/v2/models/oauth.py

@@ -9,20 +9,40 @@
 # input models
 
 
+class ClientCredentialsGrantData(BaseModel):
+    scope: str | None
+
+
+class AuthorizationCodeGrantData(BaseModel):
+    code: str
+    redirect_uri: str
+    client_id: str
+
+
+class RefreshGrantData(BaseModel):
+    refresh_token: str
+    scope: str | None
+
+
 # output models
 
 
 class GrantType(StrEnum):
     AUTHORIZATION_CODE = "authorization_code"
     CLIENT_CREDENTIALS = "client_credentials"
+    REFRESH_TOKEN = "refresh_token"
 
     # TODO: Add support for other grant types
 
 
+class TokenType(StrEnum):
+    BEARER = "Bearer"
+
+
 class Token(BaseModel):
     access_token: str
     refresh_token: str | None
     token_type: Literal["Bearer"]
     expires_in: int
     expires_at: datetime
-    scope: str
+    scope: str | None