feat(auth): add support for OAuth and MojeID

7project/backend/app/app.py

@@ -1,15 +1,16 @@
 from fastapi import Depends, FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
+import app.services.user_service
 from app.models.user import User
 
 from app.schemas.user import UserCreate, UserRead, UserUpdate
 from app.services.user_service import auth_backend, current_active_verified_user, fastapi_users
 
-app = FastAPI()
+fastApi = FastAPI()
 
 # CORS for frontend dev server
-app.add_middleware(
+fastApi.add_middleware(
     CORSMiddleware,
     allow_origins=[
         "http://localhost:5173",
@@ -20,37 +21,48 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
-app.include_router(
+fastApi.include_router(
     fastapi_users.get_auth_router(auth_backend), prefix="/auth/jwt", tags=["auth"]
 )
-app.include_router(
+fastApi.include_router(
     fastapi_users.get_register_router(UserRead, UserCreate),
     prefix="/auth",
     tags=["auth"],
 )
-app.include_router(
+fastApi.include_router(
     fastapi_users.get_reset_password_router(),
     prefix="/auth",
     tags=["auth"],
 )
-app.include_router(
+fastApi.include_router(
     fastapi_users.get_verify_router(UserRead),
     prefix="/auth",
     tags=["auth"],
 )
-app.include_router(
+fastApi.include_router(
     fastapi_users.get_users_router(UserRead, UserUpdate),
     prefix="/users",
     tags=["users"],
 )
 
+fastApi.include_router(
+    fastapi_users.get_oauth_router(
+        app.services.user_service.mojeid_oauth_service,
+        auth_backend,
+        "SECRET",
+        associate_by_email=True
+    ),
+    prefix="/auth/mojeid",
+    tags=["auth"],
+)
+
 
 # Liveness/root endpoint
-@app.get("/", include_in_schema=False)
+@fastApi.get("/", include_in_schema=False)
 async def root():
     return {"status": "ok"}
 
 
-@app.get("/authenticated-route")
+@fastApi.get("/authenticated-route")
 async def authenticated_route(user: User = Depends(current_active_verified_user)):
     return {"message": f"Hello {user.email}!"}

7project/backend/app/models/user.py

@@ -1,13 +1,20 @@
+from typing import List
+
 from sqlalchemy import Column, String
 from sqlalchemy.orm import relationship
-from fastapi_users.db import SQLAlchemyBaseUserTableUUID
+from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyBaseOAuthAccountTableUUID
 from app.core.base import Base
 
 
+class OAuthAccount(SQLAlchemyBaseOAuthAccountTableUUID, Base):
+    pass
+
+
 class User(SQLAlchemyBaseUserTableUUID, Base):
     first_name = Column(String(length=100), nullable=True)
     last_name = Column(String(length=100), nullable=True)
+    oauth_accounts = relationship("OAuthAccount", lazy="joined")
 
     # Relationship
     transactions = relationship("Transaction", back_populates="user")
-    categories = relationship("Category", back_populates="user")
+    categories = relationship("Category", back_populates="user")

7project/backend/app/oauth/moje_id.py

@@ -0,0 +1,48 @@
+import json
+from typing import Optional, Literal
+
+from httpx_oauth.clients.openid import OpenID
+from httpx_oauth.oauth2 import OAuth2Token, GetAccessTokenError, T
+
+
+# claims=%7B%22id_token%22%3A%7B%22birthdate%22%3A%7B%22essential%22%3Atrue%7D%2C%22name%22%3A%7B%22essential%22%3Atrue%7D%2C%22given_name%22%3A%7B%22essential%22%3Atrue%7D%2C%22family_name%22%3A%7B%22essential%22%3Atrue%7D%2C%22email%22%3A%7B%22essential%22%3Atrue%7D%2C%22address%22%3A%7B%22essential%22%3Afalse%7D%2C%22mojeid_valid%22%3A%7B%22essential%22%3Atrue%7D%7D%7D
+class MojeIDOAuth(OpenID):
+    def __init__(self, client_id: str, client_secret: str):
+        super().__init__(
+            client_id,
+            client_secret,
+            "https://mojeid.regtest.nic.cz/.well-known/openid-configuration/",
+            "MojeID",
+            base_scopes=["openid", "email", "profile"],
+        )
+
+    async def get_authorization_url(
+            self,
+            redirect_uri: str,
+            state: Optional[str] = None,
+            scope: Optional[list[str]] = None,
+            code_challenge: Optional[str] = None,
+            code_challenge_method: Optional[Literal["plain", "S256"]] = None,
+            extras_params: Optional[T] = None,
+    ) -> str:
+        required_fields = {
+            'id_token': {
+                'name': {'essential': True},
+                'given_name': {'essential': True},
+                'family_name': {'essential': True},
+                'email': {'essential': True},
+                'mojeid_valid': {'essential': True},
+            }}
+
+        if extras_params is None:
+            extras_params = {}
+        extras_params["claims"] = json.dumps(required_fields)
+
+        return await super().get_authorization_url(
+            redirect_uri,
+            state,
+            scope,
+            code_challenge,
+            code_challenge_method,
+            extras_params,
+        )

7project/backend/app/services/db.py

@@ -4,11 +4,13 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from fastapi_users.db import SQLAlchemyUserDatabase
 
 from ..core.db import async_session_maker
-from ..models.user import User
+from ..models.user import User, OAuthAccount
+
 
 async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
     async with async_session_maker() as session:
         yield session
 
+
 async def get_user_db(session: AsyncSession = Depends(get_async_session)):
-    yield SQLAlchemyUserDatabase(session, User)
+    yield SQLAlchemyUserDatabase(session, User, OAuthAccount)

7project/backend/app/services/user_service.py

@@ -12,6 +12,7 @@ from fastapi_users.authentication.strategy.jwt import JWTStrategy
 from fastapi_users.db import SQLAlchemyUserDatabase
 
 from app.models.user import User
+from app.oauth.moje_id import MojeIDOAuth
 from app.services.db import get_user_db
 from app.core.queue import enqueue_email
 
@@ -19,6 +20,11 @@ SECRET = os.getenv("SECRET", "CHANGE_ME_SECRET")
 FRONTEND_URL = os.getenv("FRONTEND_URL", "http://localhost:5173")
 BACKEND_URL = os.getenv("BACKEND_URL", "http://localhost:8000")
 
+mojeid_oauth_service = MojeIDOAuth(
+    os.getenv("MOJEID_CLIENT_ID", "CHANGE_ME_CLIENT_ID"),
+    os.getenv("MOJEID_CLIENT_SECRET", "CHANGE_ME_CLIENT_SECRET"),
+)
+
 class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
     reset_password_token_secret = SECRET
     verification_token_secret = SECRET