From 498c79f434856aa68e9a883247ea69a22256fce6 Mon Sep 17 00:00:00 2001
From: Tom van der Lee <tom@vanderlee.io>
Date: Wed, 22 Jul 2026 14:51:44 +0200
Subject: Added auth layer

---
 authentication/backend.py | 40 +++++++++++++++++++++++++++++++---------
 1 file changed, 31 insertions(+), 9 deletions(-)

(limited to 'authentication/backend.py')

diff --git a/authentication/backend.py b/authentication/backend.py
index 29408e3..9434906 100644
--- a/authentication/backend.py
+++ b/authentication/backend.py
@@ -1,4 +1,4 @@
-from datetime import datetime, UTC
+import re
 from typing import Optional
 
 from sqlalchemy import update
@@ -6,11 +6,11 @@ from sqlmodel import select
 from starlette.authentication import AuthenticationBackend, AuthCredentials, BaseUser, UnauthenticatedUser
 from starlette.requests import HTTPConnection
 
-from db.models import Session, User
+from db.models import Session, User, AuthToken
 from db.session import get_session_context
 
 
-class AuthBackend(AuthenticationBackend):
+class SessionAuthBackend(AuthenticationBackend):
     async def authenticate(self, request: HTTPConnection) -> Optional[tuple[AuthCredentials, BaseUser]]:
         if "id" not in request.session:
             return None
@@ -33,16 +33,38 @@ class AuthBackend(AuthenticationBackend):
                 )
                 db_session.commit()
 
+
             return (
                 AuthCredentials(
-                    [
-                        "authenticated",
-                        *[
-                            connection.type for connection in session.user.app_connections
-                        ]
-                    ]
+                    ["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) -> Optional[tuple[AuthCredentials, BaseUser]]:
+        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()
+        )
-- 
cgit v1.2.3