Add support to import blocklists (#617)

static/css/style.css

@@ -1358,6 +1358,10 @@ table.metadata td .emoji {
     cursor: pointer;
 }
 
+.message.error {
+    background-color: var(--color-bg-error);
+}
+
 /* Identity banner */
 
 .identity-banner {

takahe/urls.py

@@ -133,6 +133,11 @@ urlpatterns = [
         admin.FederationRoot.as_view(),
         name="admin_federation",
     ),
+    path(
+        "admin/federation/blocklist/",
+        admin.FederationBlocklist.as_view(),
+        name="admin_federation_blocklist",
+    ),
     path(
         "admin/federation/<domain>/",
         admin.FederationEdit.as_view(),

templates/admin/federation.html

@@ -8,6 +8,9 @@
         <input type="search" name="query" value="{{ query }}" placeholder="Search by domain">
         <button><i class="fa-solid fa-search"></i></button>
     </form>
+    <div class="view-options">
+        <a href="{% url "admin_federation_blocklist" %}?page={{ page_obj.number }}" class="button">Import Blocklist</a>
+    </div>
     <table class="items">
         {% for domain in page_obj %}
             <tr>

templates/admin/federation_blocklist.html

@@ -0,0 +1,18 @@
+{% extends "admin/base_main.html" %}
+
+{% block subtitle %}Federation Blocklist{% endblock %}
+
+{% block settings_content %}
+    <form action="." method="POST" enctype="multipart/form-data">
+        {% csrf_token %}
+        <h1>Import Blocklist</h1>
+
+        <fieldset>
+            {% include "forms/_field.html" with field=form.blocklist %}
+        </fieldset>
+        <div class="buttons">
+            <a href="{% url "admin_federation" %}?page={{ page }}" class="button secondary left">Back</a>
+            <button>Save</button>
+        </div>
+    </form>
+{% endblock %}

tests/users/services/test_domain.py

@@ -0,0 +1,11 @@
+import pytest
+
+from users.models import Domain
+from users.services import DomainService
+
+
+@pytest.mark.django_db
+def test_block():
+    DomainService.block(["block1.example.com", "block2.example.com"])
+
+    assert Domain.objects.filter(blocked=True).count() == 2

users/services/__init__.py

@@ -1,3 +1,4 @@
 from .announcement import AnnouncementService  # noqa
+from .domain import DomainService  # noqa
 from .identity import IdentityService  # noqa
 from .user import UserService  # noqa

users/services/domain.py

@@ -0,0 +1,22 @@
+from users.models import Domain
+
+
+class DomainService:
+    """
+    High-level domain handling methods
+    """
+
+    @classmethod
+    def block(cls, domains: list[str]) -> None:
+        domains_to_block = Domain.objects.filter(domain__in=domains)
+        domains_to_block.update(blocked=True)
+
+        already_blocked = domains_to_block.values_list("domain", flat=True)
+        domains_to_create = []
+        for domain in domains:
+            if domain not in already_blocked:
+                domains_to_create.append(
+                    Domain(domain=domain, blocked=True, local=False)
+                )
+
+        Domain.objects.bulk_create(domains_to_create)

users/views/admin/__init__.py

@@ -23,7 +23,11 @@ from users.views.admin.emoji import (  # noqa
     EmojiEnable,
     EmojiRoot,
 )
-from users.views.admin.federation import FederationEdit, FederationRoot  # noqa
+from users.views.admin.federation import (  # noqa
+    FederationBlocklist,
+    FederationEdit,
+    FederationRoot,
+)
 from users.views.admin.hashtags import HashtagEdit, HashtagEnable, Hashtags  # noqa
 from users.views.admin.identities import IdentitiesRoot, IdentityEdit  # noqa
 from users.views.admin.invites import InviteCreate, InvitesRoot, InviteView  # noqa

users/views/admin/federation.py

@@ -1,4 +1,8 @@
+import csv
+
 from django import forms
+from django.contrib import messages
+from django.core.validators import FileExtensionValidator, ValidationError
 from django.db import models
 from django.shortcuts import get_object_or_404, redirect
 from django.utils.decorators import method_decorator
@@ -6,11 +10,12 @@ from django.views.generic import FormView, ListView
 
 from users.decorators import admin_required
 from users.models import Domain
+from users.services import DomainService
+from users.views.admin.domains import DomainValidator
 
 
 @method_decorator(admin_required, name="dispatch")
 class FederationRoot(ListView):
-
     template_name = "admin/federation.html"
     paginate_by = 50
 
@@ -35,7 +40,6 @@ class FederationRoot(ListView):
 
 @method_decorator(admin_required, name="dispatch")
 class FederationEdit(FormView):
-
     template_name = "admin/federation_edit.html"
     extra_context = {"section": "federation"}
 
@@ -78,3 +82,61 @@ class FederationEdit(FormView):
             "blocked": self.domain.blocked,
             "notes": self.domain.notes,
         }
+
+
+@method_decorator(admin_required, name="dispatch")
+class FederationBlocklist(FormView):
+    template_name = "admin/federation_blocklist.html"
+    extra_context = {"section": "federation"}
+    error_msg = "The uploaded file has an invalid blocklist CSV format."
+    success_msg = "The blocklist CSV was processed processed with success!"
+
+    class form_class(forms.Form):
+        blocklist = forms.FileField(
+            help_text=(
+                "Blocklist file with one domain per line. "
+                "Oliphant blocklist format is also supported."
+            ),
+            validators=[FileExtensionValidator(allowed_extensions=["txt", "csv"])],
+        )
+
+    def form_valid(self, form):
+        validator = DomainValidator()
+        domains = []
+
+        try:
+            lines = form.cleaned_data["blocklist"].read().decode("utf-8").splitlines()
+
+            if "#domain" in lines[0]:
+                reader = csv.DictReader(lines)
+            else:
+                reader = csv.DictReader(lines, fieldnames=["#domain"])
+
+            for row in reader:
+                domain = row["#domain"].strip()
+
+                try:
+                    validator(domain)
+                except ValidationError:
+                    # skip adding invalid domain
+                    # to the blocklist
+                    continue
+
+                domains.append(domain)
+        except (TypeError, ValueError):
+            messages.error(self.request, self.error_msg)
+            return redirect(".")
+
+        if not domains:
+            messages.error(self.request, self.error_msg)
+            return redirect(".")
+
+        DomainService.block(domains)
+
+        messages.success(self.request, self.success_msg)
+        return redirect("admin_federation")
+
+    def get_context_data(self, **kwargs):
+        context = super().get_context_data(**kwargs)
+        context["page"] = self.request.GET.get("page")
+        return context