Updates inbox view

bookwyrm/tests/views/inbox/test_inbox.py

@@ -3,6 +3,7 @@ import json
 import pathlib
 from unittest.mock import patch
 
+from django.core.exceptions import PermissionDenied
 from django.http import HttpResponseNotAllowed, HttpResponseNotFound
 from django.test import TestCase, Client
 from django.test.client import RequestFactory
@@ -130,22 +131,24 @@ class Inbox(TestCase):
             "",
             HTTP_USER_AGENT="http.rb/4.4.1 (Mastodon/3.3.0; +https://mastodon.social/)",
         )
-        self.assertFalse(views.inbox.is_blocked_user_agent(request))
+        self.assertIsNone(views.inbox.raise_is_blocked_user_agent(request))
 
         models.FederatedServer.objects.create(
             server_name="mastodon.social", status="blocked"
         )
-        self.assertTrue(views.inbox.is_blocked_user_agent(request))
+        with self.assertRaises(PermissionDenied):
+            views.inbox.raise_is_blocked_user_agent(request)
 
     def test_is_blocked_activity(self):
         """check for blocked servers"""
         activity = {"actor": "https://mastodon.social/user/whaatever/else"}
-        self.assertFalse(views.inbox.is_blocked_activity(activity))
+        self.assertIsNone(views.inbox.raise_is_blocked_activity(activity))
 
         models.FederatedServer.objects.create(
             server_name="mastodon.social", status="blocked"
         )
-        self.assertTrue(views.inbox.is_blocked_activity(activity))
+        with self.assertRaises(PermissionDenied):
+            views.inbox.raise_is_blocked_activity(activity)
 
     @patch("bookwyrm.suggested_users.remove_user_task.delay")
     def test_create_by_deactivated_user(self, _):
@@ -157,11 +160,11 @@ class Inbox(TestCase):
         activity = self.create_json
         activity["actor"] = self.remote_user.remote_id
         activity["object"] = status_data
+        activity["type"] = "Create"
 
-        with patch("bookwyrm.views.inbox.has_valid_signature") as mock_valid:
-            mock_valid.return_value = True
-
-            result = self.client.post(
-                "/inbox", json.dumps(activity), content_type="application/json"
+        response = self.client.post(
+            "/inbox",
+            json.dumps(activity),
+            content_type="application/json",
         )
-        self.assertEqual(result.status_code, 403)
+        self.assertEqual(response.status_code, 403)

bookwyrm/views/inbox.py

@@ -3,8 +3,9 @@ import json
 import re
 from urllib.parse import urldefrag
 
-from django.http import HttpResponse, HttpResponseNotFound
-from django.http import HttpResponseBadRequest, HttpResponseForbidden
+from django.http import HttpResponse, Http404
+from django.core.exceptions import BadRequest, PermissionDenied
+from django.shortcuts import get_object_or_404
 from django.utils.decorators import method_decorator
 from django.views import View
 from django.views.decorators.csrf import csrf_exempt
@@ -21,36 +22,30 @@ from bookwyrm.utils import regex
 class Inbox(View):
     """requests sent by outside servers"""
 
-    # pylint: disable=too-many-return-statements
     def post(self, request, username=None):
         """only works as POST request"""
         # first check if this server is on our shitlist
-        if is_blocked_user_agent(request):
-            return HttpResponseForbidden()
+        raise_is_blocked_user_agent(request)
 
         # make sure the user's inbox even exists
         if username:
-            try:
-                models.User.objects.get(localname=username)
-            except models.User.DoesNotExist:
-                return HttpResponseNotFound()
+            get_object_or_404(models.User, localname=username, is_active=True)
 
         # is it valid json? does it at least vaguely resemble an activity?
         try:
             activity_json = json.loads(request.body)
         except json.decoder.JSONDecodeError:
-            return HttpResponseBadRequest()
+            raise BadRequest()
 
         # let's be extra sure we didn't block this domain
-        if is_blocked_activity(activity_json):
-            return HttpResponseForbidden()
+        raise_is_blocked_activity(activity_json)
 
         if (
             not "object" in activity_json
             or not "type" in activity_json
             or not activity_json["type"] in activitypub.activity_objects
         ):
-            return HttpResponseNotFound()
+            raise Http404()
 
         # verify the signature
         if not has_valid_signature(request, activity_json):
@@ -65,32 +60,35 @@ class Inbox(View):
         return HttpResponse()
 
 
-def is_blocked_user_agent(request):
+def raise_is_blocked_user_agent(request):
     """check if a request is from a blocked server based on user agent"""
     # check user agent
     user_agent = request.headers.get("User-Agent")
     if not user_agent:
-        return False
+        return
     url = re.search(rf"https?://{regex.DOMAIN}/?", user_agent)
     if not url:
-        return False
+        return
     url = url.group()
-    return models.FederatedServer.is_blocked(url)
+    if models.FederatedServer.is_blocked(url):
+        raise PermissionDenied()
 
 
-def is_blocked_activity(activity_json):
+def raise_is_blocked_activity(activity_json):
     """get the sender out of activity json and check if it's blocked"""
     actor = activity_json.get("actor")
 
     # check if the user is banned/deleted
     existing = models.User.find_existing_by_remote_id(actor)
     if existing and existing.deleted:
-        return True
+        raise PermissionDenied()
 
     if not actor:
         # well I guess it's not even a valid activity so who knows
-        return False
-    return models.FederatedServer.is_blocked(actor)
+        return
+
+    if models.FederatedServer.is_blocked(actor):
+        raise PermissionDenied()
 
 
 @app.task(queue="medium_priority")