api/config/settings/common.py

@@ -156,6 +156,7 @@ LOCAL_APPS = (
     "funkwhale_api.requests",
     "funkwhale_api.favorites",
     "funkwhale_api.federation",
+    "funkwhale_api.moderation",
     "funkwhale_api.radios",
     "funkwhale_api.history",
     "funkwhale_api.playlists",

api/funkwhale_api/moderation/factories.py

+import uuid
+
+import factory
+import requests
+import requests_http_signature
+from django.conf import settings
+from django.utils import timezone
+from django.utils.http import http_date
+
+from funkwhale_api.factories import registry
+from funkwhale_api.users import factories as user_factories
+
+from . import keys, models
+
+registry.register(keys.get_key_pair, name="federation.KeyPair")
+
+
+@registry.register(name="federation.SignatureAuth")
+class SignatureAuthFactory(factory.Factory):
+    algorithm = "rsa-sha256"
+    key = factory.LazyFunction(lambda: keys.get_key_pair()[0])
+    key_id = factory.Faker("url")
+    use_auth_header = False
+    headers = ["(request-target)", "user-agent", "host", "date", "content-type"]
+
+    class Meta:
+        model = requests_http_signature.HTTPSignatureAuth
+
+
+@registry.register(name="federation.SignedRequest")
+class SignedRequestFactory(factory.Factory):
+    url = factory.Faker("url")
+    method = "get"
+    auth = factory.SubFactory(SignatureAuthFactory)
+
+    class Meta:
+        model = requests.Request
+
+    @factory.post_generation
+    def headers(self, create, extracted, **kwargs):
+        default_headers = {
+            "User-Agent": "Test",
+            "Host": "test.host",
+            "Date": http_date(timezone.now().timestamp()),
+            "Content-Type": "application/activity+json",
+        }
+        if extracted:
+            default_headers.update(extracted)
+        self.headers.update(default_headers)
+
+
+@registry.register(name="federation.Link")
+class LinkFactory(factory.Factory):
+    type = "Link"
+    href = factory.Faker("url")
+    mediaType = "text/html"
+
+    class Meta:
+        model = dict
+
+    class Params:
+        audio = factory.Trait(mediaType=factory.Iterator(["audio/mp3", "audio/ogg"]))
+
+
+def create_user(actor):
+    return user_factories.UserFactory(actor=actor)
+
+
+@registry.register
+class Domain(factory.django.DjangoModelFactory):
+    name = factory.Faker("domain_name")
+
+    class Meta:
+        model = "federation.Domain"
+        django_get_or_create = ("name",)
+
+
+@registry.register
+class ActorFactory(factory.DjangoModelFactory):
+    public_key = None
+    private_key = None
+    preferred_username = factory.Faker("user_name")
+    summary = factory.Faker("paragraph")
+    domain = factory.SubFactory(Domain)
+    fid = factory.LazyAttribute(
+        lambda o: "https://{}/users/{}".format(o.domain.name, o.preferred_username)
+    )
+    followers_url = factory.LazyAttribute(
+        lambda o: "https://{}/users/{}followers".format(
+            o.domain.name, o.preferred_username
+        )
+    )
+    inbox_url = factory.LazyAttribute(
+        lambda o: "https://{}/users/{}/inbox".format(
+            o.domain.name, o.preferred_username
+        )
+    )
+    outbox_url = factory.LazyAttribute(
+        lambda o: "https://{}/users/{}/outbox".format(
+            o.domain.name, o.preferred_username
+        )
+    )
+
+    class Meta:
+        model = models.Actor
+
+    @factory.post_generation
+    def local(self, create, extracted, **kwargs):
+        if not extracted and not kwargs:
+            return
+        from funkwhale_api.users.factories import UserFactory
+
+        self.domain = models.Domain.objects.get_or_create(
+            name=settings.FEDERATION_HOSTNAME
+        )[0]
+        self.save(update_fields=["domain"])
+        if not create:
+            if extracted and hasattr(extracted, "pk"):
+                extracted.actor = self
+            else:
+                UserFactory.build(actor=self, **kwargs)
+        if extracted and hasattr(extracted, "pk"):
+            extracted.actor = self
+            extracted.save(update_fields=["user"])
+        else:
+            self.user = UserFactory(actor=self, **kwargs)
+
+    @factory.post_generation
+    def keys(self, create, extracted, **kwargs):
+        if not create:
+            # Simple build, do nothing.
+            return
+        if not extracted:
+            private, public = keys.get_key_pair()
+            self.private_key = private.decode("utf-8")
+            self.public_key = public.decode("utf-8")
+
+
+@registry.register
+class FollowFactory(factory.DjangoModelFactory):
+    target = factory.SubFactory(ActorFactory)
+    actor = factory.SubFactory(ActorFactory)
+
+    class Meta:
+        model = models.Follow
+
+    class Params:
+        local = factory.Trait(actor=factory.SubFactory(ActorFactory, local=True))
+
+
+@registry.register
+class MusicLibraryFactory(factory.django.DjangoModelFactory):
+    actor = factory.SubFactory(ActorFactory)
+    privacy_level = "me"
+    name = factory.Faker("sentence")
+    description = factory.Faker("sentence")
+    uploads_count = 0
+    fid = factory.Faker("federation_url")
+
+    class Meta:
+        model = "music.Library"
+
+    class Params:
+        local = factory.Trait(actor=factory.SubFactory(ActorFactory, local=True))
+
+    @factory.post_generation
+    def followers_url(self, create, extracted, **kwargs):
+        if not create:
+            # Simple build, do nothing.
+            return
+
+        self.followers_url = extracted or self.fid + "/followers"

api/funkwhale_api/moderation/models.py

+from django.db import models
+from django.utils import timezone
+
+
+class InstancePolicyQuerySet(models.QuerySet):
+    def find_matching(self, type):
+        query = models.Q(block_all=True)
+        if type != "block_all":
+            query |= models.Q(**{type: True})
+
+        return self.filter(query, is_active=True)
+
+
+class InstancePolicy(models.Model):
+    # who created this policy
+    actor = models.ForeignKey(
+        "federation.Actor", related_name="instance_policies", on_delete=models.CASCADE
+    )
+
+    target_domain = models.OneToOneField(
+        "federation.Domain",
+        related_name="targeting_instance_policy",
+        on_delete=models.CASCADE,
+        null=True,
+        blank=True,
+    )
+    target_domain = models.OneToOneField(
+        "federation.Actor",
+        related_name="targeting_instance_policy",
+        on_delete=models.CASCADE,
+        null=True,
+        blank=True,
+    )
+    is_active = models.BooleanField(default=True)
+
+    # inspired from https://github.com/tootsuite/mastodon/issues/8186
+    # block = not communication at all (in or out)
+    # reject_media = reject media files (audio, avatar, attachments...)
+    # mute_content = hide content from non-followers
+    # mute_notifications = hide notifications from non-followers
+    block = models.BooleanField()
+    reject_media = models.BooleanField(default=False)
+    mute_content = models.BooleanField(default=False)
+    mute_notifications = models.BooleanField(default=False)
+
+    # why is the policy in place
+    summary = models.CharField(max_length=1000, null=True, blank=True)
+
+    creation_date = models.DateTimeField(default=timezone.now)
+    last_use_date = models.DateTimeField(null=True, blank=True)
+
+    dropped_activities

api/tests/federation/test_activity.py

@@ -46,6 +46,25 @@ def test_receive_validates_basic_attributes_and_stores_activity(factories, now,
         assert ii.is_read is False
 
 
+# def test_receive_discard_blocked_domain(factories, now, mocker):
+#     local_to_actor = factories["users.User"]().create_actor()
+#     local_cc_actor = factories["users.User"]().create_actor()
+#     remote_actor = factories["federation.Actor"]()
+#     policy = factories['moderation.InstancePolicy'](block_all=True, domain=remote_actor.domain)
+#     a = {
+#         "@context": [],
+#         "actor": remote_actor.fid,
+#         "type": "Noop",
+#         "id": "https://test.activity",
+#         "to": [local_to_actor.fid, remote_actor.fid],
+#         "cc": [local_cc_actor.fid, activity.PUBLIC_ADDRESS],
+#     }
+
+#     copy = activity.receive(activity=a, on_behalf_of=remote_actor)
+
+#     assert copy is None
+
+
 def test_get_actors_from_audience_urls(settings, db):
     settings.FEDERATION_HOSTNAME = "federation.hostname"
     library_uuid1 = uuid.uuid4()

api/tests/moderation/test_models.py

+import pytest
+
+
+@pytest.mark.parametrize(
+    "factory_kwargs, type, expected",
+    [
+        ({"block_all": True, "is_active": True}, "block_all", True),
+        ({"block_all": False, "is_active": True}, "block_all", False),
+        ({"block_all": True, "is_active": False}, "block_all", False),
+        ({"block_all": False, "is_active": True}, "reject_media", True),
+        (
+            {"block_all": False, "is_active": True, "reject_media": True},
+            "reject_media",
+            True,
+        ),
+        (
+            {"block_all": True, "is_active": True, "reject_media": False},
+            "reject_media",
+            True,
+        ),
+    ],
+)
+def test_filter_instance_policies(factories, factory_kwargs, type, expected):
+    policy = factories["moderation.InstancePolicy"](**factory_kwargs)
+
+    result = list(models.InstancePolicy.objects.find_matching(type))
+    if expected:
+        assert result == [policy]
+    else:
+        assert result == []