Verified Commit 979a053c authored by Agate's avatar Agate 💬

WIP

parent 7d56e97f
......@@ -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",
......
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"
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
......@@ -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()
......
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 == []
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment