1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
|
import re
from sqlalchemy import update
from sqlmodel import select
from starlette.authentication import (
AuthCredentials,
AuthenticationBackend,
BaseUser,
UnauthenticatedUser,
)
from starlette.requests import HTTPConnection
from db.models import AuthToken, Session, User
from db.session import get_session_context
class SessionAuthBackend(AuthenticationBackend):
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"])
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.commit()
return (
AuthCredentials(
["authenticated", "session"] if session is not None else []
),
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
) -> tuple[AuthCredentials, BaseUser] | None:
token = None
if "Authorization" in request.headers:
match = self.regex.match(request.headers["Authorization"])
query = (
select(AuthToken)
.join(User)
.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(),
)
|
