Merge df0f2584ae into 4ea6876b74

7project/backend/alembic/versions/2025_10_11_2107-5ab2e654c96e_change_token_lenght.py

@@ -1,38 +0,0 @@
-"""change token length
-
-Revision ID: 5ab2e654c96e
-Revises: 7af8f296d089
-Create Date: 2025-10-11 21:07:41.930470
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-from sqlalchemy.dialects import mysql
-
-# revision identifiers, used by Alembic.
-revision: str = '5ab2e654c96e'
-down_revision: Union[str, Sequence[str], None] = '7af8f296d089'
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    """Upgrade schema."""
-    # ### commands auto generated by Alembic - please adjust! ###
-    op.alter_column('oauth_account', 'access_token',
-               existing_type=mysql.VARCHAR(length=1024),
-               type_=sa.String(length=4096),
-               existing_nullable=False)
-    # ### end Alembic commands ###
-
-
-def downgrade() -> None:
-    """Downgrade schema."""
-    # ### commands auto generated by Alembic - please adjust! ###
-    op.alter_column('oauth_account', 'access_token',
-               existing_type=sa.String(length=4096),
-               type_=mysql.VARCHAR(length=1024),
-               existing_nullable=False)
-    # ### end Alembic commands ###

7project/backend/app/app.py

@@ -1,10 +1,11 @@
 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, get_oauth_provider
+from app.services.user_service import auth_backend, current_active_verified_user, fastapi_users
 
 fastApi = FastAPI()
 
@@ -46,26 +47,15 @@ fastApi.include_router(
 
 fastApi.include_router(
     fastapi_users.get_oauth_router(
-        get_oauth_provider("MojeID"),
+        app.services.user_service.mojeid_oauth_service,
         auth_backend,
         "SECRET",
-        associate_by_email=True,
+        associate_by_email=True
     ),
     prefix="/auth/mojeid",
     tags=["auth"],
 )
 
-fastApi.include_router(
-    fastapi_users.get_oauth_router(
-        get_oauth_provider("BankID"),
-        auth_backend,
-        "SECRET",
-        associate_by_email=True,
-    ),
-    prefix="/auth/bankid",
-    tags=["auth"],
-)
-
 
 # Liveness/root endpoint
 @fastApi.get("/", include_in_schema=False)

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

@@ -1,12 +1,13 @@
+from typing import List
+
 from sqlalchemy import Column, String
-from sqlalchemy.orm import relationship, mapped_column, Mapped
+from sqlalchemy.orm import relationship
 from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyBaseOAuthAccountTableUUID
 from app.core.base import Base
 
 
 class OAuthAccount(SQLAlchemyBaseOAuthAccountTableUUID, Base):
-    # BankID token is longer than default
+    pass
-    access_token: Mapped[str] = mapped_column(String(length=4096), nullable=False)
 
 
 class User(SQLAlchemyBaseUserTableUUID, Base):

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

@@ -1,50 +0,0 @@
-import secrets
-from typing import Optional, Literal
-
-from httpx_oauth.oauth2 import T
-
-from app.oauth.custom_openid import CustomOpenID
-
-
-class BankID(CustomOpenID):
-    def __init__(self, client_id: str, client_secret: str):
-        super().__init__(
-            client_id,
-            client_secret,
-            "https://oidc.sandbox.bankid.cz/.well-known/openid-configuration",
-            "BankID",
-            base_scopes=["openid", "profile.email", "profile.name"],
-        )
-
-    async def get_user_info(self, token: str) -> dict:
-        info = await self.get_profile(token)
-
-        return {
-            "first_name": info.get("given_name"),
-            "last_name": info.get("family_name"),
-        }
-
-    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:
-        if extras_params is None:
-            extras_params = {}
-
-        # BankID requires random nonce parameter for security
-        # https://developer.bankid.cz/docs/security_sep
-        extras_params["nonce"] = secrets.token_urlsafe()
-
-        return await super().get_authorization_url(
-            redirect_uri,
-            state,
-            scope,
-            code_challenge,
-            code_challenge_method,
-            extras_params,
-        )

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

@@ -1,6 +0,0 @@
-from httpx_oauth.clients.openid import OpenID
-
-
-class CustomOpenID(OpenID):
-    async def get_user_info(self, token: str) -> dict:
-        raise NotImplementedError()

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

@@ -1,12 +1,12 @@
 import json
-from typing import Optional, Literal, Any
+from typing import Optional, Literal
 
-from httpx_oauth.oauth2 import T
+from httpx_oauth.clients.openid import OpenID
-
+from httpx_oauth.oauth2 import OAuth2Token, GetAccessTokenError, T
-from app.oauth.custom_openid import CustomOpenID
 
 
-class MojeIDOAuth(CustomOpenID):
+# 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,
@@ -16,14 +16,6 @@ class MojeIDOAuth(CustomOpenID):
             base_scopes=["openid", "email", "profile"],
         )
 
-    async def get_user_info(self, token: str) -> Optional[Any]:
-        info = await self.get_profile(token)
-
-        return {
-            "first_name": info.get("given_name"),
-            "last_name": info.get("family_name"),
-        }
-
     async def get_authorization_url(
             self,
             redirect_uri: str,

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

@@ -3,66 +3,32 @@ import uuid
 from typing import Optional
 
 from fastapi import Depends, Request
-from fastapi_users import BaseUserManager, FastAPIUsers, UUIDIDMixin, models
+from fastapi_users import BaseUserManager, FastAPIUsers, UUIDIDMixin
 from fastapi_users.authentication import (
     AuthenticationBackend,
     BearerTransport,
 )
 from fastapi_users.authentication.strategy.jwt import JWTStrategy
 from fastapi_users.db import SQLAlchemyUserDatabase
-from httpx_oauth.oauth2 import BaseOAuth2
 
 from app.models.user import User
-from app.oauth.bank_id import BankID
-from app.oauth.custom_openid import CustomOpenID
 from app.oauth.moje_id import MojeIDOAuth
 from app.services.db import get_user_db
 from app.core.queue import enqueue_email
 
 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")
 
-providers = {
+mojeid_oauth_service = MojeIDOAuth(
-    "MojeID": MojeIDOAuth(
     os.getenv("MOJEID_CLIENT_ID", "CHANGE_ME_CLIENT_ID"),
     os.getenv("MOJEID_CLIENT_SECRET", "CHANGE_ME_CLIENT_SECRET"),
-    ),
-    "BankID": BankID(
-        os.getenv("BANKID_CLIENT_ID", "CHANGE_ME_CLIENT_ID"),
-        os.getenv("BANKID_CLIENT_SECRET", "CHANGE_ME_CLIENT_SECRET"),
 )
-}
-
-
-def get_oauth_provider(name: str) -> Optional[BaseOAuth2]:
-    if name not in providers:
-        return None
-    return providers[name]
-
 
 class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
     reset_password_token_secret = SECRET
     verification_token_secret = SECRET
 
-    async def oauth_callback(self: "BaseUserManager[models.UOAP, models.ID]", oauth_name: str, access_token: str,
-                             account_id: str, account_email: str, expires_at: Optional[int] = None,
-                             refresh_token: Optional[str] = None, request: Optional[Request] = None, *,
-                             associate_by_email: bool = False, is_verified_by_default: bool = False) -> models.UOAP:
-
-        user = await super().oauth_callback(oauth_name, access_token, account_id, account_email, expires_at,
-                                            refresh_token, request, associate_by_email=associate_by_email,
-                                            is_verified_by_default=is_verified_by_default)
-
-        # set additional user info from the OAuth provider
-        provider = get_oauth_provider(oauth_name)
-        if provider is not None and isinstance(provider, CustomOpenID):
-            update_dict = await provider.get_user_info(access_token)
-            await self.user_db.update(user, update_dict)
-
-        return user
-
     async def on_after_register(self, user: User, request: Optional[Request] = None):
         await self.request_verify(user, request)
 
@@ -92,18 +58,14 @@ class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
             print("[Email Fallback] Subject:", subject)
             print("[Email Fallback] Body:\n", body)
 
-
 async def get_user_manager(user_db: SQLAlchemyUserDatabase = Depends(get_user_db)):
     yield UserManager(user_db)
 
-
 bearer_transport = BearerTransport(tokenUrl="auth/jwt/login")
 
-
 def get_jwt_strategy() -> JWTStrategy:
     return JWTStrategy(secret=SECRET, lifetime_seconds=3600)
 
-
 auth_backend = AuthenticationBackend(
     name="jwt",
     transport=bearer_transport,
@@ -114,3 +76,4 @@ fastapi_users = FastAPIUsers[User, uuid.UUID](get_user_manager, [auth_backend])
 
 current_active_user = fastapi_users.current_user(active=True)
 current_active_verified_user = fastapi_users.current_user(active=True, verified=True)
+