aboutsummaryrefslogtreecommitdiffstats
path: root/authentication/backend.py
diff options
context:
space:
mode:
authorGravatar Tom van der Lee <tom@vanderlee.io>2026-09-04 17:09:24 +0200
committerGravatar Tom van der Lee <tom@vanderlee.io>2026-09-04 17:09:24 +0200
commit6b6f20a032eb31d747366c8bc270ca81731eaf6a (patch)
treeada48080495267bebde4ee1251bbbe77d629745f /authentication/backend.py
parent2d22158dd95f76399f5716d556be60c77da04a46 (diff)
downloadserver-6b6f20a032eb31d747366c8bc270ca81731eaf6a.tar.gz
server-6b6f20a032eb31d747366c8bc270ca81731eaf6a.tar.bz2
server-6b6f20a032eb31d747366c8bc270ca81731eaf6a.zip
Fixed pre-commit issuesv3
Diffstat (limited to 'authentication/backend.py')
-rw-r--r--authentication/backend.py49
1 files changed, 21 insertions, 28 deletions
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 @@
1import re 1import re
2from typing import Optional
3 2
4from sqlalchemy import update 3from sqlalchemy import update
5from sqlmodel import select 4from sqlmodel import select
6from starlette.authentication import AuthenticationBackend, AuthCredentials, BaseUser, UnauthenticatedUser 5from starlette.authentication import (
6 AuthCredentials,
7 AuthenticationBackend,
8 BaseUser,
9 UnauthenticatedUser,
10)
7from starlette.requests import HTTPConnection 11from starlette.requests import HTTPConnection
8 12
9from db.models import Session, User, AuthToken 13from db.models import AuthToken, Session, User
10from db.session import get_session_context 14from db.session import get_session_context
11 15
12 16
13class SessionAuthBackend(AuthenticationBackend): 17class SessionAuthBackend(AuthenticationBackend):
14 async def authenticate(self, request: HTTPConnection) -> Optional[tuple[AuthCredentials, BaseUser]]: 18 async def authenticate(
19 self, request: HTTPConnection
20 ) -> tuple[AuthCredentials, BaseUser] | None:
15 if "id" not in request.session: 21 if "id" not in request.session:
16 return None 22 return None
17 23
18 query = ( 24 query = select(Session).join(User).where(Session.id == request.session["id"])
19 select(Session)
20 .join(User)
21 .where(
22 Session.id == request.session['id']
23 )
24 )
25
26 25
27 with get_session_context() as db_session: 26 with get_session_context() as db_session:
28 session: Session = db_session.exec(query).first() 27 session: Session = db_session.exec(query).first()
29 28
30 if session is not None: 29 if session is not None:
31 db_session.execute( 30 db_session.execute(update(Session).where(Session.id == session.id))
32 update(Session).where(Session.id == session.id)
33 )
34 db_session.commit() 31 db_session.commit()
35 32
36
37 return ( 33 return (
38 AuthCredentials( 34 AuthCredentials(
39 ["authenticated", "session"] 35 ["authenticated", "session"] if session is not None else []
40 if session is not None
41 else []
42 ), 36 ),
43 session.user if session is not None else UnauthenticatedUser() 37 session.user if session is not None else UnauthenticatedUser(),
44 ) 38 )
45 39
46 40
47class BearerTokenAuthBackend(AuthenticationBackend): 41class BearerTokenAuthBackend(AuthenticationBackend):
48 regex = re.compile(r"^[Bb]earer\s(?P<token>\S+)$") 42 regex = re.compile(r"^[Bb]earer\s(?P<token>\S+)$")
49 async def authenticate(self, request: HTTPConnection) -> Optional[tuple[AuthCredentials, BaseUser]]: 43
44 async def authenticate(
45 self, request: HTTPConnection
46 ) -> tuple[AuthCredentials, BaseUser] | None:
50 token = None 47 token = None
51 if "Authorization" in request.headers: 48 if "Authorization" in request.headers:
52 match = self.regex.match(request.headers["Authorization"]) 49 match = self.regex.match(request.headers["Authorization"])
@@ -54,17 +51,13 @@ class BearerTokenAuthBackend(AuthenticationBackend):
54 query = ( 51 query = (
55 select(AuthToken) 52 select(AuthToken)
56 .join(User) 53 .join(User)
57 .where(AuthToken.access_token == match.group('token')) 54 .where(AuthToken.access_token == match.group("token"))
58 ) 55 )
59 56
60 with get_session_context() as db_session: 57 with get_session_context() as db_session:
61 token: AuthToken = db_session.exec(query).first() 58 token: AuthToken = db_session.exec(query).first()
62 59
63 return ( 60 return (
64 AuthCredentials( 61 AuthCredentials(["authenticated", "token"] if token is not None else []),
65 ["authenticated", "token"] 62 token.user if token is not None else UnauthenticatedUser(),
66 if token is not None
67 else []
68 ),
69 token.user if token is not None else UnauthenticatedUser()
70 ) 63 )