import json
import os
import warnings
from base64 import b64encode
from string import Template
from typing import Dict
from jsonschema import validate
from jsonschema.exceptions import ValidationError
from xrd import XRD, Link, Element
[docs]
def generate_legacy_webfinger(template=None, *args, **kwargs):
"""Generate a legacy webfinger XRD document.
Template specific key-value pairs need to be passed as ``kwargs``, see classes.
:arg template: Ready template to fill with args, for example "diaspora" (optional)
:returns: Rendered XRD document (str)
"""
if template == "diaspora":
webfinger = DiasporaWebFinger(*args, **kwargs)
else:
webfinger = BaseLegacyWebFinger(*args, **kwargs)
return webfinger.render()
[docs]
def generate_nodeinfo2_document(**kwargs):
"""
Generate a NodeInfo2 document.
Pass in a dictionary as per NodeInfo2 1.0 schema:
https://github.com/jaywink/nodeinfo2/blob/master/schemas/1.0/schema.json
Minimum required schema:
{server:
baseUrl
name
software
version
}
openRegistrations
Protocols default will match what this library supports, ie "diaspora" currently.
:return: dict
:raises: KeyError on missing required items
"""
return {
"version": "1.0",
"server": {
"baseUrl": kwargs['server']['baseUrl'],
"name": kwargs['server']['name'],
"software": kwargs['server']['software'],
"version": kwargs['server']['version'],
},
"organization": {
"name": kwargs.get('organization', {}).get('name', None),
"contact": kwargs.get('organization', {}).get('contact', None),
"account": kwargs.get('organization', {}).get('account', None),
},
"protocols": kwargs.get('protocols', ["diaspora"]),
"relay": kwargs.get('relay', ''),
"services": {
"inbound": kwargs.get('service', {}).get('inbound', []),
"outbound": kwargs.get('service', {}).get('outbound', []),
},
"openRegistrations": kwargs['openRegistrations'],
"usage": {
"users": {
"total": kwargs.get('usage', {}).get('users', {}).get('total'),
"activeHalfyear": kwargs.get('usage', {}).get('users', {}).get('activeHalfyear'),
"activeMonth": kwargs.get('usage', {}).get('users', {}).get('activeMonth'),
"activeWeek": kwargs.get('usage', {}).get('users', {}).get('activeWeek'),
},
"localPosts": kwargs.get('usage', {}).get('localPosts'),
"localComments": kwargs.get('usage', {}).get('localComments'),
}
}
[docs]
def generate_hcard(template=None, **kwargs):
"""Generate a hCard document.
Template specific key-value pairs need to be passed as ``kwargs``, see classes.
:arg template: Ready template to fill with args, for example "diaspora" (optional)
:returns: HTML document (str)
"""
if template == "diaspora":
hcard = DiasporaHCard(**kwargs)
else:
raise NotImplementedError()
return hcard.render()
class BaseHostMeta:
def __init__(self, *args, **kwargs):
self.xrd = XRD()
def render(self):
return self.xrd.to_xml().toprettyxml(indent=" ", encoding="UTF-8")
class BaseLegacyWebFinger(BaseHostMeta):
"""Legacy XRD WebFinger.
See: https://code.google.com/p/webfinger/wiki/WebFingerProtocol
"""
def __init__(self, address, *args, **kwargs):
super().__init__(*args, **kwargs)
subject = Element("Subject", "acct:%s" % address)
self.xrd.elements.append(subject)
[docs]
class DiasporaWebFinger(BaseLegacyWebFinger):
"""Diaspora version of legacy WebFinger.
Required keyword args:
* handle (str) - eg user@domain.tld
* host (str) - eg https://domain.tld
* guid (str) - guid of user
* public_key (str) - public key
"""
def __init__(self, handle, host, guid, public_key, *args, **kwargs):
super().__init__(handle, *args, **kwargs)
self.xrd.elements.append(Element("Alias", "%s/people/%s" % (
host, guid
)))
username = handle.split("@")[0]
self.xrd.links.append(Link(
rel="http://microformats.org/profile/hcard",
type_="text/html",
href="%s/hcard/users/%s" %(
host, guid
)
))
self.xrd.links.append(Link(
rel="http://joindiaspora.com/seed_location",
type_="text/html",
href=host
))
self.xrd.links.append(Link(
rel="http://joindiaspora.com/guid",
type_="text/html",
href=guid
))
self.xrd.links.append(Link(
rel="http://webfinger.net/rel/profile-page",
type_="text/html",
href="%s/u/%s" % (
host, username
)
))
self.xrd.links.append(Link(
rel="http://schemas.google.com/g/2010#updates-from",
type_="application/atom+xml",
href="%s/public/%s.atom" % (
host, username
)
))
# Base64 the key
# See https://wiki.diasporafoundation.org/Federation_Protocol_Overview#Diaspora_Public_Key
try:
base64_key = b64encode(bytes(public_key, encoding="UTF-8")).decode("ascii")
except TypeError:
# Python 2
base64_key = b64encode(public_key).decode("ascii")
self.xrd.links.append(Link(
rel="diaspora-public-key",
type_="RSA",
href=base64_key
))
[docs]
class DiasporaHCard:
"""Diaspora hCard document.
Must receive the `required` attributes as keyword arguments to init.
"""
required = [
"hostname", "fullname", "firstname", "lastname", "photo300", "photo100", "photo50", "searchable", "guid", "public_key", "username",
]
def __init__(self, **kwargs):
self.kwargs = kwargs
template_path = os.path.join(os.path.dirname(__file__), "templates", "hcard_diaspora.html")
with open(template_path) as f:
self.template = Template(f.read())
def render(self):
required = self.required[:]
for key, value in self.kwargs.items():
required.remove(key)
assert value is not None
assert isinstance(value, str)
assert len(required) == 0
return self.template.substitute(self.kwargs)
[docs]
class SocialRelayWellKnown:
"""A `.well-known/social-relay` document in JSON.
For apps wanting to announce their preferences towards relay applications.
See WIP spec: https://wiki.diasporafoundation.org/Relay_servers_for_public_posts
Schema see `schemas/social-relay-well-known.json`
:arg subscribe: bool
:arg tags: tuple, optional
:arg scope: Should be either "all" or "tags", default is "all" if not given
"""
def __init__(self, subscribe, tags=(), scope="all", *args, **kwargs):
self.doc = {
"subscribe": subscribe,
"scope": scope,
"tags": list(tags),
}
def render(self):
self.validate_doc()
return json.dumps(self.doc)
def validate_doc(self):
schema_path = os.path.join(os.path.dirname(__file__), "schemas", "social-relay-well-known.json")
with open(schema_path) as f:
schema = json.load(f)
validate(self.doc, schema)
[docs]
class NodeInfo:
"""Generate a NodeInfo document.
See spec: http://nodeinfo.diaspora.software
NodeInfo is unnecessarely restrictive in field values. We wont be supporting such strictness, though
we will raise a warning unless validation is skipped with `skip_validate=True`.
For strictness, `raise_on_validate=True` will cause a `ValidationError` to be raised.
See schema document `federation/hostmeta/schemas/nodeinfo-1.0.json` for how to instantiate this class.
"""
def __init__(self, software, protocols, services, open_registrations, usage, metadata, skip_validate=False,
raise_on_validate=False):
self.doc = {
"version": "1.0",
"software": software,
"protocols": protocols,
"services": services,
"openRegistrations": open_registrations,
"usage": usage,
"metadata": metadata,
}
self.skip_validate = skip_validate
self.raise_on_validate = raise_on_validate
def render(self):
if not self.skip_validate:
self.validate_doc()
return json.dumps(self.doc)
def validate_doc(self):
schema_path = os.path.join(os.path.dirname(__file__), "schemas", "nodeinfo-1.0.json")
with open(schema_path) as f:
schema = json.load(f)
try:
validate(self.doc, schema)
except ValidationError:
if self.raise_on_validate:
raise
warnings.warn("NodeInfo document generated does not validate against NodeInfo 1.0 specification.")
# The default NodeInfo document path
NODEINFO_DOCUMENT_PATH = "/nodeinfo/1.0"
[docs]
def get_nodeinfo_well_known_document(url, document_path=None):
"""Generate a NodeInfo .well-known document.
See spec: http://nodeinfo.diaspora.software
:arg url: The full base url with protocol, ie https://example.com
:arg document_path: Custom NodeInfo document path if supplied (optional)
:returns: dict
"""
return {
"links": [
{
"rel": "http://nodeinfo.diaspora.software/ns/schema/1.0",
"href": "{url}{path}".format(
url=url, path=document_path or NODEINFO_DOCUMENT_PATH
)
}
]
}
[docs]
class MatrixClientWellKnown:
"""
Matrix Client well-known as per https://matrix.org/docs/spec/client_server/r0.6.1#server-discovery
"""
def __init__(self, homeserver_base_url: str, identity_server_base_url: str = None, other_keys: Dict = None):
self.homeserver_base_url = homeserver_base_url
self.identity_server_base_url = identity_server_base_url
self.other_keys = other_keys
def render(self):
doc = {
"m.homeserver": {
"base_url": self.homeserver_base_url,
}
}
if self.identity_server_base_url:
doc["m.identity_server"] = {
"base_url": self.identity_server_base_url,
}
if self.other_keys:
doc.update(self.other_keys)
return doc
[docs]
class MatrixServerWellKnown:
"""
Matrix Server well-known as per https://matrix.org/docs/spec/server_server/r0.1.4#server-discovery
"""
def __init__(self, homeserver_domain_with_port: str):
self.homeserver_domain_with_port = homeserver_domain_with_port
def render(self):
return {
"m.server": self.homeserver_domain_with_port,
}
[docs]
class RFC7033Webfinger:
"""
RFC 7033 webfinger - see https://tools.ietf.org/html/rfc7033
A Django view is also available, see the child ``django`` module for view and url configuration.
:param id: Profile ActivityPub ID in URL format
:param handle: Profile Diaspora handle
:param guid: Profile Diaspora guid
:param base_url: The base URL of the server (protocol://domain.tld)
:param profile_path: Profile path for the user (for example `/profile/johndoe/`)
:param hcard_path: (Optional) hCard path, defaults to ``/hcard/users/``.
:param atom_path: (Optional) atom feed path
:returns: dict
"""
def __init__(
self, id: str, handle: str, guid: str, base_url: str, profile_path: str, hcard_path: str="/hcard/users/",
atom_path: str=None, search_path: str=None,
):
self.id = id
self.handle = handle
self.guid = guid
self.base_url = base_url
self.hcard_path = hcard_path
self.profile_path = profile_path
self.atom_path = atom_path
self.search_path = search_path
def render(self):
webfinger = {
"subject": "acct:%s" % self.handle,
"aliases": [
f"{self.base_url}{self.profile_path}",
self.id,
],
"links": [
{
"rel": "http://microformats.org/profile/hcard",
"type": "text/html",
"href": "%s%s%s" % (self.base_url, self.hcard_path, self.guid),
},
{
"rel": "http://joindiaspora.com/seed_location",
"type": "text/html",
"href": self.base_url,
},
{
"rel": "http://webfinger.net/rel/profile-page",
"type": "text/html",
"href": "%s%s" % (self.base_url, self.profile_path),
},
{
"rel": "salmon",
"href": "%s/receive/users/%s" % (self.base_url, self.guid),
},
],
}
webfinger["links"].append({
"rel": "self",
"href": self.id,
"type": "application/activity+json",
})
if self.atom_path:
webfinger['links'].append(
{
"rel": "http://schemas.google.com/g/2010#updates-from",
"type": "application/atom+xml",
"href": "%s%s" % (self.base_url, self.atom_path),
}
)
if self.search_path:
webfinger['links'].append(
{
"rel": "http://ostatus.org/schema/1.0/subscribe",
"template": "%s%s{uri}" % (self.base_url, self.search_path),
},
)
return webfinger