Webfinger, plus tests

2019-08-17
Marnanel Thurman 2019-08-09 17:06:44 +01:00
rodzic 811e2e1fd9
commit a2f5e03ab0
4 zmienionych plików z 205 dodań i 0 usunięć

Wyświetl plik

@ -17,5 +17,6 @@ urlpatterns = [
# at the root.
path('.well-known/host-meta', django_kepi.views.HostMeta.as_view()),
path('.well-known/webfinger', django_kepi.views.Webfinger.as_view()),
]

Wyświetl plik

@ -5,6 +5,7 @@ from .activitypub import KepiView, ThingView, ActorView, \
InboxView, OutboxView
from .host_meta import HostMeta
from .webfinger import Webfinger
__all__ = [
'KepiView', 'ThingView', 'ActorView',
@ -14,4 +15,5 @@ __all__ = [
'InboxView', 'OutboxView',
'HostMeta',
'Webfinger',
]

Wyświetl plik

@ -0,0 +1,92 @@
import django.views
from django.conf import settings
from django.shortcuts import render
from django_kepi.models.actor import Actor
from django.http import HttpResponse
import logging
import re
import json
logger = logging.Logger('django_kepi')
class Webfinger(django.views.View):
"""
RFC7033 webfinger support.
"""
def _get_body(self, request):
try:
user = request.GET['resource']
except:
return HttpResponse(
status = 400,
reason = 'no resource for webfinger',
content = 'no resource for webfinger',
content_type = 'text/plain',
)
# Generally, user resources should be prefaced with "acct:",
# per RFC7565. We support this, but we don't enforce it.
user = re.sub(r'^acct:', '', user)
if '@' not in user:
return HttpResponse(
status = 404,
reason = 'absolute name required',
content = 'Please use the absolute form of the username.',
content_type = 'text/plain',
)
username, hostname = user.split('@', 2)
if hostname not in settings.ALLOWED_HOSTS:
return HttpResponse(
status = 404,
reason = 'not this server',
content = 'That user lives on another server.', content_type = 'text/plain',
)
try:
whoever = Actor.objects.get(
remote_url = None,
f_preferredUsername = username,
)
except Actor.DoesNotExist:
return HttpResponse(
status = 404,
reason = 'no such user',
content = 'We don\'t have a user with that name.',
content_type = 'text/plain',
)
actor_url = settings.KEPI['USER_URL_FORMAT'] % (username,)
result = {
"subject" : "acct:{}@{}".format(username, hostname),
"aliases" : [
actor_url,
],
"links":[
{
"rel" : "self",
"type" : "application/activity+json",
"href" : actor_url,
},
]}
return HttpResponse(
status = 200,
reason = 'Here you go',
content = bytes(json.dumps(result, indent=2),
encoding='utf-8'),
content_type='application/jrd+json; charset=utf-8')
def get(self, request):
result = self._get_body(request)
result['Access-Control-Allow-Origin'] = '*'
return result

Wyświetl plik

@ -0,0 +1,110 @@
from django.conf import settings
from django.test import TestCase, Client
from . import create_local_person
import logging
import json
WEBFINGER_BASE_URL = 'https://altair.example.com/.well-known/webfinger'
WEBFINGER_URL = WEBFINGER_BASE_URL + '?resource={}'
WEBFINGER_MIME_TYPE = 'application/jrd+json; charset=utf-8'
logger = logging.getLogger(name='django_kepi')
class TestWebfinger(TestCase):
def setUp(self):
keys = json.load(open('tests/keys/keys-0001.json', 'r'))
create_local_person(
name='alice',
publicKey=keys['public'],
privateKey=keys['private'],
)
self._alice_keys = keys
settings.ALLOWED_HOSTS = [
'altair.example.com',
'testserver',
]
def _fetch(self, url):
client = Client()
response = client.get(
url,
HTTP_ACCEPT = WEBFINGER_MIME_TYPE,
)
return response
def test_no_resource(self):
response = self._fetch(
url=WEBFINGER_BASE_URL,
)
self.assertEqual(response.status_code, 400)
def test_malformed(self):
response = self._fetch(
url=WEBFINGER_URL.format(
'I like coffee',
),
)
self.assertEqual(response.status_code, 404)
def test_wrong_server(self):
response = self._fetch(
url=WEBFINGER_URL.format(
'jamie@magic-torch.example.net',
),
)
self.assertEqual(response.status_code, 404)
def test_unknown_user(self):
response = self._fetch(
url=WEBFINGER_URL.format(
'lord_lucan@altair.example.com',
),
)
self.assertEqual(response.status_code, 404)
def test_working(self):
response = self._fetch(
url=WEBFINGER_URL.format(
'alice@altair.example.com',
),
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response['Content-Type'],
WEBFINGER_MIME_TYPE)
# per RFC:
self.assertEqual(response['Access-Control-Allow-Origin'],
'*')
parsed = json.loads(response.content)
self.assertEqual(parsed['subject'],
'acct:alice@altair.example.com',
)
self.assertIn(
'https://altair.example.com/users/alice',
parsed['aliases'],
)
self.assertIn(
{
'rel': 'self',
'type': 'application/activity+json',
'href': 'https://altair.example.com/users/alice',
},
parsed['links'],
)