Implement forms /admin endpointforms-admin

7 files changed, 141 insertions, 18 deletions

diff --git a/pydis_site/apps/forms/admin.py b/pydis_site/apps/forms/admin.py
index b97a94f6..846f6b40 100644
--- a/pydis_site/apps/forms/admin.py
+++ b/pydis_site/apps/forms/admin.py
@@ -1,2 +1 @@
-
 # Register your models here.
diff --git a/pydis_site/apps/forms/authentication.py b/pydis_site/apps/forms/authentication.py
index c613d160..121ca1d2 100644
--- a/pydis_site/apps/forms/authentication.py
+++ b/pydis_site/apps/forms/authentication.py
@@ -1,12 +1,16 @@
 """Custom authentication for the forms backend."""
 
+import functools
 import typing
+from http import HTTPStatus
 
 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 rest_framework.response import Response
+from rest_framework.views import APIView
 
 from . import discord
 from . import models
@@ -165,3 +169,45 @@ class JWTAuthentication(BaseAuthentication):
         scopes.extend(user.get_roles())
 
         return user, AuthenticationResult(scopes=tuple(scopes))
+
+
+APIHandlerMethod = typing.Callable[[APIView, HttpRequest, str | None], Response]
+"""Represents a DRF API class-based view endpoint method."""
+
+
+def require_scopes(scopes: frozenset[str]) -> APIHandlerMethod:
+    """Require the requesting user to have the given `scopes`."""
+    if not isinstance(scopes, set):
+        error = TypeError("please supply scopes as a set")
+        error.add_note(f"got {scopes!r} ({type(scopes)})")
+        raise error
+
+    required_scopes = frozenset(scopes)
+
+    def wrapper(f: APIHandlerMethod) -> APIHandlerMethod:
+        @functools.wraps(f)
+        def authenticated_endpoint(instance: APIView, request: HttpRequest, format: str | None = None) -> Response:
+            """Verify the requesting user is authenticated."""
+            if not request.user.is_authenticated:
+                return Response(
+                    data={"king-arthur-verdict": "not-allowed"},
+                    status=HTTPStatus.UNAUTHORIZED,
+                )
+
+            request_scopes = frozenset(request.auth.scopes)
+            actual_scopes = required_scopes.intersection(request_scopes)
+            lacking_scopes = required_scopes - actual_scopes
+            if lacking_scopes:
+                return Response(
+                    data={
+                        "king-arthur-verdict": "missing-scopes",
+                        "missing-scopes": tuple(lacking_scopes),
+                    },
+                    status=HTTPStatus.FORBIDDEN,
+                )
+
+            return f(instance, request, format)
+
+        return authenticated_endpoint
+
+    return wrapper
diff --git a/pydis_site/apps/forms/models.py b/pydis_site/apps/forms/models.py
index 6b781efe..07d1697c 100644
--- a/pydis_site/apps/forms/models.py
+++ b/pydis_site/apps/forms/models.py
@@ -1,4 +1,3 @@
-
 from django.core.validators import MinValueValidator, MaxValueValidator
 from django.contrib.postgres.fields import ArrayField
 from django.db import models
diff --git a/pydis_site/apps/forms/tests/test_api.py b/pydis_site/apps/forms/tests/test_api.py
index 88cdca1c..282ae222 100644
--- a/pydis_site/apps/forms/tests/test_api.py
+++ b/pydis_site/apps/forms/tests/test_api.py
@@ -1,4 +1,3 @@
-
 from django.conf import settings
 from django.test import TestCase, override_settings
 from django.urls import reverse
@@ -7,15 +6,6 @@ from pydis_site.apps.forms import authentication
 from pydis_site.apps.forms.tests.base import AuthenticatedTestCase
 
 
-class TestIndex(TestCase):
-    def test_index_returns_200(self) -> None:
-        """The index page should return a HTTP 200 response."""
-
-        url = reverse("forms:index")
-        resp = self.client.get(url)
-        self.assertEqual(resp.status_code, 200)
-
-
 class TestAuthentication(TestCase):
     def tearDown(self) -> None:
         # Removes all cookies from the test client.
