izzylib-http-async/barkshark_http_async/activitypub.py
2022-02-18 16:45:32 -05:00

772 lines
17 KiB
Python

import json, mimetypes, traceback
from datetime import datetime, timezone
from functools import partial
from typing import Union
from xml.etree.ElementTree import fromstring
from .dotdict import DotDict
from .misc import DateString, Url, boolean
pubstr = 'https://www.w3.org/ns/activitystreams#Public'
actor_types = ['Application', 'Group', 'Organization', 'Person', 'Service']
activity_types = [
'Accept', 'Add', 'Announce', 'Arrive', 'Block', 'Create', 'Delete', 'Dislike',
'Flag', 'Follow', 'Ignore', 'Invite', 'Join', 'Leave', 'Like', 'Listen',
'Move', 'Offer', 'Question', 'Reject', 'Read', 'Remove', 'TentativeAccept',
'TentativeReject', 'Travel', 'Undo', 'Update', 'View'
]
link_types = ['Mention']
object_types = [
'Article', 'Audio', 'Document', 'Event', 'Image', 'Note', 'Page', 'Place',
'Profile', 'Relationship', 'Tombstone', 'Video'
]
url_keys = [
'attributedTo', 'url', 'href', 'object', 'id', 'actor', 'partOf', 'target'
]
def parse_privacy_level(to: list=[], cc: list=[], followers=None):
if pubstr in to and followers in cc:
return 'public'
elif followers in to and pubstr in cc:
return 'unlisted'
elif pubstr not in to and pubstr not in cc and followers in cc:
return 'private'
elif not tuple(item for item in [*to, *cc] if item not in [pubstr, followers]):
return 'direct'
else:
logging.warning('Not sure what this privacy level is')
logging.debug(f'to: {json.dumps(to)}')
logging.debug(f'cc: {json.dumps(cc)}')
logging.debug(f'followers: {followers}')
def generate_privacy_fields(privacy='public', followers=None, to=[], cc=[]):
if privacy == 'public':
to = [pubstr, *to]
cc = [followers, *to]
elif privacy == 'unlisted':
to = [followers, *to]
cc = [pubstr, *to]
elif privacy == 'private':
cc = [followers, *cc]
elif privacy == 'direct':
pass
else:
raise ValueError(f'Unknown privacy level: {privacy}')
return to, cc
class Object(DotDict):
def __setitem__(self, key, value):
if type(key) == str and key in url_keys:
value = Url(value)
elif key == 'object' and isinstance(key, dict):
value = Object(value)
super().__setitem__(key, value)
@classmethod
def new_activity(cls, id: str, type: str, actor_src: Union[str, dict], object: Union[str, dict], to: list=[pubstr], cc: list=[]):
assert type in activity_types
activity = cls({
'@context': 'https://www.w3.org/ns/activitystreams',
'id': id,
'object': object,
'type': type,
'actor': actor_src
})
if to:
activity.to = to
if cc:
activity.cc = cc
return activity
@classmethod
def new_note(cls, id, url, actor, content, **kwargs):
assert False not in map(isinstance, [id, actor, url], [Url])
if kwargs.get('date'):
date = DateString.from_datetime(kwargs['date'], 'activitypub')
else:
date = DateString.now('activitypub')
return cls({
"@context": [
"https://www.w3.org/ns/activitystreams",
{
"sensitive": "as:sensitive",
"toot": "http://joinmastodon.org/ns#",
#"votersCount": "toot:votersCount",
#"litepub": "http://litepub.social/ns#",
#"directMessage": "litepub:directMessage"
}
],
"id": id,
"type": "Note",
"summary": kwargs.get('summary'),
#"inReplyTo": kwargs.get('replyto'),
"published": date,
"url": url,
"attributedTo": actor,
"to": [
"https://www.w3.org/ns/activitystreams#Public"
],
"cc": [
f'{actor}/followers'
],
"sensitive": kwargs.get('sensitive', False),
"content": f'{content}',
#"contentMap": {
#"en": content
#},
#"attachment": [],
#"tag": [],
#"replies": {
#"id": f"{id}/replies",
#"type": "Collection",
#"first": {
#"type": "CollectionPage",
#"next": f"{id}/replies?only_other_accounts=true&page=true",
#"partOf": f"{id}/replies",
#"items": []
#}
#}
})
@classmethod
def new_actor(cls, actor, handle, pubkey, published=None, table={}, full=True, **kwargs):
actor_type = kwargs.get('type', 'Person').title()
assert actor_type in actor_types
actor = Url(actor)
data = cls({
'@context': [
'https://www.w3.org/ns/activitystreams',
'https://w3id.org/security/v1',
{
'schema': 'http://schema.org',
'toot': 'https://joinmastodon.org/ns#',
#'Device': 'toot:Device',
#'Ed25519Signature': 'toot:Ed25519Signature',
#'Ed25519Key': 'toot:Ed25519Key',
#'Curve25519Key': 'toot:Curve25519Key',
#'EncryptedMessage': 'toot:EncryptedMessage',
#'publicKeyBase64': 'toot:publicKeyBase64',
#'deviceId': 'toot:deviceId',
#'messageFranking': 'toot:messageFranking',
'messageType': 'toot:messageType',
#'cipherText': 'toot:cipherText',
#'suspended': 'toot:suspended',
"claim": {
"@type": "@id",
"@id": "toot:claim"
}
}
],
'id': actor,
'type': actor_type,
'inbox': kwargs.get('inbox', f'{actor}'),
'outbox': f'{actor}/outbox',
'preferredUsername': handle,
'url': kwargs.get('url', actor),
'manuallyApprovesFollowers': kwargs.get('locked', False),
'discoverable': kwargs.get('discoverable', False),
'published': published or DateString.now('activitypub'),
'publicKey': {
'id': f'{actor}#main-key',
'owner': actor,
'publicKeyPem': pubkey
},
'endpoints': {
'sharedInbox': kwargs.get('shared_inbox', f'https://{actor.host}/inbox')
}
})
for key, value in table.items():
data.attachment.append(PropertyValue(key, value))
if kwargs.get('avatar_url'):
data.icon = Object.new_image(kwargs.get('avatar_url'), kwargs.get('avatar_type'))
if full:
data.update({
'name': kwargs.get('display_name', handle),
'summary': kwargs.get('bio'),
'featured': f'{actor}/collections/featured',
'tags': f'{actor}/collections/tags',
'following': f'{actor}/following',
'followers': f'{actor}/followers',
'tag': [],
'attachment': [],
})
data['@context'][2].update({
'manuallyApprovesFollowers': 'as:manuallyApprovesFollowers',
'discoverable': 'toot:discoverable',
'PropertyValue': 'schema:PropertyValue',
'value': 'schema:value',
'Emoji': 'toot:Emoji',
"featured": {
"@id": "toot:featured",
"@type": "@id"
},
"featuredTags": {
"@id": "toot:featuredTags",
"@type": "@id"
},
"alsoKnownAs": {
"@id": "as:alsoKnownAs",
"@type": "@id"
},
"movedTo": {
"@id": "as:movedTo",
"@type": "@id"
},
"claim": {
"@type": "@id",
"@id": "toot:claim"
},
"focalPoint": {
"@container": "@list",
"@id": "toot:focalPoint"
}
})
return data
# not complete
@classmethod
def new_actor_old(cls, actor, handle, pubkey, published=None, table={}, full=True, **kwargs):
actor_type = kwargs.get('type', 'Person').title()
assert actor_type in actor_types
actor = Url(actor)
data = cls({
'@context': [
'https://www.w3.org/ns/activitystreams',
'https://w3id.org/security/v1',
{
'schema': 'http://schema.org',
'toot': 'https://joinmastodon.org/ns#',
'manuallyApprovesFollowers': 'as:manuallyApprovesFollowers',
'PropertyValue': 'schema:PropertyValue',
'value': 'schema:value',
'IdentityProof': 'toot:IdentityProof',
'discoverable': 'toot:discoverable',
'Device': 'toot:Device',
'Ed25519Signature': 'toot:Ed25519Signature',
'Ed25519Key': 'toot:Ed25519Key',
'Curve25519Key': 'toot:Curve25519Key',
'EncryptedMessage': 'toot:EncryptedMessage',
'publicKeyBase64': 'toot:publicKeyBase64',
'deviceId': 'toot:deviceId',
'messageFranking': 'toot:messageFranking',
'messageType': 'toot:messageType',
'cipherText': 'toot:cipherText',
'suspended': 'toot:suspended',
'Emoji': 'toot:Emoji',
"featured": {
"@id": "toot:featured",
"@type": "@id"
},
"featuredTags": {
"@id": "toot:featuredTags",
"@type": "@id"
},
"alsoKnownAs": {
"@id": "as:alsoKnownAs",
"@type": "@id"
},
"movedTo": {
"@id": "as:movedTo",
"@type": "@id"
},
"claim": {
"@type": "@id",
"@id": "toot:claim"
},
"fingerprintKey": {
"@type": "@id",
"@id": "toot:fingerprintKey"
},
"identityKey": {
"@type": "@id",
"@id": "toot:identityKey"
},
"devices": {
"@type": "@id",
"@id": "toot:devices"
},
"focalPoint": {
"@container": "@list",
"@id": "toot:focalPoint"
}
}
],
'id': actor,
'type': actor_type,
'following': f'{actor}/following',
'followers': f'{actor}/followers',
'inbox': kwargs.get('inbox', f'{actor}'),
'outbox': f'{actor}/outbox',
'featured': f'{actor}/collections/featured',
'featuredTags': f'{actor}/collections/tags',
'preferredUsername': handle,
'name': kwargs.get('display_name', handle),
'summary': kwargs.get('bio'),
'url': kwargs.get('url', actor),
'manuallyApprovesFollowers': kwargs.get('locked', False),
'discoverable': kwargs.get('discoverable', False),
'published': published or DateString.now('activitypub'),
'devices': f'{actor}/collections/devices',
'publicKey': {
'id': f'{actor}#main-key',
'owner': actor,
'publicKeyPem': pubkey
},
'tag': [],
'attachment': [],
'endpoints': {
'sharedInbox': kwargs.get('shared_inbox', f'https://{actor.host}/inbox')
}
})
for key, value in table.items():
data.attachment.append(PropertyValue(key, value))
if kwargs.get('avatar_url'):
data.icon = Object.new_image(kwargs.get('avatar_url'), kwargs.get('avatar_type'))
if kwargs.get('header_url'):
data.image = Object.new_image(kwargs.get('header_url'), kwargs.get('header_type'))
# need to add data when "full" is true
if not full:
del data.featured
del data.featuredTags
del data.devices
del data.following
del data.followers
del data.outbox
return data
@classmethod
def new_follow(cls, id, actor, target):
return cls({
'@context': 'https://www.w3.org/ns/activitystreams',
'id': id,
'type': 'Follow',
'actor': actor,
'object': target
})
@classmethod
def new_emoji(cls, id, name, url, image):
return cls({
'@context': 'https://www.w3.org/ns/activitystreams',
'id': id,
'type': 'Emoji',
'name': name,
'updated': updated or DateTime.now('activitypub'),
'icon': image
})
@property
def privacy_level(self):
return parse_privacy_level(
self.get('to', []),
self.get('cc', []),
self.get('attributedTo', '') + '/followers'
)
@property
def shared_inbox(self):
try: return self.endpoints.shared_inbox
except AttributeError: pass
@property
def pubkey(self):
try: return self.publicKey.publicKeyPem
except AttributeError: pass
@property
def handle(self):
return self['preferredUsername']
@property
def display_name(self):
return self.get('name')
@property
def type(self):
return self['type'].capitalize()
@property
def info_table(self):
return DotDict({p['name']: p['value'] for p in self.get('attachment', {})})
@property
def domain(self):
return self.id.host
@property
def bio(self):
return self.get('summary')
@property
def avatar(self):
return self.icon.url
@property
def header(self):
return self.image.url
class Collection(Object):
@classmethod
def new_replies(cls, statusid):
return cls({
'@context': 'https://www.w3.org/ns/activitystreams',
'id': f'{statusid}/replies',
'type': 'Collection',
'first': {
'type': 'CollectionPage',
'next': f'{statusid}/replies?only_other_accounts=true&page=true',
'partOf': f'{statusid}/replies',
'items': []
}
})
@classmethod
def new_collection(cls, outbox, min_id=0, total=0):
return cls({
'@context': 'https://www.w3.org/ns/activitystreams',
'id': outbox,
'type': 'OrderedCollection',
'totalItems': total,
'first': f'{outbox}?page=true',
'last': f'{outbox}?min_id=0&page=true'
})
@classmethod
def new_page(cls, outbox, min_id, max_id, *items):
return cls({
'@context': [
'https://www.w3.org/ns/activitystreams',
{
'sensitive': 'as:sensitive',
'toot': 'http://joinmastodon.org/ns#',
'votersCount': 'toot:votersCount',
'litepub': 'http://litepub.social/ns#',
'directMessage': 'litepub:directMessage',
}
],
'id': f'{outbox}?page=true',
'type': 'OrderedCollectionPage',
'next': f'{outbox}?max_id={max_id}&page=true',
'prev': f'{outbox}?min_id={min_id}&page=true',
'partOf': outbox,
'orderedItems': items
})
### sub-objects ###
class PropertyValue(DotDict):
def __init__(self, key, value):
super().__init__({
'type': 'PropertyValue',
'name': key,
'value': value
})
def __setitem__(self, key, value):
key = key.lower()
assert key in ['type', 'name', 'value']
assert type(value) == str
super().__setitem__(key, value)
def set_pair(self, key, value):
self.name = key
self.value = value
class Media(Object):
@classmethod
def new(cls, type, url, mime=None):
return cls(
type = 'Image',
mediaType = mime or mimetypes.guess_type(url)[0] or 'image/png',
url = url
)
@classmethod
def new_image(cls, url, mime=None):
return cls.new('Image', url, mime)
@classmethod
def new_video(cls, url, mime=None):
return cls.new('Video', url, mime)
@classmethod
def new_audio(cls, url, mime=None):
return cls.new('Audio', url, mime)
class Emoji(DotDict):
@classmethod
def new(cls, id, name, image):
return cls({
'id': id,
'type': Emoji,
'name': f':{name}:',
'icon': image
})
### Not activitypub objects, but related ###
class Nodeinfo(DotDict):
@property
def name(self):
return self.software.name
@property
def version(self):
return self.software.version
@property
def repo(self):
return self.software.repository
@property
def homepage(self):
return self.software.homepage
@property
def users(self):
return self.usage.users.total
@property
def posts(self):
return self.usage.localPosts
@classmethod
def new_20(cls, name, version, **metadata):
return cls.new(name, version, '2.0', **metadata)
@classmethod
def new_21(cls, name, version, **metadata):
return cls.new(name, version, '2.1', **metadata)
@classmethod
def new(cls, name, version, niversion='2.1', **kwargs):
assert niversion in ['2.0', '2.1']
open_regs = boolean(kwargs.pop('open_regs', True))
posts = int(kwargs.pop('posts', 0))
users = int(kwargs.pop('users', 0))
users_halfyear = int(kwargs.pop('halfyear', 0))
users_month = int(kwargs.pop('month', 0))
comments = int(kwargs.pop('comments', 0))
repository = kwargs.pop('repository', None)
homepage = kwargs.pop('homepage', None)
data = cls(
version = niversion,
openRegistrations = open_regs,
software = {
'name': name.lower().replace(' ', '-'),
'version': version
},
usage = {
'users': {
'total': users
}
},
protocols = [
'activitypub'
],
services = {
'inbound': kwargs.pop('inbound', []),
'outbound': kwargs.pop('outbound', [])
}
)
if niversion == '2.1':
if repository:
data.software.repository = repository
if homepage:
data.software.homepage = homepage
if users_halfyear:
data.users.activeHalfyear = halfyear
if users_month:
data.users.activeMonth = month
if posts:
data.usage.localPosts = posts
if comments:
data.usage.localComments = comments
if kwargs:
data.metadata = kwargs
return data
class WellknownNodeinfo(DotDict):
def url(self, version='2.1'):
assert version in ['2.0', '2.1']
for link in self.links:
if link['rel'].endswith(version):
return link['href']
@classmethod
def new(cls, path, version='2.1'):
data = cls(links=[])
data.append(path, version)
return data
def append(self, path, version='2.1'):
assert version in ['2.0', '2.1']
self.links.append({
'rel': f'http://nodeinfo.dispora.software/ns/schema/{version}',
'href': path
})
class Hostmeta(str):
def __new__(cls, text):
return str.__new__(cls, text)
@classmethod
def new(cls, domain):
return cls(f'<?xml version="1.0" encoding="UTF-8"?><XRD xmlns="http://docs.oasis-open.org/ns/xri/xrd-1.0"><Link rel="lrdd" template="https://{domain}/.well-known/webfinger?resource={{uri}}"/></XRD>')
@property
def link(self):
return Url(fromstring(self)[0].attrib['template'])
class Webfinger(DotDict):
@property
def profile(self):
for link in self.links:
if link['rel'] == 'http://webfinger.net/rel/profile-page':
return link['href']
@property
def actor(self):
for link in self.links:
if link['rel'] == 'self':
return link['href']
@property
def fullname(self):
return self.subject[5:]
@property
def handle(self):
return self.fullname.split('@')[0]
@property
def domain(self):
return self.fullname.split('@')[1]
@classmethod
def new(cls, handle, domain, actor, profile=None):
data = cls(
subject = f'acct:{handle}@{domain}',
aliases = [actor],
links = [
{
'rel': 'self',
'type': 'application/activity+json',
'href': actor
}
]
)
if profile:
data.aliases.append(profile)
data.links.append({
'rel': 'http://webfinger.net/rel/profile-page',
'type': 'text/html',
'href': profile
})
return data