Set up forms app with authentication

.coveragerc

@@ -17,6 +17,11 @@ omit =
     pydis_site/settings.py
     pydis_site/utils/resources.py
     pydis_site/apps/home/views.py
+    # XXX: Will be covered later when FormsUser is used properly.
+    pydis_site/apps/forms/authentication.py
+    pydis_site/apps/forms/discord.py
+    # XXX: Will be tested later
+    pydis_site/apps/forms/permissions.py
 
 [report]
 fail_under = 100

pydis_site/apps/forms/admin.py

@@ -0,0 +1,2 @@
+
+# Register your models here.

pydis_site/apps/forms/apps.py

@@ -0,0 +1,7 @@
+from django.apps import AppConfig
+
+
+class FormsConfig(AppConfig):
+    """Django AppConfig for the forms app."""
+
+    name = "pydis_site.apps.forms"

pydis_site/apps/forms/authentication.py

@@ -0,0 +1,167 @@
+"""Custom authentication for the forms backend."""
+
+import typing
+
+import jwt
+from django.conf import settings
+from django.http import HttpRequest
+from rest_framework.authentication import BaseAuthentication
+from rest_framework.exceptions import AuthenticationFailed
+
+from . import discord
+from . import models
+
+
+def encode_jwt(info: dict, *, signing_secret_key: str = settings.SECRET_KEY) -> str:
+    """Encode JWT information with either the configured signing key or a passed one."""
+    return jwt.encode(info, signing_secret_key, algorithm="HS256")
+
+
+class FormsUser:
+    """Stores authentication information for a forms user."""
+
+    # This allows us to safely use the same checks that we could use on a Django user.
+    is_authenticated: bool = True
+
+    def __init__(
+        self,
+        token: str,
+        payload: dict[str, typing.Any],
+        member: models.DiscordMember | None,
+    ) -> None:
+        """Set up a forms user."""
+        self.token = token
+        self.payload = payload
+        self.admin = False
+        self.member = member
+
+    @property
+    def display_name(self) -> str:
+        """Return username and discriminator as display name."""
+        return f"{self.payload['username']}#{self.payload['discriminator']}"
+
+    @property
+    def discord_mention(self) -> str:
+        """Return a mention for this user on Discord."""
+        return f"<@{self.payload['id']}>"
+
+    @property
+    def user_id(self) -> str:
+        """Return this user's ID as a string."""
+        return str(self.payload["id"])
+
+    @property
+    def decoded_token(self) -> dict[str, any]:
+        """Decode the information stored in this user's JWT token."""
+        return jwt.decode(self.token, settings.SECRET_KEY, algorithms=["HS256"])
+
+    def get_roles(self) -> tuple[str, ...]:
+        """Get a tuple of the user's discord roles by name."""
+        if not self.member:
+            return []
+
+        server_roles = discord.get_roles()
+        roles = [role.name for role in server_roles if role.id in self.member.roles]
+
+        if "admin" in roles:
+            # Protect against collision with the forms admin role
+            roles.remove("admin")
+            roles.append("discord admin")
+
+        return tuple(roles)
+
+    def is_admin(self) -> bool:
+        """Return whether this user is an administrator."""
+        self.admin = models.Admin.objects.filter(id=self.payload["id"]).exists()
+        return self.admin
+
+    def refresh_data(self) -> None:
+        """Fetches user data from discord, and updates the instance."""
+        self.member = discord.get_member(self.payload["id"])
+
+        if self.member:
+            self.payload = self.member.user.dict()
+        else:
+            self.payload = discord.fetch_user_details(self.decoded_token.get("token"))
+
+        updated_info = self.decoded_token
+        updated_info["user_details"] = self.payload
+
+        self.token = encode_jwt(updated_info)
+
+
+class AuthenticationResult(typing.NamedTuple):
+    """Return scopes that the user has authenticated with."""
+
+    scopes: tuple[str, ...]
+
+
+# See https://www.django-rest-framework.org/api-guide/authentication/#custom-authentication
+class JWTAuthentication(BaseAuthentication):
+    """Custom DRF authentication backend for JWT."""
+
+    @staticmethod
+    def get_token_from_cookie(cookie: str) -> str:
+        """Parse JWT token from cookie."""
+        try:
+            prefix, token = cookie.split()
+        except ValueError:
+            msg = "Unable to split prefix and token from authorization cookie."
+            raise AuthenticationFailed(msg)
+
+        if prefix.upper() != "JWT":
+            msg = f"Invalid authorization cookie prefix '{prefix}'."
+            raise AuthenticationFailed(msg)
+
+        return token
+
+    def authenticate(
+        self,
+        request: HttpRequest,
+    ) -> tuple[FormsUser, None] | None:
+        """Handles JWT authentication process."""
+        cookie = request.COOKIES.get("token")
+        if not cookie:
+            return None
+
+        token = self.get_token_from_cookie(cookie)
+
+        try:
+            # New key.
+            payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
+        except jwt.InvalidTokenError:
+            try:
+                # Old key. Should be removed at a certain point.
+                payload = jwt.decode(token, settings.FORMS_SECRET_KEY, algorithms=["HS256"])
+            except jwt.InvalidTokenError as e:
+                raise AuthenticationFailed(str(e))
+
+        scopes = ["authenticated"]
+
+        if not payload.get("token"):
+            msg = "Token is missing from JWT."
+            raise AuthenticationFailed(msg)
+        if not payload.get("refresh"):
+            msg = "Refresh token is missing from JWT."
+            raise AuthenticationFailed(msg)
+
+        try:
+            user_details = payload.get("user_details")
+            if not user_details or not user_details.get("id"):
+                msg = "Improper user details."
+                raise AuthenticationFailed(msg)
+        except Exception:
+            msg = "Could not parse user details."
+            raise AuthenticationFailed(msg)
+
+        user = FormsUser(
+            token,
+            user_details,
+            discord.get_member(user_details["id"]),
+        )
+        if user.is_admin():
+            scopes.append("admin")
+
+        scopes.extend(user.get_roles())
+
+        return user, AuthenticationResult(scopes=tuple(scopes))

pydis_site/apps/forms/discord.py

@@ -0,0 +1,143 @@
+"""API functions for Discord access."""
+
+import httpx
+from django.conf import settings
+
+from . import models
+from . import util
+
+
+__all__ = ("get_member", "get_roles")
+
+
+def fetch_and_update_roles() -> tuple[models.DiscordRole, ...]:
+    """Get information about roles from Discord."""
+    with httpx.Client() as client:
+        r = client.get(
+            f"{settings.DISCORD_API_BASE_URL}/guilds/{settings.DISCORD_GUILD_ID}/roles",
+            headers={"Authorization": f"Bot {settings.DISCORD_BOT_TOKEN}"},
+        )
+
+        r.raise_for_status()
+        return tuple(models.DiscordRole(**role) for role in r.json())
+
+
+def fetch_member_details(member_id: int) -> models.DiscordMember | None:
+    """Get a member by ID from the configured guild using the discord API."""
+    with httpx.Client() as client:
+        r = client.get(
+            f"{settings.DISCORD_API_BASE_URL}/guilds/{settings.DISCORD_GUILD_ID}/members/{member_id}",
+            headers={"Authorization": f"Bot {settings.DISCORD_BOT_TOKEN}"},
+        )
+
+        if r.status_code == 404:
+            return None
+
+        r.raise_for_status()
+        return models.DiscordMember(**r.json())
+
+
+def fetch_user_details(bearer_token: str) -> dict:
+    """Fetch information about the Discord user associated with the given ``bearer_token``."""
+    with httpx.Client() as client:
+        r = client.get(
+            f"{settings.DISCORD_API_BASE_URL}/users/@me",
+            headers={
+                "Authorization": f"Bearer {bearer_token}",
+            },
+        )
+
+        r.raise_for_status()
+
+        return r.json()
+
+
+def fetch_bearer_token(code: str, redirect: str, *, refresh: bool) -> dict:
+    """
+    Fetch an OAuth2 bearer token.
+
+    ## Arguments
+
+    - ``code``: The code or refresh token for the operation. Usually provided by Discord.
+    - ``redirect``: Where to redirect the client after successful login.
+
+    ## Keyword arguments
+
+    - ``refresh``: Whether to fetch a refresh token.
+    """
+    with httpx.Client() as client:
+        data = {
+            "client_id": settings.DISCORD_OAUTH2_CLIENT_ID,
+            "client_secret": settings.DISCORD_OAUTH2_CLIENT_SECRET,
+            "redirect_uri": f"{redirect}/callback",
+        }
+
+        if refresh:
+            data["grant_type"] = "refresh_token"
+            data["refresh_token"] = code
+        else:
+            data["grant_type"] = "authorization_code"
+            data["code"] = code
+
+        r = client.post(
+            f"{settings.DISCORD_API_BASE_URL}/oauth2/token",
+            headers={
+                "Content-Type": "application/x-www-form-urlencoded",
+            },
+            data=data,
+        )
+
+        r.raise_for_status()
+
+        return r.json()
+
+
+def get_roles(*, force_refresh: bool = False, stale_after: int = 60 * 60 * 24) -> tuple[models.DiscordRole, ...]:
+    """
+    Get a tuple of all roles from the cache, or discord API if not available.
+
+    ## Keyword arguments
+
+    - `force_refresh` (`bool`): Skip the cache and always update the roles from
+      Discord.
+    - `stale_after` (`int`): Seconds after which to consider the stored roles
+      as stale and to refresh them.
+    """
+    if not force_refresh:
+        roles = models.DiscordRole.objects.all()
+        oldest = min(role.last_update for role in roles)
+        if not util.is_stale(oldest, 60 * 60 * 24):  # 1 day
+            return tuple(roles)
+
+    return fetch_and_update_roles()
+
+
+def get_member(
+    user_id: int,
+    *,
+    force_refresh: bool = False,
+) -> models.DiscordMember | None:
+    """
+    Get a member from the cache, or from the discord API.
+
+    ## Keyword arguments
+
+    - `force_refresh` (`bool`): Skip the cache and always update the roles from
+      Discord.
+    - `stale_after` (`int`): Seconds after which to consider the stored roles
+      as stale and to refresh them.
+
+    ## Return value
+
+    Returns `None` if the member object does not exist.
+    """
+    if not force_refresh:
+        member = models.DiscordMember.objects.get(id=user_id)
+        if not util.is_stale(member.last_update, 60 * 60):
+            return member
+
+    member = fetch_member_details(user_id)
+    if member:
+        member.save()
+
+    return member