diff --git a/pydis_site/apps/forms/tests/test_views.py b/pydis_site/apps/forms/tests/test_views.py
new file mode 100644
index 00000000..ae851682
--- /dev/null
+++ b/pydis_site/apps/forms/tests/test_views.py
@@ -0,0 +1,58 @@
+from http import HTTPStatus
+
+from django.test import TestCase
+from django.urls import reverse
+
+from .base import AuthenticatedTestCase
+from pydis_site.apps.forms.models import Admin
+
+
+class IndexTestCase(TestCase):
+    def test_index_returns_200(self) -> None:
+        """The index page should return a HTTP 200 response."""
+
+        url = reverse("forms:index")
+        resp = self.client.get(url)
+        self.assertEqual(resp.status_code, 200)
+
+
+class UnauthorizedAdminTestCase(TestCase):
+    def test_create_admin_unauthorized(self) -> None:
+        """The POST /admin URL should return an error without auth."""
+
+        url = reverse("forms:admin")
+        resp = self.client.post(url, {"hello": "world"})
+        self.assertEqual(resp.status_code, HTTPStatus.UNAUTHORIZED)
+
+
+class ForbiddenAdminTestCase(AuthenticatedTestCase):
+    def test_create_admin_forbidden(self) -> None:
+        """The POST /admin URL should return an error without the required scopes."""
+
+        url = reverse("forms:admin")
+        resp = self.client.post(url, {"hello": "world"})
+        self.assertEqual(resp.status_code, HTTPStatus.FORBIDDEN)
+
+
+class AdminTestCase(AuthenticatedTestCase):
+    authenticate_as_admin = True
+
+    def test_create_admin_fresh(self) -> None:
+        """Creating an admin should work when an admin makes the request."""
+
+        url = reverse("forms:admin")
+        resp = self.client.post(url, {"id": 1234})
+        self.assertEqual(resp.status_code, HTTPStatus.OK)
+        admin = Admin.objects.get(id=1234)
+        self.assertIsNotNone(admin)
+        resp = self.client.post(url, {"id": 1234})
+        self.assertEqual(resp.status_code, HTTPStatus.BAD_REQUEST)
+        self.assertEqual(resp.json()["error"], "already_exists")
+
+    def test_create_admin_missing_field(self) -> None:
+        """Creating an admin should require the `id` field to be sent."""
+
+        url = reverse("forms:admin")
+        resp = self.client.post(url, {"oranges and lemons": "say the bells of st clement's"})
+        self.assertEqual(resp.status_code, HTTPStatus.BAD_REQUEST)
+        self.assertEqual(resp.json()["error"], "which-id-please")
diff --git a/pydis_site/apps/forms/urls.py b/pydis_site/apps/forms/urls.py
index b5462960..3e20d11a 100644
--- a/pydis_site/apps/forms/urls.py
+++ b/pydis_site/apps/forms/urls.py
@@ -1,7 +1,10 @@
 from django.urls import path
 
-from .views import IndexView
+from .views import AdminView, IndexView
 
 
 app_name = "forms"
-urlpatterns = (path("", IndexView.as_view(), name="index"),)
+urlpatterns = (
+    path("admin", AdminView.as_view(), name="admin"),
+    path("", IndexView.as_view(), name="index"),
+)
diff --git a/pydis_site/apps/forms/views.py b/pydis_site/apps/forms/views.py
index 4c9da8cc..8c220513 100644
--- a/pydis_site/apps/forms/views.py
+++ b/pydis_site/apps/forms/views.py
@@ -1,11 +1,14 @@
 import platform
+from http import HTTPStatus
 
 from django.conf import settings
-from django.http import HttpRequest
-from rest_framework.views import APIView
+from django.db import IntegrityError
 from rest_framework.response import Response
+from rest_framework.request import Request
+from rest_framework.views import APIView
 
-from .authentication import JWTAuthentication
+from .authentication import JWTAuthentication, require_scopes
+from .models import Admin
 
 
 class IndexView(APIView):
@@ -36,7 +39,7 @@ class IndexView(APIView):
     authentication_classes = (JWTAuthentication,)
     permission_classes = ()
 
-    def get(self, request: HttpRequest, format: str | None = None) -> Response:
+    def get(self, request: Request, format: str | None = None) -> Response:
         """Return a hello from Python Discord forms!"""
         response_data = {
             "message": "Hello, world!",
@@ -56,3 +59,28 @@ class IndexView(APIView):
             }
 
         return Response(response_data)
+
+
+class AdminView(APIView):
+    """Manage administrators."""
+
+    authentication_classes = (JWTAuthentication,)
+    permission_classes = ()
+
+    @require_scopes({"admin"})
+    def post(self, request: Request, format: str | None = None) -> Response:
+        """Grant a user administrator privileges."""
+        if "id" not in request.data:
+            return Response(
+                {"error": "which-id-please"},
+                status=HTTPStatus.BAD_REQUEST,
+            )
+
+        try:
+            _ = Admin.objects.create(id=request.data["id"])
+            return Response({"status": "ok"})
+        except IntegrityError:
+            return Response(
+                {"error": "already_exists"},
+                status=HTTPStatus.BAD_REQUEST,
+            )