Check if incoming domains are blocked

bookwyrm/models/federated_server.py

@@ -2,16 +2,27 @@
 from django.db import models
 from .base_model import BookWyrmModel
 
+FederationStatus = models.TextChoices(
+    "Status",
+    [
+        "federated",
+        "blocked",
+    ],
+)
+
 
 class FederatedServer(BookWyrmModel):
     """ store which servers we federate with """
 
     server_name = models.CharField(max_length=255, unique=True)
-    # federated, blocked, whatever else
-    status = models.CharField(max_length=255, default="federated")
+    status = models.CharField(
+        max_length=255, default="federated", choices=FederationStatus.choices
+    )
     # is it mastodon, bookwyrm, etc
     application_type = models.CharField(max_length=255, null=True)
     application_version = models.CharField(max_length=255, null=True)
 
-
-# TODO: blocked servers
+    def block(self):
+        """ block a server """
+        self.status = "blocked"
+        self.save()

bookwyrm/views/inbox.py

@@ -1,9 +1,10 @@
 """ incoming activities """
 import json
-from urllib.parse import urldefrag
+import re
+from urllib.parse import urldefrag, urlparse
 
-from django.http import HttpResponse
-from django.http import HttpResponseBadRequest, HttpResponseNotFound
+from django.http import HttpResponse, HttpResponseNotFound
+from django.http import HttpResponseBadRequest, HttpResponseForbidden
 from django.utils.decorators import method_decorator
 from django.views import View
 from django.views.decorators.csrf import csrf_exempt
@@ -12,6 +13,7 @@ import requests
 from bookwyrm import activitypub, models
 from bookwyrm.tasks import app
 from bookwyrm.signatures import Signature
+from bookwyrm.utils import regex
 
 
 @method_decorator(csrf_exempt, name="dispatch")
@@ -21,6 +23,10 @@ class Inbox(View):
 
     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()
+
         # make sure the user's inbox even exists
         if username:
             try:
@@ -34,6 +40,10 @@ class Inbox(View):
         except json.decoder.JSONDecodeError:
             return HttpResponseBadRequest()
 
+        # let's be extra sure we didn't block this domain
+        if is_blocked_activity(activity_json):
+            return HttpResponseForbidden()
+
         if (
             not "object" in activity_json
             or not "type" in activity_json
@@ -54,6 +64,34 @@ class Inbox(View):
         return HttpResponse()
 
 
+def 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")
+    domain = re.match(regex.domain, user_agent)
+    if not domain:
+        # idk, we'll try again later with the actor
+        return False
+    return is_blocked(domain)
+
+
+def is_blocked_activity(activity_json):
+    """ get the sender out of activity json and check if it's blocked """
+    actor = activity_json.get("actor")
+    if not actor:
+        # well I guess it's not even a valid activity so who knows
+        return False
+    url = urlparse(actor)
+    return is_blocked(url.netloc)
+
+
+def is_blocked(domain):
+    """ is this domain blocked? """
+    return models.FederatedServer.object.filter(
+        server_name=domain, status="blocked"
+    ).exists()
+
+
 @app.task
 def activity_task(activity_json):
     """ do something with this json we think is legit """