Tighten legacy v1 API boundaries

apps/downloads/api.py

@@ -6,6 +6,7 @@
 from rest_framework.response import Response
 from tastypie import fields
 from tastypie.constants import ALL, ALL_WITH_RELATIONS
+from tastypie.exceptions import BadRequest
 
 from apps.downloads.models import OS, Release, ReleaseFile
 from apps.downloads.serializers import OSSerializer, ReleaseFileSerializer, ReleaseSerializer
@@ -88,6 +89,7 @@ class Meta(GenericResource.Meta):
 
         queryset = ReleaseFile.objects.all()
         resource_name = "downloads/release_file"
+        list_allowed_methods = ["get", "post", "delete"]
         fields = [
             "name",
             "slug",
@@ -117,6 +119,13 @@ class Meta(GenericResource.Meta):
         }
         abstract = False
 
+    def delete_list(self, request, **kwargs):
+        """Delete release files only when scoped to a single release."""
+        if set(request.GET) != {"release"} or len(request.GET.getlist("release")) != 1:
+            msg = "Deleting release files requires exactly one 'release' filter."
+            raise BadRequest(msg)
+        return super().delete_list(request, **kwargs)
+
 
 # Django Rest Framework
 

apps/downloads/tests/test_views.py

@@ -6,7 +6,7 @@
 from django.urls import reverse
 from rest_framework.test import APITestCase
 
-from apps.downloads.models import Release
+from apps.downloads.models import OS, Release
 from apps.downloads.tests.base import BaseDownloadTests, DownloadMixin
 from apps.pages.factories import PageFactory
 from apps.users.factories import UserFactory
@@ -485,6 +485,121 @@ def setUp(self):
         self.Authorization = f"{self.token_header} {self.staff_user.username}:{self.staff_key}"
         self.Authorization_invalid = f"{self.token_header} invalid:token"
 
+    def test_staff_session_cannot_write_without_api_key(self):
+        self.client.force_login(self.staff_user)
+        url = self.create_url("os")
+        data = {
+            "name": "Session-only OS",
+            "slug": "session-only",
+        }
+
+        response = self.json_client("post", url, data)
+
+        self.assertEqual(response.status_code, 401)
+        self.assertFalse(OS.objects.filter(slug="session-only").exists())
+
+    def test_staff_session_cannot_write_with_malformed_api_key_header(self):
+        self.client.force_login(self.staff_user)
+        url = self.create_url("os")
+        data = {
+            "name": "Malformed API key OS",
+            "slug": "malformed-api-key",
+        }
+
+        response = self.json_client("post", url, data, HTTP_AUTHORIZATION="ApiKey malformed")
+
+        self.assertEqual(response.status_code, 401)
+        self.assertFalse(OS.objects.filter(slug="malformed-api-key").exists())
+
+    def test_inactive_staff_api_key_cannot_write(self):
+        self.staff_user.is_active = False
+        self.staff_user.save()
+        url = self.create_url("os")
+        data = {
+            "name": "Inactive API key OS",
+            "slug": "inactive-api-key",
+        }
+
+        response = self.json_client("post", url, data, HTTP_AUTHORIZATION=self.Authorization)
+
+        self.assertEqual(response.status_code, 401)
+        self.assertFalse(OS.objects.filter(slug="inactive-api-key").exists())
+
+    def test_query_string_api_key_cannot_write(self):
+        url = self.create_url(
+            "os",
+            filters={
+                "username": self.staff_user.username,
+                "api_key": self.staff_key,
+            },
+        )
+        data = {
+            "name": "Query string API key OS",
+            "slug": "query-string-api-key",
+        }
+
+        response = self.json_client("post", url, data)
+
+        self.assertEqual(response.status_code, 401)
+        self.assertFalse(OS.objects.filter(slug="query-string-api-key").exists())
+
+    def test_form_api_key_cannot_write(self):
+        url = self.create_url("os")
+        response = self.client.post(
+            url,
+            {
+                "username": self.staff_user.username,
+                "api_key": self.staff_key,
+                "name": "Form API key OS",
+                "slug": "form-api-key",
+            },
+        )
+
+        self.assertEqual(response.status_code, 401)
+        self.assertFalse(OS.objects.filter(slug="form-api-key").exists())
+
+    def test_v1_list_update_methods_are_not_allowed(self):
+        url = self.create_url("os")
+
+        response = self.json_client("put", url, HTTP_AUTHORIZATION=self.Authorization)
+        self.assertEqual(response.status_code, 405)
+
+        response = self.json_client("patch", url, HTTP_AUTHORIZATION=self.Authorization)
+        self.assertEqual(response.status_code, 405)
+
+    def test_v1_collection_delete_requires_release_file_scope(self):
+        url = self.create_url("os")
+        response = self.json_client("delete", url, HTTP_AUTHORIZATION=self.Authorization)
+        self.assertEqual(response.status_code, 405)
+
+        url = self.create_url("release_file")
+        response = self.json_client("delete", url, HTTP_AUTHORIZATION=self.Authorization)
+        self.assertEqual(response.status_code, 400)
+
+        url = self.create_url(
+            "release_file",
+            filters={
+                "release": self.release_275.pk,
+                "os": self.linux.pk,
+            },
+        )
+        response = self.json_client("delete", url, HTTP_AUTHORIZATION=self.Authorization)
+        self.assertEqual(response.status_code, 400)
+
+        url = f"{self.create_url('release_file')}?release={self.release_275.pk}&release={self.draft_release.pk}"
+        response = self.json_client("delete", url, HTTP_AUTHORIZATION=self.Authorization)
+        self.assertEqual(response.status_code, 400)
+
+    def test_v1_release_file_delete_by_release_still_works(self):
+        url = self.create_url("release_file", filters={"release": self.release_275.pk})
+
+        response = self.json_client("delete", url, HTTP_AUTHORIZATION=self.Authorization)
+
+        self.assertEqual(response.status_code, 204)
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(self.get_json(response)), 0)
+
 
 class DownloadApiV2ViewsTest(BaseDownloadApiViewsTest, BaseDownloadTests, APITestCase):
     api_version = "v2"

