kopia lustrzana https://dev.funkwhale.audio/funkwhale/funkwhale
530 wiersze
19 KiB
Python
530 wiersze
19 KiB
Python
from django import forms
|
|
from django.conf import settings
|
|
from django.core import paginator
|
|
from django.db.models import Prefetch
|
|
from django.http import HttpResponse
|
|
from django.urls import reverse
|
|
from rest_framework import exceptions, mixins, permissions, response, viewsets
|
|
from rest_framework.decorators import action
|
|
|
|
from funkwhale_api.common import preferences
|
|
from funkwhale_api.common import utils as common_utils
|
|
from funkwhale_api.federation import utils as federation_utils
|
|
from funkwhale_api.moderation import models as moderation_models
|
|
from funkwhale_api.music import models as music_models
|
|
from funkwhale_api.music import utils as music_utils
|
|
|
|
from . import (
|
|
activity,
|
|
actors,
|
|
authentication,
|
|
models,
|
|
renderers,
|
|
serializers,
|
|
utils,
|
|
webfinger,
|
|
)
|
|
|
|
|
|
def redirect_to_html(public_url):
|
|
response = HttpResponse(status=302)
|
|
response["Location"] = common_utils.join_url(settings.FUNKWHALE_URL, public_url)
|
|
return response
|
|
|
|
|
|
def get_collection_response(
|
|
conf, querystring, collection_serializer, page_access_check=None
|
|
):
|
|
page = querystring.get("page")
|
|
if page is None:
|
|
data = collection_serializer.data
|
|
else:
|
|
if page_access_check and not page_access_check():
|
|
raise exceptions.AuthenticationFailed(
|
|
"You do not have access to this resource"
|
|
)
|
|
try:
|
|
page_number = int(page)
|
|
except Exception:
|
|
return response.Response({"page": ["Invalid page number"]}, status=400)
|
|
conf["page_size"] = preferences.get("federation__collection_page_size")
|
|
p = paginator.Paginator(conf["items"], conf["page_size"])
|
|
try:
|
|
page = p.page(page_number)
|
|
conf["page"] = page
|
|
serializer = serializers.CollectionPageSerializer(conf)
|
|
data = serializer.data
|
|
except paginator.EmptyPage:
|
|
return response.Response(status=404)
|
|
|
|
return response.Response(data)
|
|
|
|
|
|
class AuthenticatedIfAllowListEnabled(permissions.BasePermission):
|
|
def has_permission(self, request, view):
|
|
allow_list_enabled = preferences.get("moderation__allow_list_enabled")
|
|
if not allow_list_enabled:
|
|
return True
|
|
return bool(request.actor)
|
|
|
|
|
|
class FederationMixin:
|
|
permission_classes = [AuthenticatedIfAllowListEnabled]
|
|
|
|
def dispatch(self, request, *args, **kwargs):
|
|
if not preferences.get("federation__enabled"):
|
|
return HttpResponse(status=405)
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
|
|
class SharedViewSet(FederationMixin, viewsets.GenericViewSet):
|
|
authentication_classes = [authentication.SignatureAuthentication]
|
|
renderer_classes = renderers.get_ap_renderers()
|
|
|
|
@action(
|
|
methods=["post"],
|
|
detail=False,
|
|
content_negotiation_class=renderers.IgnoreClientContentNegotiation,
|
|
)
|
|
def inbox(self, request, *args, **kwargs):
|
|
if request.method.lower() == "post" and request.actor is None:
|
|
raise exceptions.AuthenticationFailed(
|
|
"You need a valid signature to send an activity"
|
|
)
|
|
if request.method.lower() == "post":
|
|
activity.receive(activity=request.data, on_behalf_of=request.actor)
|
|
return response.Response({}, status=200)
|
|
|
|
|
|
class ActorViewSet(FederationMixin, mixins.RetrieveModelMixin, viewsets.GenericViewSet):
|
|
lookup_field = "preferred_username"
|
|
authentication_classes = [authentication.SignatureAuthentication]
|
|
renderer_classes = renderers.get_ap_renderers()
|
|
queryset = (
|
|
models.Actor.objects.local()
|
|
.select_related("user", "channel__artist", "channel__attributed_to")
|
|
.prefetch_related("channel__artist__tagged_items__tag")
|
|
)
|
|
serializer_class = serializers.ActorSerializer
|
|
|
|
def get_queryset(self):
|
|
queryset = super().get_queryset()
|
|
return queryset.exclude(channel__attributed_to=actors.get_service_actor())
|
|
|
|
def get_permissions(self):
|
|
# cf #1999 it must be possible to fetch actors without being authenticated
|
|
# otherwise we end up in a loop
|
|
if self.action == "retrieve":
|
|
return []
|
|
return super().get_permissions()
|
|
|
|
def retrieve(self, request, *args, **kwargs):
|
|
instance = self.get_object()
|
|
if utils.should_redirect_ap_to_html(request.headers.get("accept")):
|
|
if instance.get_channel():
|
|
return redirect_to_html(instance.channel.get_absolute_url())
|
|
return redirect_to_html(instance.get_absolute_url())
|
|
|
|
serializer = self.get_serializer(instance)
|
|
return response.Response(serializer.data)
|
|
|
|
@action(
|
|
methods=["get", "post"],
|
|
detail=True,
|
|
content_negotiation_class=renderers.IgnoreClientContentNegotiation,
|
|
)
|
|
def inbox(self, request, *args, **kwargs):
|
|
inbox_actor = self.get_object()
|
|
if request.method.lower() == "post" and request.actor is None:
|
|
raise exceptions.AuthenticationFailed(
|
|
"You need a valid signature to send an activity"
|
|
)
|
|
if request.method.lower() == "post":
|
|
activity.receive(
|
|
activity=request.data,
|
|
on_behalf_of=request.actor,
|
|
inbox_actor=inbox_actor,
|
|
)
|
|
return response.Response({}, status=200)
|
|
|
|
@action(methods=["get", "post"], detail=True)
|
|
def outbox(self, request, *args, **kwargs):
|
|
actor = self.get_object()
|
|
channel = actor.get_channel()
|
|
if channel:
|
|
return self.get_channel_outbox_response(request, channel)
|
|
return response.Response({}, status=200)
|
|
|
|
def get_channel_outbox_response(self, request, channel):
|
|
conf = {
|
|
"id": channel.actor.outbox_url,
|
|
"actor": channel.actor,
|
|
"items": channel.library.uploads.for_federation()
|
|
.order_by("-creation_date")
|
|
.prefetch_related("library__channel__actor", "track__artist"),
|
|
"item_serializer": serializers.ChannelCreateUploadSerializer,
|
|
}
|
|
return get_collection_response(
|
|
conf=conf,
|
|
querystring=request.GET,
|
|
collection_serializer=serializers.ChannelOutboxSerializer(channel),
|
|
)
|
|
|
|
@action(methods=["get"], detail=True)
|
|
def followers(self, request, *args, **kwargs):
|
|
self.get_object()
|
|
# XXX to implement
|
|
return response.Response({})
|
|
|
|
@action(methods=["get"], detail=True)
|
|
def following(self, request, *args, **kwargs):
|
|
self.get_object()
|
|
# XXX to implement
|
|
return response.Response({})
|
|
|
|
|
|
class EditViewSet(FederationMixin, mixins.RetrieveModelMixin, viewsets.GenericViewSet):
|
|
lookup_field = "uuid"
|
|
authentication_classes = [authentication.SignatureAuthentication]
|
|
renderer_classes = renderers.get_ap_renderers()
|
|
# queryset = common_models.Mutation.objects.local().select_related()
|
|
# serializer_class = serializers.ActorSerializer
|
|
|
|
|
|
class ReportViewSet(
|
|
FederationMixin, mixins.RetrieveModelMixin, viewsets.GenericViewSet
|
|
):
|
|
lookup_field = "uuid"
|
|
authentication_classes = [authentication.SignatureAuthentication]
|
|
renderer_classes = renderers.get_ap_renderers()
|
|
queryset = moderation_models.Report.objects.none()
|
|
|
|
|
|
class WellKnownViewSet(viewsets.GenericViewSet):
|
|
authentication_classes = []
|
|
permission_classes = []
|
|
renderer_classes = [renderers.JSONRenderer, renderers.WebfingerRenderer]
|
|
|
|
@action(methods=["get"], detail=False)
|
|
def nodeinfo(self, request, *args, **kwargs):
|
|
data = {
|
|
"links": [
|
|
{
|
|
"rel": "http://nodeinfo.diaspora.software/ns/schema/2.0",
|
|
"href": utils.full_url(reverse("api:v1:instance:nodeinfo-2.0")),
|
|
}
|
|
]
|
|
}
|
|
return response.Response(data)
|
|
|
|
@action(methods=["get"], detail=False)
|
|
def webfinger(self, request, *args, **kwargs):
|
|
if not preferences.get("federation__enabled"):
|
|
return HttpResponse(status=405)
|
|
try:
|
|
resource_type, resource = webfinger.clean_resource(request.GET["resource"])
|
|
cleaner = getattr(webfinger, f"clean_{resource_type}")
|
|
result = cleaner(resource)
|
|
handler = getattr(self, f"handler_{resource_type}")
|
|
data = handler(result)
|
|
except forms.ValidationError as e:
|
|
return response.Response({"errors": {"resource": e.message}}, status=400)
|
|
except KeyError:
|
|
return response.Response(
|
|
{"errors": {"resource": "This field is required"}}, status=400
|
|
)
|
|
|
|
return response.Response(data)
|
|
|
|
def handler_acct(self, clean_result):
|
|
username, hostname = clean_result
|
|
|
|
try:
|
|
actor = models.Actor.objects.local().get(preferred_username=username)
|
|
except models.Actor.DoesNotExist:
|
|
raise forms.ValidationError("Invalid username")
|
|
|
|
return serializers.ActorWebfingerSerializer(actor).data
|
|
|
|
|
|
def has_library_access(request, library):
|
|
if library.privacy_level == "everyone":
|
|
return True
|
|
if request.user.is_authenticated and request.user.is_superuser:
|
|
return True
|
|
|
|
try:
|
|
actor = request.actor
|
|
except AttributeError:
|
|
return False
|
|
|
|
return library.received_follows.filter(actor=actor, approved=True).exists()
|
|
|
|
|
|
class MusicLibraryViewSet(
|
|
FederationMixin, mixins.RetrieveModelMixin, viewsets.GenericViewSet
|
|
):
|
|
authentication_classes = [authentication.SignatureAuthentication]
|
|
renderer_classes = renderers.get_ap_renderers()
|
|
serializer_class = serializers.LibrarySerializer
|
|
queryset = (
|
|
music_models.Library.objects.all()
|
|
.local()
|
|
.select_related("actor")
|
|
.filter(channel=None)
|
|
)
|
|
lookup_field = "uuid"
|
|
|
|
def retrieve(self, request, *args, **kwargs):
|
|
lb = self.get_object()
|
|
if utils.should_redirect_ap_to_html(request.headers.get("accept")):
|
|
return redirect_to_html(lb.get_absolute_url())
|
|
conf = {
|
|
"id": lb.get_federation_id(),
|
|
"actor": lb.actor,
|
|
"name": lb.name,
|
|
"summary": lb.description,
|
|
"items": lb.uploads.for_federation()
|
|
.order_by("-creation_date")
|
|
.prefetch_related(
|
|
Prefetch(
|
|
"track",
|
|
queryset=music_models.Track.objects.select_related(
|
|
"album__artist__attributed_to",
|
|
"artist__attributed_to",
|
|
"artist__attachment_cover",
|
|
"attachment_cover",
|
|
"album__attributed_to",
|
|
"attributed_to",
|
|
"album__attachment_cover",
|
|
"album__artist__attachment_cover",
|
|
"description",
|
|
).prefetch_related(
|
|
"tagged_items__tag",
|
|
"album__tagged_items__tag",
|
|
"album__artist__tagged_items__tag",
|
|
"artist__tagged_items__tag",
|
|
"artist__description",
|
|
"album__description",
|
|
),
|
|
)
|
|
),
|
|
"item_serializer": serializers.UploadSerializer,
|
|
}
|
|
|
|
return get_collection_response(
|
|
conf=conf,
|
|
querystring=request.GET,
|
|
collection_serializer=serializers.LibrarySerializer(lb),
|
|
page_access_check=lambda: has_library_access(request, lb),
|
|
)
|
|
|
|
@action(methods=["get"], detail=True)
|
|
def followers(self, request, *args, **kwargs):
|
|
self.get_object()
|
|
# XXX Implement this
|
|
return response.Response({})
|
|
|
|
|
|
class MusicUploadViewSet(
|
|
FederationMixin, mixins.RetrieveModelMixin, viewsets.GenericViewSet
|
|
):
|
|
authentication_classes = [authentication.SignatureAuthentication]
|
|
renderer_classes = renderers.get_ap_renderers()
|
|
queryset = music_models.Upload.objects.local().select_related(
|
|
"library__actor",
|
|
"track__artist",
|
|
"track__album__artist",
|
|
"track__description",
|
|
"track__album__attachment_cover",
|
|
"track__album__artist__attachment_cover",
|
|
"track__artist__attachment_cover",
|
|
"track__attachment_cover",
|
|
)
|
|
serializer_class = serializers.UploadSerializer
|
|
lookup_field = "uuid"
|
|
|
|
def retrieve(self, request, *args, **kwargs):
|
|
instance = self.get_object()
|
|
if utils.should_redirect_ap_to_html(request.headers.get("accept")):
|
|
return redirect_to_html(instance.track.get_absolute_url())
|
|
|
|
serializer = self.get_serializer(instance)
|
|
return response.Response(serializer.data)
|
|
|
|
def get_queryset(self):
|
|
queryset = super().get_queryset()
|
|
actor = music_utils.get_actor_from_request(self.request)
|
|
return queryset.playable_by(actor)
|
|
|
|
def get_serializer(self, obj):
|
|
if obj.library.get_channel():
|
|
return serializers.ChannelUploadSerializer(obj)
|
|
return super().get_serializer(obj)
|
|
|
|
@action(
|
|
methods=["get"],
|
|
detail=True,
|
|
content_negotiation_class=renderers.IgnoreClientContentNegotiation,
|
|
)
|
|
def activity(self, request, *args, **kwargs):
|
|
object = self.get_object()
|
|
serializer = serializers.ChannelCreateUploadSerializer(object)
|
|
return response.Response(serializer.data)
|
|
|
|
|
|
class MusicArtistViewSet(
|
|
FederationMixin, mixins.RetrieveModelMixin, viewsets.GenericViewSet
|
|
):
|
|
authentication_classes = [authentication.SignatureAuthentication]
|
|
renderer_classes = renderers.get_ap_renderers()
|
|
queryset = music_models.Artist.objects.local().select_related(
|
|
"description", "attachment_cover"
|
|
)
|
|
serializer_class = serializers.ArtistSerializer
|
|
lookup_field = "uuid"
|
|
|
|
def retrieve(self, request, *args, **kwargs):
|
|
instance = self.get_object()
|
|
if utils.should_redirect_ap_to_html(request.headers.get("accept")):
|
|
return redirect_to_html(instance.get_absolute_url())
|
|
|
|
serializer = self.get_serializer(instance)
|
|
return response.Response(serializer.data)
|
|
|
|
|
|
class MusicAlbumViewSet(
|
|
FederationMixin, mixins.RetrieveModelMixin, viewsets.GenericViewSet
|
|
):
|
|
authentication_classes = [authentication.SignatureAuthentication]
|
|
renderer_classes = renderers.get_ap_renderers()
|
|
queryset = music_models.Album.objects.local().select_related(
|
|
"artist__description", "description", "artist__attachment_cover"
|
|
)
|
|
serializer_class = serializers.AlbumSerializer
|
|
lookup_field = "uuid"
|
|
|
|
def retrieve(self, request, *args, **kwargs):
|
|
instance = self.get_object()
|
|
if utils.should_redirect_ap_to_html(request.headers.get("accept")):
|
|
return redirect_to_html(instance.get_absolute_url())
|
|
|
|
serializer = self.get_serializer(instance)
|
|
return response.Response(serializer.data)
|
|
|
|
|
|
class MusicTrackViewSet(
|
|
FederationMixin, mixins.RetrieveModelMixin, viewsets.GenericViewSet
|
|
):
|
|
authentication_classes = [authentication.SignatureAuthentication]
|
|
renderer_classes = renderers.get_ap_renderers()
|
|
queryset = music_models.Track.objects.local().select_related(
|
|
"album__artist",
|
|
"album__description",
|
|
"artist__description",
|
|
"description",
|
|
"attachment_cover",
|
|
"album__artist__attachment_cover",
|
|
"album__attachment_cover",
|
|
"artist__attachment_cover",
|
|
)
|
|
serializer_class = serializers.TrackSerializer
|
|
lookup_field = "uuid"
|
|
|
|
def retrieve(self, request, *args, **kwargs):
|
|
instance = self.get_object()
|
|
if utils.should_redirect_ap_to_html(request.headers.get("accept")):
|
|
return redirect_to_html(instance.get_absolute_url())
|
|
|
|
serializer = self.get_serializer(instance)
|
|
return response.Response(serializer.data)
|
|
|
|
|
|
class ChannelViewSet(
|
|
FederationMixin, mixins.RetrieveModelMixin, viewsets.GenericViewSet
|
|
):
|
|
authentication_classes = [authentication.SignatureAuthentication]
|
|
renderer_classes = renderers.get_ap_renderers()
|
|
queryset = music_models.Artist.objects.local().select_related(
|
|
"description", "attachment_cover"
|
|
)
|
|
serializer_class = serializers.ArtistSerializer
|
|
lookup_field = "uuid"
|
|
|
|
def retrieve(self, request, *args, **kwargs):
|
|
instance = self.get_object()
|
|
if utils.should_redirect_ap_to_html(request.headers.get("accept")):
|
|
return redirect_to_html(instance.get_absolute_url())
|
|
|
|
serializer = self.get_serializer(instance)
|
|
return response.Response(serializer.data)
|
|
|
|
|
|
class IndexViewSet(FederationMixin, viewsets.GenericViewSet):
|
|
authentication_classes = [authentication.SignatureAuthentication]
|
|
renderer_classes = renderers.get_ap_renderers()
|
|
|
|
def dispatch(self, request, *args, **kwargs):
|
|
if not preferences.get("federation__public_index"):
|
|
return HttpResponse(status=405)
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
@action(
|
|
methods=["get"],
|
|
detail=False,
|
|
)
|
|
def libraries(self, request, *args, **kwargs):
|
|
libraries = (
|
|
music_models.Library.objects.local()
|
|
.filter(channel=None, privacy_level="everyone")
|
|
.prefetch_related("actor")
|
|
.order_by("creation_date")
|
|
)
|
|
conf = {
|
|
"id": federation_utils.full_url(
|
|
reverse("federation:index:index-libraries")
|
|
),
|
|
"items": libraries,
|
|
"item_serializer": serializers.LibrarySerializer,
|
|
"page_size": 100,
|
|
"actor": None,
|
|
}
|
|
return get_collection_response(
|
|
conf=conf,
|
|
querystring=request.GET,
|
|
collection_serializer=serializers.IndexSerializer(conf),
|
|
)
|
|
|
|
return response.Response({}, status=200)
|
|
|
|
@action(
|
|
methods=["get"],
|
|
detail=False,
|
|
)
|
|
def channels(self, request, *args, **kwargs):
|
|
actors = (
|
|
models.Actor.objects.local()
|
|
.exclude(channel=None)
|
|
.order_by("channel__creation_date")
|
|
.prefetch_related(
|
|
"channel__attributed_to",
|
|
"channel__artist",
|
|
"channel__artist__description",
|
|
"channel__artist__attachment_cover",
|
|
)
|
|
)
|
|
conf = {
|
|
"id": federation_utils.full_url(reverse("federation:index:index-channels")),
|
|
"items": actors,
|
|
"item_serializer": serializers.ActorSerializer,
|
|
"page_size": 100,
|
|
"actor": None,
|
|
}
|
|
return get_collection_response(
|
|
conf=conf,
|
|
querystring=request.GET,
|
|
collection_serializer=serializers.IndexSerializer(conf),
|
|
)
|
|
|
|
return response.Response({}, status=200)
|