Fixed pre-commit issuesv3

5 files changed, 197 insertions, 128 deletions

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<token>\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