pydotorg/resources.py

@@ -1,13 +1,17 @@
 """Tastypie API resource classes and authentication/authorization backends."""
 
 from django.contrib.auth import get_user_model
+from django.contrib.auth.models import AnonymousUser
 from tastypie.authentication import ApiKeyAuthentication
 from tastypie.authorization import Authorization
 from tastypie.exceptions import Unauthorized
 from tastypie.http import HttpUnauthorized
 from tastypie.resources import ModelResource
 from tastypie.throttle import CacheThrottle
 
+API_KEY_AUTHENTICATED_ATTR = "_pydotorg_api_key_authenticated"
+LEGACY_CREDENTIAL_PARAMS = frozenset(("username", "api_key"))
+
 
 class ApiKeyOrGuestAuthentication(ApiKeyAuthentication):
     """Authentication backend that falls back to guest access when no API key is provided."""
@@ -18,19 +22,33 @@ def is_authenticated(self, request, **kwargs):
         Copypasted from tastypie, modified to avoid issues with
         app-loading and custom user model.
         """
-        User = get_user_model()  # noqa: N806 - Django convention for user model reference
-        username_field = User.USERNAME_FIELD
+        setattr(request, API_KEY_AUTHENTICATED_ATTR, False)
+
+        if self._has_legacy_credentials(request):
+            return HttpUnauthorized()
+
+        if not request.headers.get("authorization"):
+            return self._authenticate_guest(request)
 
-        # Note that it's only safe to return 'True'
-        # in the guest case. If there is an API key supplied
-        # then we must not return 'True' unless the
-        # API key is valid.
         try:
             username, api_key = self.extract_credentials(request)
         except ValueError:
-            return True  # Allow guests.
+            return HttpUnauthorized()
+
         if not username or not api_key:
-            return True  # Allow guests.
+            return HttpUnauthorized()
+
+        User = get_user_model()  # noqa: N806 - Django convention for user model reference
+        return self._authenticate_api_key(request, username, api_key, User.USERNAME_FIELD)
+
+    def _authenticate_guest(self, request):
+        """Reset session identity and allow anonymous guest access."""
+        request.user = AnonymousUser()
+        return True
+
+    def _authenticate_api_key(self, request, username, api_key, username_field):
+        """Authenticate API-key credentials and mark successful requests."""
+        User = get_user_model()  # noqa: N806 - Django convention for user model reference
 
         # IMPORTANT: Beyond this point we are no longer
         # handling the guest case, so all incorrect usernames
@@ -48,19 +66,40 @@ def is_authenticated(self, request, **kwargs):
         key_auth_check = self.get_key(user, api_key)
         if key_auth_check and not isinstance(key_auth_check, HttpUnauthorized):
             request.user = user
+            setattr(request, API_KEY_AUTHENTICATED_ATTR, True)
 
         return key_auth_check
 
+    def extract_credentials(self, request):
+        """Return API key credentials from the 'Authorization' header only."""
+        data = self.get_authorization_data(request)
+        if data.count(":") != 1:
+            msg = "API key credentials must use the username:key format."
+            raise ValueError(msg)
+        username, api_key = data.split(":", 1)
+        if not username or not api_key:
+            msg = "API key credentials must include both username and key."
+            raise ValueError(msg)
+        return username, api_key
+
     def get_identifier(self, request):
         """Return the username for authenticated users or IP/hostname for guests."""
-        if request.user.is_authenticated:
+        user = getattr(request, "user", None)
+        if user is not None and user.is_authenticated:
             return super().get_identifier(request)
         # returns a combination of IP address and hostname.
-        return "{}_{}".format(request.META.get("REMOTE_ADDR", "noaddr"), request.META.get("REMOTE_HOST", "nohost"))
+        return "{}_{}".format(
+            request.META.get("REMOTE_ADDR", "noaddr"),
+            request.META.get("REMOTE_HOST", "nohost"),
+        )
 
-    def check_active(self, user):
-        """Return True, allowing inactive users to authenticate."""
-        return True
+    def _has_legacy_credentials(self, request):
+        """Return True when credentials are supplied outside the 'Authorization' header."""
+        return any(
+            param in credential_source
+            for credential_source in (request.GET, request.POST)
+            for param in LEGACY_CREDENTIAL_PARAMS
+        )
 
 
 class StaffAuthorization(Authorization):
@@ -78,40 +117,44 @@ def read_detail(self, object_list, bundle):
 
     def create_list(self, object_list, bundle):
         """Allow only staff users to create objects in bulk."""
-        if bundle.request.user.is_staff:
+        if self._is_authenticated_staff_via_api_key(bundle.request):
             return object_list
         msg = "Operation restricted to staff users."
         raise Unauthorized(msg)
 
     def create_detail(self, object_list, bundle):
         """Allow only staff users to create individual objects."""
-        return bundle.request.user.is_staff
+        return self._is_authenticated_staff_via_api_key(bundle.request)
 
     def update_list(self, object_list, bundle):
         """Allow only staff users to update objects in bulk."""
-        if bundle.request.user.is_staff:
+        if self._is_authenticated_staff_via_api_key(bundle.request):
             return object_list
         msg = "Operation restricted to staff users."
         raise Unauthorized(msg)
 
     def update_detail(self, object_list, bundle):
         """Allow only staff users to update individual objects."""
-        return bundle.request.user.is_staff
+        return self._is_authenticated_staff_via_api_key(bundle.request)
 
     def delete_list(self, object_list, bundle):
         """Allow only staff users to delete objects in bulk."""
-        if not bundle.request.user.is_staff:
+        if not self._is_authenticated_staff_via_api_key(bundle.request):
             msg = "Operation restricted to staff users."
             raise Unauthorized(msg)
         return object_list
 
     def delete_detail(self, object_list, bundle):
         """Allow only staff users to delete individual objects."""
-        if not bundle.request.user.is_staff:
+        if not self._is_authenticated_staff_via_api_key(bundle.request):
             msg = "Operation restricted to staff users."
             raise Unauthorized(msg)
         return True
 
+    def _is_authenticated_staff_via_api_key(self, request):
+        """Return True only for staff authenticated by v1 API key, not cookies."""
+        return request.user.is_staff and getattr(request, API_KEY_AUTHENTICATED_ATTR, False) is True
+
 
 class OnlyPublishedAuthorization(StaffAuthorization):
     """Only staff users can see unpublished objects."""
@@ -137,5 +180,7 @@ class Meta:
 
         authentication = ApiKeyOrGuestAuthentication()
         authorization = StaffAuthorization()
+        list_allowed_methods = ["get", "post"]
+        detail_allowed_methods = ["get", "delete"]
         throttle = CacheThrottle(throttle_at=600)  # default is 150 req/hr
         abstract = True