Skip to content
Snippets Groups Projects
Commit 200670b7 authored by Georg Krause's avatar Georg Krause
Browse files

Refactor NodeInfo Endpoint to use proper serializer

parent a7b70126
Branches
No related tags found
No related merge requests found
from cache_memoize import cache_memoize
from django.urls import reverse
import funkwhale_api
from funkwhale_api.common import preferences
from funkwhale_api.federation import actors, models as federation_models
from funkwhale_api.federation import utils as federation_utils
from funkwhale_api.moderation import models as moderation_models
from funkwhale_api.music import utils as music_utils
from . import stats
def get():
all_preferences = preferences.all()
share_stats = all_preferences.get("instance__nodeinfo_stats_enabled")
allow_list_enabled = all_preferences.get("moderation__allow_list_enabled")
allow_list_public = all_preferences.get("moderation__allow_list_public")
auth_required = all_preferences.get("common__api_authentication_required")
banner = all_preferences.get("instance__banner")
unauthenticated_report_types = all_preferences.get(
"moderation__unauthenticated_report_types"
)
if allow_list_enabled and allow_list_public:
allowed_domains = list(
federation_models.Domain.objects.filter(allowed=True)
.order_by("name")
.values_list("name", flat=True)
)
else:
allowed_domains = None
data = {
"version": "2.0",
"software": {"name": "funkwhale", "version": funkwhale_api.__version__},
"protocols": ["activitypub"],
"services": {"inbound": [], "outbound": []},
"openRegistrations": all_preferences.get("users__registration_enabled"),
"usage": {"users": {"total": 0, "activeHalfyear": 0, "activeMonth": 0}},
"metadata": {
"actorId": actors.get_service_actor().fid,
"private": all_preferences.get("instance__nodeinfo_private"),
"shortDescription": all_preferences.get("instance__short_description"),
"longDescription": all_preferences.get("instance__long_description"),
"rules": all_preferences.get("instance__rules"),
"contactEmail": all_preferences.get("instance__contact_email"),
"terms": all_preferences.get("instance__terms"),
"nodeName": all_preferences.get("instance__name"),
"banner": federation_utils.full_url(banner.url) if banner else None,
"defaultUploadQuota": all_preferences.get("users__upload_quota"),
"library": {
"federationEnabled": all_preferences.get("federation__enabled"),
"anonymousCanListen": not all_preferences.get(
"common__api_authentication_required"
),
},
"supportedUploadExtensions": music_utils.SUPPORTED_EXTENSIONS,
"allowList": {"enabled": allow_list_enabled, "domains": allowed_domains},
"reportTypes": [
{"type": t, "label": l, "anonymous": t in unauthenticated_report_types}
for t, l in moderation_models.REPORT_TYPES
],
"funkwhaleSupportMessageEnabled": all_preferences.get(
"instance__funkwhale_support_message_enabled"
),
"instanceSupportMessage": all_preferences.get("instance__support_message"),
"endpoints": {"knownNodes": None, "channels": None, "libraries": None},
},
}
if share_stats:
getter = cache_memoize(600, prefix="memoize:instance:stats")(stats.get)
statistics = getter()
data["usage"]["users"]["total"] = statistics["users"]["total"]
data["usage"]["users"]["activeHalfyear"] = statistics["users"][
"active_halfyear"
]
data["usage"]["users"]["activeMonth"] = statistics["users"]["active_month"]
data["metadata"]["library"]["tracks"] = {"total": statistics["tracks"]}
data["metadata"]["library"]["artists"] = {"total": statistics["artists"]}
data["metadata"]["library"]["albums"] = {"total": statistics["albums"]}
data["metadata"]["library"]["music"] = {"hours": statistics["music_duration"]}
data["metadata"]["usage"] = {
"favorites": {"tracks": {"total": statistics["track_favorites"]}},
"listenings": {"total": statistics["listenings"]},
"downloads": {"total": statistics["downloads"]},
}
if not auth_required:
data["metadata"]["endpoints"]["knownNodes"] = federation_utils.full_url(
reverse("api:v1:federation:domains-list")
)
if not auth_required and preferences.get("federation__public_index"):
data["metadata"]["endpoints"]["libraries"] = federation_utils.full_url(
reverse("federation:index:index-libraries")
)
data["metadata"]["endpoints"]["channels"] = federation_utils.full_url(
reverse("federation:index:index-channels")
)
return data
from rest_framework import serializers
from funkwhale_api.federation.utils import full_url
from drf_spectacular.utils import extend_schema_field
class SoftwareSerializer(serializers.Serializer):
name = serializers.SerializerMethodField()
version = serializers.CharField()
def get_name(self, obj) -> str:
return "funkwhale"
class ServicesSerializer(serializers.Serializer):
inbound = serializers.ListField(child=serializers.CharField(), default=[])
outbound = serializers.ListField(child=serializers.CharField(), default=[])
class UsersUsageSerializer(serializers.Serializer):
total = serializers.IntegerField()
activeHalfyear = serializers.SerializerMethodField()
activeMonth = serializers.SerializerMethodField()
def get_activeHalfyear(self, obj) -> int:
return obj.get("active_halfyear", 0)
def get_activeMonth(self, obj) -> int:
return obj.get("active_month", 0)
class UsageSerializer(serializers.Serializer):
users = UsersUsageSerializer()
class TotalCountSerializer(serializers.Serializer):
total = serializers.SerializerMethodField()
def get_total(self, obj) -> int:
return obj
class TotalHoursSerializer(serializers.Serializer):
hours = serializers.SerializerMethodField()
def get_hours(self, obj) -> int:
return obj
class NodeInfoLibrarySerializer(serializers.Serializer):
federationEnabled = serializers.BooleanField()
anonymousCanListen = serializers.BooleanField()
tracks = TotalCountSerializer(default=0)
artists = TotalCountSerializer(default=0)
albums = TotalCountSerializer(default=0)
music = TotalHoursSerializer(source="music_duration", default=0)
class AllowListStatSerializer(serializers.Serializer):
enabled = serializers.BooleanField()
domains = serializers.ListField(child=serializers.CharField())
class ReportTypeSerializer(serializers.Serializer):
type = serializers.CharField()
label = serializers.CharField()
anonymous = serializers.BooleanField()
class EndpointsSerializer(serializers.Serializer):
knownNodes = serializers.URLField(default=None)
channels = serializers.URLField(default=None)
libraries = serializers.URLField(default=None)
class MetadataUsageFavoriteSerializer(serializers.Serializer):
tracks = serializers.SerializerMethodField()
@extend_schema_field(TotalCountSerializer)
def get_tracks(self, obj):
return TotalCountSerializer(obj).data
class MetadataUsageSerializer(serializers.Serializer):
favorites = MetadataUsageFavoriteSerializer(source="track_favorites")
listenings = TotalCountSerializer()
downloads = TotalCountSerializer()
class MetadataSerializer(serializers.Serializer):
actorId = serializers.CharField()
private = serializers.SerializerMethodField()
shortDescription = serializers.SerializerMethodField()
longDescription = serializers.SerializerMethodField()
rules = serializers.SerializerMethodField()
contactEmail = serializers.SerializerMethodField()
terms = serializers.SerializerMethodField()
nodeName = serializers.SerializerMethodField()
banner = serializers.SerializerMethodField()
defaultUploadQuota = serializers.SerializerMethodField()
library = serializers.SerializerMethodField()
supportedUploadExtensions = serializers.ListField(child=serializers.CharField())
allowList = serializers.SerializerMethodField()
reportTypes = ReportTypeSerializer(source="report_types", many=True)
funkwhaleSupportMessageEnabled = serializers.SerializerMethodField()
instanceSupportMessage = serializers.SerializerMethodField()
endpoints = EndpointsSerializer()
usage = serializers.SerializerMethodField(source="stats")
def get_private(self, obj) -> bool:
return obj["preferences"].get("instance__nodeinfo_private")
def get_shortDescription(self, obj) -> str:
return obj["preferences"].get("instance__short_description")
def get_longDescription(self, obj) -> str:
return obj["preferences"].get("instance__long_description")
def get_rules(self, obj) -> str:
return obj["preferences"].get("instance__rules")
def get_contactEmail(self, obj) -> str:
return obj["preferences"].get("instance__contact_email")
def get_terms(self, obj) -> str:
return obj["preferences"].get("instance__terms")
def get_nodeName(self, obj) -> str:
return obj["preferences"].get("instance__name")
@extend_schema_field(serializers.CharField)
def get_banner(self, obj) -> (str, None):
if obj["preferences"].get("instance__banner"):
return full_url(obj["preferences"].get("instance__banner").url)
return None
def get_defaultUploadQuota(self, obj) -> int:
return obj["preferences"].get("users__upload_quota")
def get_library(self, obj) -> bool:
data = obj["stats"] or {}
data["federationEnabled"] = obj["preferences"].get("federation__enabled")
data["anonymousCanListen"] = not obj["preferences"].get(
"common__api_authentication_required"
)
return NodeInfoLibrarySerializer(data).data
@extend_schema_field(AllowListStatSerializer)
def get_allowList(self, obj):
return AllowListStatSerializer(
{
"enabled": obj["preferences"].get("moderation__allow_list_enabled"),
"domains": obj["allowed_domains"] or None,
}
).data
def get_funkwhaleSupportMessageEnabled(self, obj) -> bool:
return obj["preferences"].get("instance__funkwhale_support_message_enabled")
def get_instanceSupportMessage(self, obj) -> str:
return obj["preferences"].get("instance__support_message")
@extend_schema_field(MetadataUsageSerializer)
def get_usage(self, obj):
return MetadataUsageSerializer(obj["stats"]).data
class NodeInfo20Serializer(serializers.Serializer):
version = serializers.SerializerMethodField()
software = SoftwareSerializer()
protocols = serializers.SerializerMethodField()
services = ServicesSerializer(default={})
openRegistrations = serializers.SerializerMethodField()
usage = serializers.SerializerMethodField()
metadata = serializers.SerializerMethodField()
def get_version(self, obj) -> str:
return "2.0"
def get_protocols(self, obj) -> list:
return ["activitypub"]
def get_services(self, obj) -> object:
return {"inbound": [], "outbound": []}
def get_openRegistrations(self, obj) -> bool:
return obj["preferences"]["users__registration_enabled"]
@extend_schema_field(UsageSerializer)
def get_usage(self, obj):
usage = None
if obj["preferences"]["instance__nodeinfo_stats_enabled"]:
usage = obj["stats"]
else:
usage = {"users": {"total": 0, "activeMonth": 0, "activeHalfyear": 0}}
return UsageSerializer(usage).data
@extend_schema_field(MetadataSerializer)
def get_metadata(self, obj):
return MetadataSerializer(obj).data
import json
import logging
from cache_memoize import cache_memoize
from django.conf import settings
from django.urls import reverse
from dynamic_preferences.api import serializers
from dynamic_preferences.api.serializers import GlobalPreferenceSerializer
from dynamic_preferences.api import viewsets as preferences_viewsets
from dynamic_preferences.registries import global_preferences_registry
from rest_framework import views
from rest_framework import generics
from rest_framework import views
from rest_framework.response import Response
from funkwhale_api import __version__ as funkwhale_version
from funkwhale_api.common import middleware
from funkwhale_api.common import preferences
from funkwhale_api.federation import utils as federation_utils
from funkwhale_api.federation.models import Domain
from funkwhale_api.federation.actors import get_service_actor
from funkwhale_api.users.oauth import permissions as oauth_permissions
from funkwhale_api.music.utils import SUPPORTED_EXTENSIONS
from funkwhale_api.moderation.models import REPORT_TYPES
from drf_spectacular.utils import extend_schema
from . import nodeinfo
from . import serializers
from . import stats
NODEINFO_2_CONTENT_TYPE = "application/json; profile=http://nodeinfo.diaspora.software/ns/schema/2.0#; charset=utf-8" # noqa
......@@ -32,7 +42,7 @@ class AdminSettings(preferences_viewsets.GlobalPreferencesViewSet):
class InstanceSettings(generics.GenericAPIView):
permission_classes = []
authentication_classes = []
serializer_class = serializers.GlobalPreferenceSerializer
serializer_class = GlobalPreferenceSerializer
def get_queryset(self):
manager = global_preferences_registry.manager()
......@@ -45,21 +55,66 @@ class InstanceSettings(generics.GenericAPIView):
def get(self, request):
queryset = self.get_queryset()
serializer = serializers.GlobalPreferenceSerializer(queryset, many=True)
return Response(serializer.data)
data = GlobalPreferenceSerializer(queryset, many=True).data
return Response(data, status=200)
class NodeInfo(views.APIView):
permission_classes = []
authentication_classes = []
def get(self, request, *args, **kwargs):
try:
data = nodeinfo.get()
except ValueError:
logger.warn("nodeinfo returned invalid json")
data = {}
return Response(data, status=200, content_type=NODEINFO_2_CONTENT_TYPE)
@extend_schema(responses=serializers.NodeInfo20Serializer)
def get(self, request):
pref = preferences.all()
if (
pref["moderation__allow_list_public"]
and pref["moderation__allow_list_enabled"]
):
allowed_domains = list(
Domain.objects.filter(allowed=True)
.order_by("name")
.values_list("name", flat=True)
)
else:
allowed_domains = None
data = {
"software": {"version": funkwhale_version},
"preferences": pref,
"stats": cache_memoize(600, prefix="memoize:instance:stats")(stats.get)()
if pref["instance__nodeinfo_stats_enabled"]
else None,
"actorId": get_service_actor().fid,
"supportedUploadExtensions": SUPPORTED_EXTENSIONS,
"allowed_domains": allowed_domains,
"report_types": [
{
"type": t,
"label": l,
"anonymous": t
in pref.get("moderation__unauthenticated_report_types"),
}
for t, l in REPORT_TYPES
],
"endpoints": {},
}
if not pref.get("common__api_authentication_required"):
if pref.get("instance__nodeinfo_stats_enabled"):
data["endpoints"]["knownNodes"] = reverse(
"api:v1:federation:domains-list"
)
if pref.get("federation__public_index"):
data["endpoints"]["libraries"] = reverse(
"federation:index:index-libraries"
)
data["endpoints"]["channels"] = reverse(
"federation:index:index-channels"
)
serializer = serializers.NodeInfo20Serializer(data)
return Response(
serializer.data, status=200, content_type=NODEINFO_2_CONTENT_TYPE
)
class SpaManifest(views.APIView):
......
import pytest
from django.urls import reverse
import funkwhale_api
from funkwhale_api.instance import nodeinfo
from funkwhale_api.federation import actors
from funkwhale_api.federation import utils as federation_utils
from funkwhale_api.music import utils as music_utils
from collections import OrderedDict
def test_nodeinfo_dump(preferences, mocker, avatar):
preferences["instance__banner"] = avatar
preferences["instance__nodeinfo_stats_enabled"] = True
preferences["common__api_authentication_required"] = False
preferences["moderation__unauthenticated_report_types"] = [
"takedown_request",
"other",
"other_category_that_doesnt_exist",
]
stats = {
"users": {"total": 1, "active_halfyear": 12, "active_month": 13},
"tracks": 2,
"albums": 3,
"artists": 4,
"track_favorites": 5,
"music_duration": 6,
"listenings": 7,
"downloads": 42,
}
mocker.patch("funkwhale_api.instance.stats.get", return_value=stats)
def test_nodeinfo_default(api_client):
url = reverse("api:v1:instance:nodeinfo-2.0")
response = api_client.get(url)
expected = {
"version": "2.0",
"software": {"name": "funkwhale", "version": funkwhale_api.__version__},
"software": OrderedDict([("name", "funkwhale"), ("version", "1.2.7")]),
"protocols": ["activitypub"],
"services": {"inbound": [], "outbound": []},
"openRegistrations": preferences["users__registration_enabled"],
"usage": {"users": {"total": 1, "activeHalfyear": 12, "activeMonth": 13}},
"services": OrderedDict([("inbound", []), ("outbound", [])]),
"openRegistrations": False,
"usage": {
"users": OrderedDict(
[("total", 0), ("activeHalfyear", 0), ("activeMonth", 0)]
)
},
"metadata": {
"actorId": actors.get_service_actor().fid,
"private": preferences["instance__nodeinfo_private"],
"shortDescription": preferences["instance__short_description"],
"longDescription": preferences["instance__long_description"],
"nodeName": preferences["instance__name"],
"rules": preferences["instance__rules"],
"contactEmail": preferences["instance__contact_email"],
"defaultUploadQuota": preferences["users__upload_quota"],
"terms": preferences["instance__terms"],
"banner": federation_utils.full_url(preferences["instance__banner"].url),
"actorId": "https://test.federation/federation/actors/service",
"private": False,
"shortDescription": "",
"longDescription": "",
"rules": "",
"contactEmail": "",
"terms": "",
"nodeName": "",
"banner": None,
"defaultUploadQuota": 1000,
"library": {
"federationEnabled": preferences["federation__enabled"],
"anonymousCanListen": not preferences[
"common__api_authentication_required"
],
"tracks": {"total": stats["tracks"]},
"artists": {"total": stats["artists"]},
"albums": {"total": stats["albums"]},
"music": {"hours": stats["music_duration"]},
},
"usage": {
"favorites": {"tracks": {"total": stats["track_favorites"]}},
"listenings": {"total": stats["listenings"]},
"downloads": {"total": stats["downloads"]},
"federationEnabled": True,
"anonymousCanListen": False,
"tracks": OrderedDict([("total", 0)]),
"artists": OrderedDict([("total", 0)]),
"albums": OrderedDict([("total", 0)]),
"music": OrderedDict([("hours", 0)]),
},
"supportedUploadExtensions": music_utils.SUPPORTED_EXTENSIONS,
"supportedUploadExtensions": [
"aac",
"aif",
"aiff",
"flac",
"m4a",
"mp3",
"ogg",
"opus",
],
"allowList": {"enabled": False, "domains": None},
"reportTypes": [
{
"type": "takedown_request",
"label": "Takedown request",
"anonymous": True,
},
{
"type": "invalid_metadata",
"label": "Invalid metadata",
"anonymous": False,
},
{
"type": "illegal_content",
"label": "Illegal content",
"anonymous": False,
},
{
"type": "offensive_content",
"label": "Offensive content",
"anonymous": False,
},
{"type": "other", "label": "Other", "anonymous": True},
],
"funkwhaleSupportMessageEnabled": preferences[
"instance__funkwhale_support_message_enabled"
],
"instanceSupportMessage": preferences["instance__support_message"],
"endpoints": {
"knownNodes": federation_utils.full_url(
reverse("api:v1:federation:domains-list")
OrderedDict(
[
("type", "takedown_request"),
("label", "Takedown request"),
("anonymous", True),
]
),
"libraries": federation_utils.full_url(
reverse("federation:index:index-libraries")
OrderedDict(
[
("type", "invalid_metadata"),
("label", "Invalid metadata"),
("anonymous", False),
]
),
"channels": federation_utils.full_url(
reverse("federation:index:index-channels")
OrderedDict(
[
("type", "illegal_content"),
("label", "Illegal content"),
("anonymous", True),
]
),
},
},
}
assert nodeinfo.get() == expected
def test_nodeinfo_dump_stats_disabled(preferences, mocker):
preferences["instance__nodeinfo_stats_enabled"] = False
preferences["federation__public_index"] = False
preferences["moderation__unauthenticated_report_types"] = [
"takedown_request",
"other",
OrderedDict(
[
("type", "offensive_content"),
("label", "Offensive content"),
("anonymous", False),
]
expected = {
"version": "2.0",
"software": {"name": "funkwhale", "version": funkwhale_api.__version__},
"protocols": ["activitypub"],
"services": {"inbound": [], "outbound": []},
"openRegistrations": preferences["users__registration_enabled"],
"usage": {"users": {"total": 0, "activeHalfyear": 0, "activeMonth": 0}},
"metadata": {
"actorId": actors.get_service_actor().fid,
"private": preferences["instance__nodeinfo_private"],
"shortDescription": preferences["instance__short_description"],
"longDescription": preferences["instance__long_description"],
"nodeName": preferences["instance__name"],
"rules": preferences["instance__rules"],
"contactEmail": preferences["instance__contact_email"],
"defaultUploadQuota": preferences["users__upload_quota"],
"terms": preferences["instance__terms"],
"banner": None,
"library": {
"federationEnabled": preferences["federation__enabled"],
"anonymousCanListen": not preferences[
"common__api_authentication_required"
),
OrderedDict(
[("type", "other"), ("label", "Other"), ("anonymous", False)]
),
],
"funkwhaleSupportMessageEnabled": True,
"instanceSupportMessage": "",
"endpoints": OrderedDict(
[("knownNodes", None), ("channels", None), ("libraries", None)]
),
"usage": {
"favorites": OrderedDict([("tracks", {"total": 0})]),
"listenings": OrderedDict([("total", 0)]),
"downloads": OrderedDict([("total", 0)]),
},
"supportedUploadExtensions": music_utils.SUPPORTED_EXTENSIONS,
"allowList": {"enabled": False, "domains": None},
"reportTypes": [
{
"type": "takedown_request",
"label": "Takedown request",
"anonymous": True,
},
{
"type": "invalid_metadata",
"label": "Invalid metadata",
"anonymous": False,
},
{
"type": "illegal_content",
"label": "Illegal content",
"anonymous": False,
},
{
"type": "offensive_content",
"label": "Offensive content",
"anonymous": False,
},
{"type": "other", "label": "Other", "anonymous": True},
],
"funkwhaleSupportMessageEnabled": preferences[
"instance__funkwhale_support_message_enabled"
],
"instanceSupportMessage": preferences["instance__support_message"],
"endpoints": {"knownNodes": None, "libraries": None, "channels": None},
},
}
assert nodeinfo.get() == expected
@pytest.mark.parametrize(
"enabled, public, expected",
[
(True, True, {"enabled": True, "domains": ["allowed.example"]}),
(True, False, {"enabled": True, "domains": None}),
(False, False, {"enabled": False, "domains": None}),
],
)
def test_nodeinfo_allow_list_enabled(preferences, factories, enabled, public, expected):
preferences["moderation__allow_list_enabled"] = enabled
preferences["moderation__allow_list_public"] = public
factories["federation.Domain"](name="allowed.example", allowed=True)
factories["federation.Domain"](allowed=False)
factories["federation.Domain"](allowed=None)
assert nodeinfo.get()["metadata"]["allowList"] == expected
assert response.data == expected
......@@ -5,15 +5,12 @@ from django.urls import reverse
from funkwhale_api.federation import utils as federation_utils
def test_nodeinfo_endpoint(db, api_client, mocker):
payload = {"test": "test"}
mocker.patch("funkwhale_api.instance.nodeinfo.get", return_value=payload)
def test_nodeinfo_endpoint(db, api_client):
url = reverse("api:v1:instance:nodeinfo-2.0")
response = api_client.get(url)
ct = "application/json; profile=http://nodeinfo.diaspora.software/ns/schema/2.0#; charset=utf-8" # noqa
assert response.status_code == 200
assert response["Content-Type"] == ct
assert response.data == payload
def test_settings_only_list_public_settings(db, api_client, preferences):
......
Refactor node info endpoint to use proper serializers
\ No newline at end of file
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment