From 6b6f20a032eb31d747366c8bc270ca81731eaf6a Mon Sep 17 00:00:00 2001 From: Tom van der Lee Date: Fri, 4 Sep 2026 17:09:24 +0200 Subject: Fixed pre-commit issues --- authentication/authlib.py | 77 ++++++++++++------- authentication/backend.py | 49 ++++++------ authentication/endpoints.py | 177 +++++++++++++++++++++++++++----------------- authentication/mixins.py | 12 ++- authentication/utils.py | 10 +-- 5 files changed, 197 insertions(+), 128 deletions(-) (limited to 'authentication') diff --git a/authentication/authlib.py b/authentication/authlib.py index 8472062..a37d68f 100644 --- a/authentication/authlib.py +++ b/authentication/authlib.py @@ -1,31 +1,37 @@ import json from collections import defaultdict +from typing import ClassVar from unittest.mock import patch from authlib.common.security import generate_token as generate_random_token -from authlib.oauth2 import AuthorizationServer as BaseAuthorizationServer, OAuth2Request -from authlib.oauth2.rfc6749 import AuthorizationCodeGrant as BaseAuthorizationCodeGrant, \ - RefreshTokenGrant as BaseRefreshTokenGrant, OAuth2Payload +from authlib.oauth2 import AuthorizationServer as BaseAuthorizationServer +from authlib.oauth2 import OAuth2Request +from authlib.oauth2.rfc6749 import AuthorizationCodeGrant as BaseAuthorizationCodeGrant +from authlib.oauth2.rfc6749 import OAuth2Payload +from authlib.oauth2.rfc6749 import RefreshTokenGrant as BaseRefreshTokenGrant from authlib.oauth2.rfc6750 import BearerTokenGenerator from authlib.oauth2.rfc7636 import CodeChallenge +from fastapi import Request, Response from sqlalchemy.exc import NoResultFound - from sqlmodel import select -from fastapi import Request, Response -from db.models import ClientApplication, AuthToken, AuthCode, User +from db.models import AuthCode, AuthToken, ClientApplication, User from db.session import get_session_context class AuthorizationCodeGrant(BaseAuthorizationCodeGrant): - TOKEN_ENDPOINT_AUTH_METHODS = ['client_secret_basic', 'client_secret_post', 'none'] + TOKEN_ENDPOINT_AUTH_METHODS: ClassVar[list[str]] = [ + "client_secret_basic", + "client_secret_post", + "none", + ] def save_authorization_code(self, code: str, request: OAuth2Request): with get_session_context() as db_session: db_session.add(request.user) - code_challenge = request.payload.data.get('code_challenge') - code_challenge_method = request.payload.data.get('code_challenge_method') + code_challenge = request.payload.data.get("code_challenge") + code_challenge_method = request.payload.data.get("code_challenge_method") auth_code = AuthCode( code=code, @@ -41,10 +47,16 @@ class AuthorizationCodeGrant(BaseAuthorizationCodeGrant): return auth_code - def query_authorization_code(self, code: str, client: ClientApplication) -> AuthCode | None: + def query_authorization_code( + self, code: str, client: ClientApplication + ) -> AuthCode | None: try: with get_session_context() as db_session: - auth_code = db_session.exec(select(AuthCode).where(AuthCode.code == code, AuthCode.client_id == client.client_id)).one() + auth_code = db_session.exec( + select(AuthCode).where( + AuthCode.code == code, AuthCode.client_id == client.client_id + ) + ).one() except NoResultFound: return None @@ -65,7 +77,9 @@ class RefreshTokenGrant(BaseRefreshTokenGrant): def authenticate_refresh_token(self, refresh_token) -> AuthToken | None: try: with get_session_context() as db_session: - token = db_session.exec(select(AuthToken).where(AuthToken.refresh_token == refresh_token)).one() + token = db_session.exec( + select(AuthToken).where(AuthToken.refresh_token == refresh_token) + ).one() except NoResultFound: return None @@ -82,6 +96,7 @@ class RefreshTokenGrant(BaseRefreshTokenGrant): credential.revoked = True db_session.add(credential) + async def prepare_oauth_request(request: Request) -> Request: async with request.form() as form: request.state.form_data = dict(form) @@ -96,7 +111,7 @@ class FastApiOAuth2Payload(OAuth2Payload): def data(self): return { **self._request.query_params, - **getattr(self._request.state, 'form_data', {}), + **getattr(self._request.state, "form_data", {}), } @property @@ -106,9 +121,10 @@ class FastApiOAuth2Payload(OAuth2Payload): values[k].extend([self.data[k]]) return values + class FastApiOAuth2Request(OAuth2Request): def __init__(self, request: Request): - with patch('authlib.oauth2.rfc6749.errors.InsecureTransportError'): + with patch("authlib.oauth2.rfc6749.errors.InsecureTransportError"): super().__init__( method=request.method, uri=str(request.url), @@ -128,7 +144,7 @@ class FastApiOAuth2Request(OAuth2Request): @property def form(self): - return getattr(self._request.state, 'form_data', {}) + return getattr(self._request.state, "form_data", {}) class AuthorizationServer(BaseAuthorizationServer): @@ -136,22 +152,33 @@ class AuthorizationServer(BaseAuthorizationServer): super().__init__(*args, **kwargs) self.register_grant(AuthorizationCodeGrant, extensions=[CodeChallenge()]) self.register_grant(RefreshTokenGrant) - self.register_token_generator("default", BearerTokenGenerator( - access_token_generator=lambda *args, **kwargs: generate_random_token(42), - refresh_token_generator=lambda *args, **kwargs: generate_random_token(48), - )) + self.register_token_generator( + "default", + BearerTokenGenerator( + access_token_generator=lambda *args, **kwargs: generate_random_token( + 42 + ), + refresh_token_generator=lambda *args, **kwargs: generate_random_token( + 48 + ), + ), + ) def query_client(self, client_id: str) -> ClientApplication: with get_session_context() as db_session: - return db_session.exec(select(ClientApplication).where(ClientApplication.client_id == client_id)).one() + return db_session.exec( + select(ClientApplication).where( + ClientApplication.client_id == client_id + ) + ).one() def save_token(self, token: dict, request: OAuth2Request): with get_session_context() as db_session: - db_session.add(AuthToken( - user_id=request.user.id, - client_id=request.client.client_id, - **token - )) + db_session.add( + AuthToken( + user_id=request.user.id, client_id=request.client.client_id, **token + ) + ) def create_oauth2_request(self, request: Request) -> OAuth2Request: return FastApiOAuth2Request(request) diff --git a/authentication/backend.py b/authentication/backend.py index 9434906..58842de 100644 --- a/authentication/backend.py +++ b/authentication/backend.py @@ -1,52 +1,49 @@ import re -from typing import Optional from sqlalchemy import update from sqlmodel import select -from starlette.authentication import AuthenticationBackend, AuthCredentials, BaseUser, UnauthenticatedUser +from starlette.authentication import ( + AuthCredentials, + AuthenticationBackend, + BaseUser, + UnauthenticatedUser, +) from starlette.requests import HTTPConnection -from db.models import Session, User, AuthToken +from db.models import AuthToken, Session, User from db.session import get_session_context class SessionAuthBackend(AuthenticationBackend): - async def authenticate(self, request: HTTPConnection) -> Optional[tuple[AuthCredentials, BaseUser]]: + async def authenticate( + self, request: HTTPConnection + ) -> tuple[AuthCredentials, BaseUser] | None: if "id" not in request.session: return None - query = ( - select(Session) - .join(User) - .where( - Session.id == request.session['id'] - ) - ) - + query = select(Session).join(User).where(Session.id == request.session["id"]) with get_session_context() as db_session: session: Session = db_session.exec(query).first() if session is not None: - db_session.execute( - update(Session).where(Session.id == session.id) - ) + db_session.execute(update(Session).where(Session.id == session.id)) db_session.commit() - return ( AuthCredentials( - ["authenticated", "session"] - if session is not None - else [] + ["authenticated", "session"] if session is not None else [] ), - session.user if session is not None else UnauthenticatedUser() + session.user if session is not None else UnauthenticatedUser(), ) class BearerTokenAuthBackend(AuthenticationBackend): regex = re.compile(r"^[Bb]earer\s(?P\S+)$") - async def authenticate(self, request: HTTPConnection) -> Optional[tuple[AuthCredentials, BaseUser]]: + + async def authenticate( + self, request: HTTPConnection + ) -> tuple[AuthCredentials, BaseUser] | None: token = None if "Authorization" in request.headers: match = self.regex.match(request.headers["Authorization"]) @@ -54,17 +51,13 @@ class BearerTokenAuthBackend(AuthenticationBackend): query = ( select(AuthToken) .join(User) - .where(AuthToken.access_token == match.group('token')) + .where(AuthToken.access_token == match.group("token")) ) with get_session_context() as db_session: token: AuthToken = db_session.exec(query).first() return ( - AuthCredentials( - ["authenticated", "token"] - if token is not None - else [] - ), - token.user if token is not None else UnauthenticatedUser() + AuthCredentials(["authenticated", "token"] if token is not None else []), + token.user if token is not None else UnauthenticatedUser(), ) diff --git a/authentication/endpoints.py b/authentication/endpoints.py index 3af49fc..17d49d7 100644 --- a/authentication/endpoints.py +++ b/authentication/endpoints.py @@ -4,8 +4,8 @@ from functools import partial from typing import Annotated, Self from uuid import uuid7 -from fastapi import FastAPI, Request, Form, Query -from pydantic import BaseModel, ValidationError, model_validator, EmailStr +from fastapi import FastAPI, Form, Query, Request +from pydantic import BaseModel, EmailStr, ValidationError, model_validator from sqlalchemy import func from sqlmodel import select from starlette.authentication import requires @@ -15,10 +15,14 @@ from starlette.responses import HTMLResponse, RedirectResponse from starlette.templating import Jinja2Templates import conf -from authentication.authlib import AuthorizationServer, AuthorizationCodeGrant, prepare_oauth_request +from authentication.authlib import ( + AuthorizationCodeGrant, + AuthorizationServer, + prepare_oauth_request, +) from authentication.backend import SessionAuthBackend from authentication.utils import bcrypt_sha256_verify, get_url_path_with_query -from db.models import User, Session, ClientApplication +from db.models import ClientApplication, Session, User from db.session import get_session_context authentication = FastAPI() @@ -31,10 +35,7 @@ oauth.add_middleware(SessionMiddleware, secret_key=conf.SECRET_KEY) auth_server = AuthorizationServer() code_grant = partial(AuthorizationCodeGrant, server=auth_server) -templates = Jinja2Templates(directory=[ - 'templates', - 'authentication/templates' -]) +templates = Jinja2Templates(directory=["templates", "authentication/templates"]) class LoginForm(BaseModel): @@ -42,57 +43,75 @@ class LoginForm(BaseModel): password: str -async def base_login(request: Request, form: LoginForm | None = None, errors: dict[str, str] | None = None): - return templates.TemplateResponse(request, 'login.html', context={ - 'form': form, - 'errors': errors, - }) +async def base_login( + request: Request, + form: LoginForm | None = None, + errors: dict[str, str] | None = None, +): + return templates.TemplateResponse( + request, + "login.html", + context={ + "form": form, + "errors": errors, + }, + ) -@authentication.get('/login/', response_class=HTMLResponse) -async def get_login(request: Request, next: Annotated[str, Query()] = None): +@authentication.get("/login/", response_class=HTMLResponse) +async def get_login(request: Request, next: Annotated[str | None, Query()] = None): return await base_login(request) -@authentication.post('/login/', response_class=HTMLResponse) -async def post_login(request: Request, username: Annotated[str, Form()] = '', password: Annotated[str, Form()] = '', - next: Annotated[str, Query()] = None): +@authentication.post("/login/", response_class=HTMLResponse) +async def post_login( + request: Request, + username: Annotated[str, Form()] = "", + password: Annotated[str, Form()] = "", + next: Annotated[str | None, Query()] = None, +): form = None errors = {} try: - form = LoginForm.model_validate({ - 'username': username, - 'password': password - }) + form = LoginForm.model_validate({"username": username, "password": password}) with get_session_context() as db_session: user = db_session.query(User).filter(User.username == form.username).first() - if not bcrypt_sha256_verify(password, user.password if user is not None else '') or user is None: - errors.update({'password': 'Invalid username or password'}) + if ( + not bcrypt_sha256_verify( + password, user.password if user is not None else "" + ) + or user is None + ): + errors.update({"password": "Invalid username or password"}) else: with get_session_context() as db_session: session = Session(id=uuid7(), user=user) db_session.add(session) - request.session.update({'id': str(session.id)}) + request.session.update({"id": str(session.id)}) if next is not None: return RedirectResponse(next, status_code=303) except ValidationError as exc: - errors.update({e['loc'][0]: e['msg'] for e in exc.errors()}) + errors.update({e["loc"][0]: e["msg"] for e in exc.errors()}) return await base_login(request, form, errors) -@authentication.get('/logout/') +@authentication.get("/logout/") def logout(request: Request): with get_session_context() as db_session: - db_session.query(Session).filter(Session.id == request.session.get('id')).delete() + db_session.query(Session).filter( + Session.id == request.session.get("id") + ).delete() request.session.clear() - return RedirectResponse(request.headers.get('Referer', '/management/dashboard/'), status_code=303) + return RedirectResponse( + request.headers.get("Referer", "/management/dashboard/"), status_code=303 + ) class RegisterForm(BaseModel): @@ -100,10 +119,10 @@ class RegisterForm(BaseModel): password: str verify_password: str - @model_validator(mode='after') + @model_validator(mode="after") def check_passwords_match(self) -> Self: if self.password != self.verify_password: - raise ValueError('Passwords do not match') + raise ValueError("Passwords do not match") return self @@ -113,31 +132,42 @@ def is_first_run() -> bool: return user_count.first() == 0 -async def base_register(request: Request, form: RegisterForm | None = None, errors: dict[str, str] | None = None): - return templates.TemplateResponse(request, 'register.html', context={ - 'form': form, - 'errors': errors, - 'is_first_run': is_first_run(), - }) +async def base_register( + request: Request, + form: RegisterForm | None = None, + errors: dict[str, str] | None = None, +): + return templates.TemplateResponse( + request, + "register.html", + context={ + "form": form, + "errors": errors, + "is_first_run": is_first_run(), + }, + ) -@authentication.get('/register/', response_class=HTMLResponse) -async def get_register(request: Request, next: Annotated[str, Query()] = None): +@authentication.get("/register/", response_class=HTMLResponse) +async def get_register(request: Request, next: Annotated[str | None, Query()] = None): return await base_register(request) -@authentication.post('/register/', response_class=HTMLResponse) -async def post_register(request: Request, email: Annotated[str, Form()] = '', password: Annotated[str, Form()] = '', - verify_password: Annotated[str, Form()] = '', next: Annotated[str, Query()] = None): +@authentication.post("/register/", response_class=HTMLResponse) +async def post_register( + request: Request, + email: Annotated[str, Form()] = "", + password: Annotated[str, Form()] = "", + verify_password: Annotated[str, Form()] = "", + next: Annotated[str | None, Query()] = None, +): form = None errors = {} try: - form = RegisterForm.model_validate({ - 'email': email, - 'password': password, - 'verify_password': verify_password - }) + form = RegisterForm.model_validate( + {"email": email, "password": password, "verify_password": verify_password} + ) with get_session_context() as db_session: user = User(username=email, email=email) @@ -147,13 +177,18 @@ async def post_register(request: Request, email: Annotated[str, Form()] = '', pa session = Session(id=uuid7(), user=user) db_session.add(session) - request.session.update({'id': str(session.id)}) + request.session.update({"id": str(session.id)}) if next is not None: return RedirectResponse(next, status_code=303) except ValidationError as exc: - errors.update({(e['loc'][0] if e['loc'] else 'verify_password'): e['msg'] for e in exc.errors()}) + errors.update( + { + (e["loc"][0] if e["loc"] else "verify_password"): e["msg"] + for e in exc.errors() + } + ) return await base_register(request, form, errors) @@ -162,49 +197,59 @@ async def base_authorize(request: Request, user: User | None = None): return auth_server.create_authorization_response(request, request.user, grant=None) -@oauth.get('/authorize/') -@requires(['authenticated', 'session'], redirect='get_login') +@oauth.get("/authorize/") +@requires(["authenticated", "session"], redirect="get_login") async def get_authorize(request: Request): return await base_authorize(request, request.user) -@oauth.post('/authorize/') -@requires(['authenticated', 'session'], redirect='get_login') +@oauth.post("/authorize/") +@requires(["authenticated", "session"], redirect="get_login") async def post_authorize(request: Request): return await base_authorize(request, request.user) -@oauth.post('/token/') +@oauth.post("/token/") async def post_token(request: Request): await prepare_oauth_request(request) return auth_server.create_token_response(request) -@authentication.get('/connect/') +@authentication.get("/connect/") async def connect(request: Request): - origin = request.url.scheme + '://' + request.url.netloc + origin = request.url.scheme + "://" + request.url.netloc with get_session_context() as session: app = session.exec(select(ClientApplication)).first() return RedirectResponse( - '/oauth' + get_url_path_with_query(oauth, 'get_authorize', { - 'client_id': app.client_id, - 'redirect_uri': f'{origin}/oauth' + get_url_path_with_query(oauth, 'callback'), - 'response_type': 'code', - 'state': b64encode(json.dumps({ - 'client_id': app.client_id, - }).encode()).decode(), - }), + "/oauth" + + get_url_path_with_query( + oauth, + "get_authorize", + { + "client_id": app.client_id, + "redirect_uri": f"{origin}/oauth" + + get_url_path_with_query(oauth, "callback"), + "response_type": "code", + "state": b64encode( + json.dumps( + { + "client_id": app.client_id, + } + ).encode() + ).decode(), + }, + ), status_code=303, ) -@oauth.get('/config/') +@oauth.get("/config/") async def get_config(request: Request): with get_session_context() as session: app = session.exec(select(ClientApplication)).first() return { - 'client_id': app.client_id, + "client_id": app.client_id, } diff --git a/authentication/mixins.py b/authentication/mixins.py index 3d6a91f..70816a1 100644 --- a/authentication/mixins.py +++ b/authentication/mixins.py @@ -1,7 +1,11 @@ import json import time -from authlib.integrations.sqla_oauth2 import OAuth2ClientMixin as BaseOAuth2ClientMixin, OAuth2AuthorizationCodeMixin as BaseOAuth2AuthorizationCodeMixin, OAuth2TokenMixin as BaseOAuth2TokenMixin +from authlib.integrations.sqla_oauth2 import ( + OAuth2AuthorizationCodeMixin as BaseOAuth2AuthorizationCodeMixin, +) +from authlib.integrations.sqla_oauth2 import OAuth2ClientMixin as BaseOAuth2ClientMixin +from authlib.integrations.sqla_oauth2 import OAuth2TokenMixin as BaseOAuth2TokenMixin from sqlalchemy.sql.sqltypes import Text from sqlmodel import Field @@ -11,9 +15,9 @@ class OAuth2ClientMixin(BaseOAuth2ClientMixin): client_secret: str = Field(max_length=120, nullable=True) client_id_issued_at: int = Field(nullable=False, default=0) client_secret_expires_at: int = Field(nullable=False, default=0) - client_metadata_field: str = Field(sa_type=Text, sa_column_kwargs={ - 'name': 'client_metadata' - }, nullable=True) + client_metadata_field: str = Field( + sa_type=Text, sa_column_kwargs={"name": "client_metadata"}, nullable=True + ) _client_metadata = None diff --git a/authentication/utils.py b/authentication/utils.py index 4f19bc3..b0c9989 100644 --- a/authentication/utils.py +++ b/authentication/utils.py @@ -15,11 +15,11 @@ def bcrypt_sha256_hash(password: str) -> str: def bcrypt_sha256_verify(password: str, hashed: str) -> bool: digest = hashlib.sha256(password.encode("utf-8")).digest() encoded = base64.b64encode(digest) - return bcrypt.checkpw(encoded, hashed.encode('utf-8')) + return bcrypt.checkpw(encoded, hashed.encode("utf-8")) -def get_url_path_with_query(app: FastAPI, name: str, params: dict | None = None, /, **path_params: str): +def get_url_path_with_query( + app: FastAPI, name: str, params: dict | None = None, /, **path_params: str +): base = app.url_path_for(name, **path_params) - return base + '?' + urlencode(params) if params is not None else base - - + return base + "?" + urlencode(params) if params is not None else base -- cgit v1.3