3 files changed, 79 insertions, 0 deletions
diff --git a/backend/authentication/__init__.py b/backend/authentication/__init__.py
new file mode 100644
index 0000000..43601a7
--- /dev/null
+++ b/backend/authentication/__init__.py
@@ -0,0 +1,4 @@
+from .backend import JWTAuthenticationBackend
+from .user import User
+
+__all__ = ["JWTAuthenticationBackend", "User"]
diff --git a/backend/authentication/backend.py b/backend/authentication/backend.py
new file mode 100644
index 0000000..38668eb
--- /dev/null
+++ b/backend/authentication/backend.py
@@ -0,0 +1,53 @@
+import jwt
+import typing as t
+from abc import ABC
+
+from starlette import authentication
+from starlette.requests import Request
+
+from backend import constants
+# We must import user such way here to avoid circular imports
+from .user import User
+
+
+class JWTAuthenticationBackend(authentication.AuthenticationBackend, ABC):
+    """Custom Starlette authentication backend for JWT."""
+
+    @staticmethod
+    def get_token_from_header(header: str) -> t.Optional[str]:
+        """Parse JWT token from header value."""
+        try:
+            prefix, token = header.split()
+        except ValueError:
+            raise authentication.AuthenticationError(
+                "Unable to split prefix and token from Authorization header."
+            )
+
+        if prefix.upper() != "JWT":
+            raise authentication.AuthenticationError(
+                f"Invalid Authorization header prefix '{prefix}'."
+            )
+
+        return token
+
+    async def authenticate(
+        self, request: Request
+    ) -> t.Optional[t.Tuple[authentication.AuthCredentials, authentication.BaseUser]]:
+        """Handles JWT authentication process."""
+        if "Authorization" not in request.headers:
+            return
+
+        auth = request.headers["Authorization"]
+        token = self.get_token_from_header(auth)
+
+        try:
+            payload = jwt.decode(token, constants.SECRET_KEY, algorithms=["HS256"])
+        except jwt.InvalidTokenError as e:
+            raise authentication.AuthenticationError(str(e))
+
+        scopes = ["authenticated"]
+
+        if payload.get("admin", False) is True:
+            scopes.append("admin")
+
+        return authentication.AuthCredentials(scopes), User(token, payload)
diff --git a/backend/authentication/user.py b/backend/authentication/user.py
new file mode 100644
index 0000000..afa243f
--- /dev/null
+++ b/backend/authentication/user.py
@@ -0,0 +1,22 @@
+import typing as t
+from abc import ABC
+
+from starlette.authentication import BaseUser
+
+
+class User(BaseUser, ABC):
+    """Starlette BaseUser implementation for JWT authentication."""
+
+    def __init__(self, token: str, payload: t.Dict) -> None:
+        self.token = token
+        self.payload = payload
+
+    @property
+    def is_authenticated(self) -> bool:
+        """Returns True because user is always authenticated at this stage."""
+        return True
+
+    @property
+    def display_name(self) -> str:
+        """Return username and discriminator as display name."""
+        return f"{self.payload['username']}#{self.payload['discriminator']}"