Add donation link

added version info

docs/commands.md

@@ -24,6 +24,27 @@ Run the setup wizard to configure your relay.
 	activityrelay setup
 
 
+## Config
+
+Manage the relay config
+
+	activityrelay config
+
+
+### List
+
+List the current config key/value pairs
+
+	activityrelay config list
+
+
+### Set
+
+Set a value for a config option
+
+	activityrelay config set <key> <value>
+
+
 ## Inbox
 
 Manage the list of subscribed instances.
@@ -97,6 +118,13 @@ Remove a domain from the whitelist.
 	activityrelay whitelist remove <domain>
 
 
+### Import
+
+Add all current inboxes to the whitelist
+
+	activityrelay whitelist import
+
+
 ## Instance
 
 Manage the instance ban list.

docs/configuration.md

@@ -1,6 +1,8 @@
 # Configuration
 
-## DB
+## General
+
+### DB
 
 The path to the database. It contains the relay actor private key and all subscribed
 instances. If the path is not absolute, it is relative to the working directory.
@@ -8,7 +10,7 @@ instances. If the path is not absolute, it is relative to the working directory.
 	db: relay.jsonld
 
 
-## Listener
+### Listener
 
 The address and port the relay will listen on. If the reverse proxy (nginx, apache, caddy, etc)
 is running on the same host, it is recommended to change `listen` to `localhost`
@@ -17,22 +19,41 @@ is running on the same host, it is recommended to change `listen` to `localhost`
 	port: 8080
 
 
-## Note
+### Note
 
 A small blurb to describe your relay instance. This will show up on the relay's home page.
 
 	note: "Make a note about your instance here."
 
 
-## Post Limit
+### Post Limit
 
 The maximum number of messages to send out at once. For each incoming message, a message will be
 sent out to every subscribed instance minus the instance which sent the message. This limit
 is to prevent too many outgoing connections from being made, so adjust if necessary.
 
+Note: If the `workers` option is set to anything above 0, this limit will be per worker.
+
 	push_limit: 512
 
 
+### Push Workers
+
+The relay can be configured to use threads to push messages out. For smaller relays, this isn't
+necessary, but bigger ones (>100 instances) will want to set this to the number of available cpu
+threads.
+
+	workers: 0
+
+
+### JSON GET cache limit
+
+JSON objects (actors, nodeinfo, etc) will get cached when fetched. This will set the max number of
+objects to keep in the cache.
+
+	json_cache: 1024
+
+
 ## AP
 
 Various ActivityPub-related settings
@@ -82,29 +103,3 @@ setting this to the below list will block all other relays and prevent relay cha
 	- aoderelay
 	- social.seattle.wa.us-relay
 	- unciarelay
-
-
-## Cache
-
-These are object limits for various caches. Only change if you know what you're doing.
-
-
-### Objects
-
-The urls of messages which have been processed by the relay.
-
-	objects: 1024
-
-
-### Actors
-
-The ActivityPub actors of incoming messages.
-
-	actors: 1024
-
-
-### Actors
-
-The base64 encoded hashes of messages.
-
-	digests: 1024

docs/installation.md

@@ -15,7 +15,7 @@ the [official pipx docs](https://pypa.github.io/pipx/installation/) for more in-
 
 Now simply install ActivityRelay directly from git
 
-	pipx install git+https://git.pleroma.social/pleroma/relay@0.2.0
+	pipx install git+https://git.pleroma.social/pleroma/relay@0.2.4
 
 Or from a cloned git repo.
 
@@ -39,7 +39,7 @@ be installed via [pyenv](https://github.com/pyenv/pyenv).
 The instructions for installation via pip are very similar to pipx. Installation can be done from
 git
 
-	python3 -m pip install git+https://git.pleroma.social/pleroma/relay@0.2.0
+	python3 -m pip install git+https://git.pleroma.social/pleroma/relay@0.2.4
 
 or a cloned git repo.
 

relay.spec

@@ -0,0 +1,50 @@
+# -*- mode: python ; coding: utf-8 -*-
+
+
+block_cipher = None
+
+
+a = Analysis(
+    ['relay/__main__.py'],
+    pathex=[],
+    binaries=[],
+    datas=[],
+    hiddenimports=[
+        'aputils.enums',
+        'aputils.errors',
+        'aputils.misc',
+        'aputils.objects',
+        'aputils.signer'
+    ],
+    hookspath=[],
+    hooksconfig={},
+    runtime_hooks=[],
+    excludes=[],
+    win_no_prefer_redirects=False,
+    win_private_assemblies=False,
+    cipher=block_cipher,
+    noarchive=False,
+)
+pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
+
+exe = EXE(
+    pyz,
+    a.scripts,
+    a.binaries,
+    a.zipfiles,
+    a.datas,
+    [],
+    name='activityrelay',
+    debug=False,
+    bootloader_ignore_signals=False,
+    strip=False,
+    upx=True,
+    upx_exclude=[],
+    runtime_tmpdir=None,
+    console=True,
+    disable_windowed_traceback=False,
+    argv_emulation=False,
+    target_arch=None,
+    codesign_identity=None,
+    entitlements_file=None,
+)

relay.yaml.example

@@ -9,13 +9,19 @@ port: 8080
 # Note
 note: "Make a note about your instance here."
 
-# maximum number of inbox posts to do at once
-post_limit: 512
+# Number of worker threads to start. If 0, use asyncio futures instead of threads.
+workers: 0
+
+# Maximum number of inbox posts to do at once
+# If workers is set to 1 or above, this is the max for each worker
+push_limit: 512
+
+# The amount of json objects to cache from GET requests
+json_cache: 1024
 
-# this section is for ActivityPub
 ap:
-  # this is used for generating activitypub messages, as well as instructions for
-  # linking AP identities.  it should be an SSL-enabled domain reachable by https.
+  # This is used for generating activitypub messages, as well as instructions for
+  # linking AP identities.  It should be an SSL-enabled domain reachable by https.
   host: 'relay.example.com'
 
   blocked_instances:
@@ -35,9 +41,3 @@ ap:
   #- 'aoderelay'
   #- 'social.seattle.wa.us-relay'
   #- 'unciarelay'
-
-# cache limits as number of items. only change this if you know what you're doing
-cache:
-  objects: 1024
-  actors: 1024
-  digests: 1024

relay/__init__.py

@@ -1,8 +1,3 @@
-__version__ = '0.2.2'
-
-from aiohttp.web import Application
+__version__ = '0.2.4'
 
 from . import logger
-
-
-app = Application()

relay/__main__.py

@@ -1,4 +1,4 @@
-from .manage import main
+from relay.manage import main
 
 
 if __name__ == '__main__':

relay/application.py

@@ -0,0 +1,216 @@
+import asyncio
+import logging
+import os
+import queue
+import signal
+import threading
+import traceback
+
+from aiohttp import web
+from datetime import datetime, timedelta
+
+from .config import RelayConfig
+from .database import RelayDatabase
+from .http_client import HttpClient
+from .misc import DotDict, check_open_port, set_app
+from .views import routes
+
+
+class Application(web.Application):
+	def __init__(self, cfgpath):
+		web.Application.__init__(self)
+
+		self['starttime'] = None
+		self['running'] = False
+		self['config'] = RelayConfig(cfgpath)
+
+		if not self['config'].load():
+			self['config'].save()
+
+		if self.config.is_docker:
+			self.config.update({
+				'db': '/data/relay.jsonld',
+				'listen': '0.0.0.0',
+				'port': 8080
+			})
+
+		self['workers'] = []
+		self['last_worker'] = 0
+
+		set_app(self)
+
+		self['database'] = RelayDatabase(self['config'])
+		self['database'].load()
+
+		self['client'] = HttpClient(
+			database = self.database,
+			limit = self.config.push_limit,
+			timeout = self.config.timeout,
+			cache_size = self.config.json_cache
+		)
+
+		self.set_signal_handler()
+
+
+	@property
+	def client(self):
+		return self['client']
+
+
+	@property
+	def config(self):
+		return self['config']
+
+
+	@property
+	def database(self):
+		return self['database']
+
+
+	@property
+	def uptime(self):
+		if not self['starttime']:
+			return timedelta(seconds=0)
+
+		uptime = datetime.now() - self['starttime']
+
+		return timedelta(seconds=uptime.seconds)
+
+
+	def push_message(self, inbox, message):
+		if self.config.workers <= 0:
+			return asyncio.ensure_future(self.client.post(inbox, message))
+
+		worker = self['workers'][self['last_worker']]
+		worker.queue.put((inbox, message))
+
+		self['last_worker'] += 1
+
+		if self['last_worker'] >= len(self['workers']):
+			self['last_worker'] = 0
+
+
+	def set_signal_handler(self):
+		for sig in {'SIGHUP', 'SIGINT', 'SIGQUIT', 'SIGTERM'}:
+			try:
+				signal.signal(getattr(signal, sig), self.stop)
+
+			# some signals don't exist in windows, so skip them
+			except AttributeError:
+				pass
+
+
+	def run(self):
+		if not check_open_port(self.config.listen, self.config.port):
+			return logging.error(f'A server is already running on port {self.config.port}')
+
+		for route in routes:
+			self.router.add_route(*route)
+
+		logging.info(f'Starting webserver at {self.config.host} ({self.config.listen}:{self.config.port})')
+		asyncio.run(self.handle_run())
+
+
+	def stop(self, *_):
+		self['running'] = False
+
+
+	async def handle_run(self):
+		self['running'] = True
+
+		if self.config.workers > 0:
+			for i in range(self.config.workers):
+				worker = PushWorker(self)
+				worker.start()
+
+				self['workers'].append(worker)
+
+		runner = web.AppRunner(self, access_log_format='%{X-Forwarded-For}i "%r" %s %b "%{User-Agent}i"')
+		await runner.setup()
+
+		site = web.TCPSite(runner,
+			host = self.config.listen,
+			port = self.config.port,
+			reuse_address = True
+		)
+
+		await site.start()
+		self['starttime'] = datetime.now()
+
+		while self['running']:
+			await asyncio.sleep(0.25)
+
+		await site.stop()
+
+		self['starttime'] = None
+		self['running'] = False
+		self['workers'].clear()
+
+
+class PushWorker(threading.Thread):
+	def __init__(self, app):
+		threading.Thread.__init__(self)
+		self.app = app
+		self.queue = queue.Queue()
+
+
+	def run(self):
+		self.client = HttpClient(
+			database = self.app.database,
+			limit = self.app.config.push_limit,
+			timeout = self.app.config.timeout,
+			cache_size = self.app.config.json_cache
+		)
+
+		asyncio.run(self.handle_queue())
+
+
+	async def handle_queue(self):
+		while self.app['running']:
+			try:
+				inbox, message = self.queue.get(block=True, timeout=0.25)
+				self.queue.task_done()
+				logging.verbose(f'New push from Thread-{threading.get_ident()}')
+				await self.client.post(inbox, message)
+
+			except queue.Empty:
+				pass
+
+			## make sure an exception doesn't bring down the worker
+			except Exception:
+				traceback.print_exc()
+
+		await self.client.close()
+
+
+## Can't sub-class web.Request, so let's just add some properties
+def request_actor(self):
+	try: return self['actor']
+	except KeyError: pass
+
+
+def request_instance(self):
+	try: return self['instance']
+	except KeyError: pass
+
+
+def request_message(self):
+	try: return self['message']
+	except KeyError: pass
+
+
+def request_signature(self):
+	if 'signature' not in self._state:
+		try: self['signature'] = DotDict.new_from_signature(self.headers['signature'])
+		except KeyError: return
+
+	return self['signature']
+
+
+setattr(web.Request, 'actor', property(request_actor))
+setattr(web.Request, 'instance', property(request_instance))
+setattr(web.Request, 'message', property(request_message))
+setattr(web.Request, 'signature', property(request_signature))
+
+setattr(web.Request, 'config', property(lambda self: self.app.config))
+setattr(web.Request, 'database', property(lambda self: self.app.database))

relay/config.py

@@ -1,109 +1,51 @@
 import json
+import os
 import yaml
 
+from functools import cached_property
 from pathlib import Path
 from urllib.parse import urlparse
 
+from .misc import DotDict, boolean
 
-relay_software_names = [
-	'activityrelay',
-	'aoderelay',
-	'social.seattle.wa.us-relay',
-	'unciarelay'
+
+RELAY_SOFTWARE = [
+	'activityrelay', # https://git.pleroma.social/pleroma/relay
+	'aoderelay', # https://git.asonix.dog/asonix/relay
+	'feditools-relay' # https://git.ptzo.gdn/feditools/relay
+]
+
+APKEYS = [
+	'host',
+	'whitelist_enabled',
+	'blocked_software',
+	'blocked_instances',
+	'whitelist'
 ]
 
 
-class DotDict(dict):
-	def __getattr__(self, k):
-		try:
-			return self[k]
-
-		except KeyError:
-			raise AttributeError(f'{self.__class__.__name__} object has no attribute {k}') from None
-
-
-	def __setattr__(self, k, v):
-		try:
-			if k in self._ignore_keys:
-				super().__setattr__(k, v)
-
-		except AttributeError:
-			pass
-
-		if k.startswith('_'):
-			super().__setattr__(k, v)
-
-		else:
-			self[k] = v
-
-
-	def __setitem__(self, k, v):
-		if type(v) == dict:
-			v = DotDict(v)
-
-		super().__setitem__(k, v)
-
-
-	def __delattr__(self, k):
-		try:
-			dict.__delitem__(self, k)
-
-		except KeyError:
-			raise AttributeError(f'{self.__class__.__name__} object has no attribute {k}') from None
-
-
 class RelayConfig(DotDict):
-	apkeys = {
-		'host',
-		'whitelist_enabled',
-		'blocked_software',
-		'blocked_instances',
-		'whitelist'
-	}
+	def __init__(self, path):
+		DotDict.__init__(self, {})
 
-	cachekeys = {
-		'json',
-		'objects',
-		'digests'
-	}
+		if self.is_docker:
+			path = '/data/config.yaml'
 
-
-	def __init__(self, path, is_docker):
-		if is_docker:
-			path = '/data/relay.yaml'
-
-		self._isdocker = is_docker
 		self._path = Path(path).expanduser()
-
-		super().__init__({
-			'db': str(self._path.parent.joinpath(f'{self._path.stem}.jsonld')),
-			'listen': '0.0.0.0',
-			'port': 8080,
-			'note': 'Make a note about your instance here.',
-			'push_limit': 512,
-			'host': 'relay.example.com',
-			'blocked_software': [],
-			'blocked_instances': [],
-			'whitelist': [],
-			'whitelist_enabled': False,
-			'json': 1024,
-			'objects': 1024,
-			'digests': 1024
-		})
+		self.reset()
 
 
 	def __setitem__(self, key, value):
-		if self._isdocker and key in ['db', 'listen', 'port']:
-			return
-
 		if key in ['blocked_instances', 'blocked_software', 'whitelist']:
 			assert isinstance(value, (list, set, tuple))
 
-		elif key in ['port', 'json', 'objects', 'digests']:
-			assert isinstance(value, (int))
+		elif key in ['port', 'workers', 'json_cache', 'timeout']:
+			if not isinstance(value, int):
+				value = int(value)
 
 		elif key == 'whitelist_enabled':
-			assert isinstance(value, bool)
+			if not isinstance(value, bool):
+				value = boolean(value)
 
 		super().__setitem__(key, value)
 
@@ -133,6 +75,30 @@ class RelayConfig(DotDict):
 		return f'{self.actor}#main-key'
 
 
+	@cached_property
+	def is_docker(self):
+		return bool(os.environ.get('DOCKER_RUNNING'))
+
+
+	def reset(self):
+		self.clear()
+		self.update({
+			'db': str(self._path.parent.joinpath(f'{self._path.stem}.jsonld')),
+			'listen': '0.0.0.0',
+			'port': 8080,
+			'note': 'Make a note about your instance here.',
+			'push_limit': 512,
+			'json_cache': 1024,
+			'timeout': 10,
+			'workers': 0,
+			'host': 'relay.example.com',
+			'whitelist_enabled': False,
+			'blocked_software': [],
+			'blocked_instances': [],
+			'whitelist': []
+		})
+
+
 	def ban_instance(self, instance):
 		if instance.startswith('http'):
 			instance = urlparse(instance).hostname
@@ -218,6 +184,8 @@ class RelayConfig(DotDict):
 
 
 	def load(self):
+		self.reset()
+
 		options = {}
 
 		try:
@@ -237,13 +205,15 @@ class RelayConfig(DotDict):
 			return False
 
 		for key, value in config.items():
-			if key in ['ap', 'cache']:
+			if key in ['ap']:
 				for k, v in value.items():
 					if k not in self:
 						continue
 
 					self[k] = v
 
+				continue
+
 			elif key not in self:
 				continue
 
@@ -257,13 +227,16 @@ class RelayConfig(DotDict):
 
 	def save(self):
 		config = {
-			'db': self['db'],
+			# just turning config.db into a string is good enough for now
+			'db': str(self.db),
 			'listen': self.listen,
 			'port': self.port,
 			'note': self.note,
 			'push_limit': self.push_limit,
-			'ap': {key: self[key] for key in self.apkeys},
-			'cache': {key: self[key] for key in self.cachekeys}
+			'workers': self.workers,
+			'json_cache': self.json_cache,
+			'timeout': self.timeout,
+			'ap': {key: self[key] for key in APKEYS}
 		}
 
 		with open(self._path, 'w') as fd:

relay/database.py

@@ -1,50 +1,33 @@
+import aputils
+import asyncio
 import json
 import logging
 import traceback
 
-from Crypto.PublicKey import RSA
 from urllib.parse import urlparse
 
 
-class RelayDatabase:
+class RelayDatabase(dict):
 	def __init__(self, config):
+		dict.__init__(self, {
+			'relay-list': {},
+			'private-key': None,
+			'follow-requests': {},
+			'version': 1
+		})
+
 		self.config = config
-		self.data = None
-		self.PRIVKEY = None
-
-
-	@property
-	def PUBKEY(self):
-		return self.PRIVKEY.publickey()
-
-
-	@property
-	def pubkey(self):
-		return self.PUBKEY.exportKey('PEM').decode('utf-8')
-
-
-	@property
-	def privkey(self):
-		try:
-			return self.data['private-key']
-
-		except KeyError:
-			return False
+		self.signer = None
 
 
 	@property
 	def hostnames(self):
-		return [urlparse(inbox).hostname for inbox in self.inboxes]
+		return tuple(self['relay-list'].keys())
 
 
 	@property
 	def inboxes(self):
-		return self.data.get('relay-list', [])
-
-
-	def generate_key(self):
-		self.PRIVKEY = RSA.generate(4096)
-		self.data['private-key'] = self.PRIVKEY.exportKey('PEM').decode('utf-8')
+		return tuple(data['inbox'] for data in self['relay-list'].values())
 
 
 	def load(self):
@@ -52,14 +35,36 @@ class RelayDatabase:
 
 		try:
 			with self.config.db.open() as fd:
-				self.data = json.load(fd)
+				data = json.load(fd)
 
-			key = self.data.pop('actorKeys', None)
+			self['version'] = data.get('version', None)
+			self['private-key'] = data.get('private-key')
 
-			if key:
-				self.data['private-key'] = key.get('privateKey')
+			if self['version'] == None:
+				self['version'] = 1
+
+				if 'actorKeys' in data:
+					self['private-key'] = data['actorKeys']['privateKey']
+
+				for item in data.get('relay-list', []):
+					domain = urlparse(item).hostname
+					self['relay-list'][domain] = {
+						'domain': domain,
+						'inbox': item,
+						'followid': None
+					}
+
+			else:
+				self['relay-list'] = data.get('relay-list', {})
+
+			for domain, instance in self['relay-list'].items():
+				if self.config.is_banned(domain) or (self.config.whitelist_enabled and not self.config.is_whitelisted(domain)):
+					self.del_inbox(domain)
+					continue
+
+				if not instance.get('domain'):
+					instance['domain'] = domain
 
-			self.data.pop('actors', None)
 			new_db = False
 
 		except FileNotFoundError:
@@ -69,20 +74,13 @@ class RelayDatabase:
 			if self.config.db.stat().st_size > 0:
 				raise e from None
 
-		if not self.data:
-			logging.info('No database was found. Making a new one.')
-			self.data = {}
-
-		for inbox in self.inboxes:
-			if self.config.is_banned(inbox) or (self.config.whitelist_enabled and not self.config.is_whitelisted(inbox)):
-				self.del_inbox(inbox)
-
-		if not self.privkey:
+		if not self['private-key']:
 			logging.info("No actor keys present, generating 4096-bit RSA keypair.")
-			self.generate_key()
+			self.signer = aputils.Signer.new(self.config.keyid, size=4096)
+			self['private-key'] = self.signer.export()
 
 		else:
-			self.PRIVKEY = RSA.importKey(self.privkey)
+			self.signer = aputils.Signer(self['private-key'], self.config.keyid)
 
 		self.save()
 		return not new_db
@@ -90,34 +88,110 @@ class RelayDatabase:
 
 	def save(self):
 		with self.config.db.open('w') as fd:
-			data = {
-				'relay-list': self.inboxes,
-				'private-key': self.privkey
-			}
-
-			json.dump(data, fd, indent=4)
+			json.dump(self, fd, indent=4)
 
 
-	def get_inbox(self, domain):
+	def get_inbox(self, domain, fail=False):
 		if domain.startswith('http'):
 			domain = urlparse(domain).hostname
 
-		for inbox in self.inboxes:
-			if domain == urlparse(inbox).hostname:
-				return inbox
+		inbox = self['relay-list'].get(domain)
+
+		if inbox:
+			return inbox
+
+		if fail:
+			raise KeyError(domain)
 
 
-	def add_inbox(self, inbox):
-		assert inbox.startswith('https')
-		assert not self.get_inbox(inbox)
+	def add_inbox(self, inbox, followid=None, software=None):
+		assert inbox.startswith('https'), 'Inbox must be a url'
+		domain = urlparse(inbox).hostname
+		instance = self.get_inbox(domain)
 
-		self.data['relay-list'].append(inbox)
+		if instance:
+			if followid:
+				instance['followid'] = followid
+
+			if software:
+				instance['software'] = software
+
+			return instance
+
+		self['relay-list'][domain] = {
+			'domain': domain,
+			'inbox': inbox,
+			'followid': followid,
+			'software': software
+		}
+
+		logging.verbose(f'Added inbox to database: {inbox}')
+		return self['relay-list'][domain]
 
 
-	def del_inbox(self, inbox_url):
-		inbox = self.get_inbox(inbox_url)
+	def del_inbox(self, domain, followid=None, fail=False):
+		data = self.get_inbox(domain, fail=False)
 
-		if not inbox:
-			raise KeyError(inbox_url)
+		if not data:
+			if fail:
+				raise KeyError(domain)
 
-		self.data['relay-list'].remove(inbox)
+			return False
+
+		if not data['followid'] or not followid or data['followid'] == followid:
+			del self['relay-list'][data['domain']]
+			logging.verbose(f'Removed inbox from database: {data["inbox"]}')
+			return True
+
+		if fail:
+			raise ValueError('Follow IDs do not match')
+
+		logging.debug(f'Follow ID does not match: db = {data["followid"]}, object = {followid}')
+		return False
+
+
+	def get_request(self, domain, fail=True):
+		if domain.startswith('http'):
+			domain = urlparse(domain).hostname
+
+		try:
+			return self['follow-requests'][domain]
+
+		except KeyError as e:
+			if fail:
+				raise e
+
+
+	def add_request(self, actor, inbox, followid):
+		domain = urlparse(inbox).hostname
+
+		try:
+			request = self.get_request(domain)
+			request['followid'] = followid
+
+		except KeyError:
+			pass
+
+		self['follow-requests'][domain] = {
+			'actor': actor,
+			'inbox': inbox,
+			'followid': followid
+		}
+
+
+	def del_request(self, domain):
+		if domain.startswith('http'):
+			domain = urlparse(inbox).hostname
+
+		del self['follow-requests'][domain]
+
+
+	def distill_inboxes(self, message):
+		src_domains = {
+			message.domain,
+			urlparse(message.objectid).netloc
+		}
+
+		for domain, instance in self['relay-list'].items():
+			if domain not in src_domains:
+				yield instance['inbox']

relay/http_client.py

@@ -0,0 +1,222 @@
+import logging
+import traceback
+
+from aiohttp import ClientSession, ClientTimeout, TCPConnector
+from aiohttp.client_exceptions import ClientConnectionError, ClientSSLError
+from asyncio.exceptions import TimeoutError as AsyncTimeoutError
+from aputils import Nodeinfo, WellKnownNodeinfo
+from datetime import datetime
+from cachetools import LRUCache
+from json.decoder import JSONDecodeError
+from urllib.parse import urlparse
+
+from . import __version__
+from .misc import (
+	MIMETYPES,
+	DotDict,
+	Message
+)
+
+
+HEADERS = {
+	'Accept': f'{MIMETYPES["activity"]}, {MIMETYPES["json"]};q=0.9',
+	'User-Agent': f'ActivityRelay/{__version__}'
+}
+
+
+class Cache(LRUCache):
+	def set_maxsize(self, value):
+		self.__maxsize = int(value)
+
+
+class HttpClient:
+	def __init__(self, database, limit=100, timeout=10, cache_size=1024):
+		self.database = database
+		self.cache = Cache(cache_size)
+		self.cfg = {'limit': limit, 'timeout': timeout}
+		self._conn = None
+		self._session = None
+
+
+	async def __aenter__(self):
+		await self.open()
+		return self
+
+
+	async def __aexit__(self, *_):
+		await self.close()
+
+
+	@property
+	def limit(self):
+		return self.cfg['limit']
+
+
+	@property
+	def timeout(self):
+		return self.cfg['timeout']
+
+
+	async def open(self):
+		if self._session:
+			return
+
+		self._conn = TCPConnector(
+			limit = self.limit,
+			ttl_dns_cache = 300,
+		)
+
+		self._session = ClientSession(
+			connector = self._conn,
+			headers = HEADERS,
+			connector_owner = True,
+			timeout = ClientTimeout(total=self.timeout)
+		)
+
+
+	async def close(self):
+		if not self._session:
+			return
+
+		await self._session.close()
+		await self._conn.close()
+
+		self._conn = None
+		self._session = None
+
+
+	async def get(self, url, sign_headers=False, loads=None, force=False):
+		await self.open()
+
+		try: url, _ = url.split('#', 1)
+		except: pass
+
+		if not force and url in self.cache:
+			return self.cache[url]
+
+		headers = {}
+
+		if sign_headers:
+			headers.update(self.database.signer.sign_headers('GET', url, algorithm='original'))
+
+		try:
+			logging.verbose(f'Fetching resource: {url}')
+
+			async with self._session.get(url, headers=headers) as resp:
+				## Not expecting a response with 202s, so just return
+				if resp.status == 202:
+					return
+
+				elif resp.status != 200:
+					logging.verbose(f'Received error when requesting {url}: {resp.status}')
+					logging.verbose(await resp.read()) # change this to debug
+					return
+
+				if loads:
+					message = await resp.json(loads=loads)
+
+				elif resp.content_type == MIMETYPES['activity']:
+					message = await resp.json(loads=Message.new_from_json)
+
+				elif resp.content_type == MIMETYPES['json']:
+					message = await resp.json(loads=DotDict.new_from_json)
+
+				else:
+					# todo: raise TypeError or something
+					logging.verbose(f'Invalid Content-Type for "{url}": {resp.content_type}')
+					return logging.debug(f'Response: {resp.read()}')
+
+				logging.debug(f'{url} >> resp {message.to_json(4)}')
+
+				self.cache[url] = message
+				return message
+
+		except JSONDecodeError:
+			logging.verbose(f'Failed to parse JSON')
+
+		except ClientSSLError:
+			logging.verbose(f'SSL error when connecting to {urlparse(url).netloc}')
+
+		except (AsyncTimeoutError, ClientConnectionError):
+			logging.verbose(f'Failed to connect to {urlparse(url).netloc}')
+
+		except Exception as e:
+			traceback.print_exc()
+
+
+	async def post(self, url, message):
+		await self.open()
+
+		instance = self.database.get_inbox(url)
+
+		## Using the old algo by default is probably a better idea right now
+		if instance and instance.get('software') in {'mastodon'}:
+			algorithm = 'hs2019'
+
+		else:
+			algorithm = 'original'
+
+		headers = {'Content-Type': 'application/activity+json'}
+		headers.update(self.database.signer.sign_headers('POST', url, message, algorithm=algorithm))
+
+		try:
+			logging.verbose(f'Sending "{message.type}" to {url}')
+
+			async with self._session.post(url, headers=headers, data=message.to_json()) as resp:
+				## Not expecting a response, so just return
+				if resp.status in {200, 202}:
+					return logging.verbose(f'Successfully sent "{message.type}" to {url}')
+
+				logging.verbose(f'Received error when pushing to {url}: {resp.status}')
+				return logging.verbose(await resp.read()) # change this to debug
+
+		except ClientSSLError:
+			logging.warning(f'SSL error when pushing to {urlparse(url).netloc}')
+
+		except (AsyncTimeoutError, ClientConnectionError):
+			logging.warning(f'Failed to connect to {urlparse(url).netloc} for message push')
+
+		## prevent workers from being brought down
+		except Exception as e:
+			traceback.print_exc()
+
+
+	## Additional methods ##
+	async def fetch_nodeinfo(self, domain):
+		nodeinfo_url = None
+		wk_nodeinfo = await self.get(
+			f'https://{domain}/.well-known/nodeinfo',
+			loads = WellKnownNodeinfo.new_from_json
+		)
+
+		if not wk_nodeinfo:
+			logging.verbose(f'Failed to fetch well-known nodeinfo url for domain: {domain}')
+			return False
+
+		for version in ['20', '21']:
+			try:
+				nodeinfo_url = wk_nodeinfo.get_url(version)
+
+			except KeyError:
+				pass
+
+		if not nodeinfo_url:
+			logging.verbose(f'Failed to fetch nodeinfo url for domain: {domain}')
+			return False
+
+		return await self.get(nodeinfo_url, loads=Nodeinfo.new_from_json) or False
+
+
+async def get(database, *args, **kwargs):
+	async with HttpClient(database) as client:
+		return await client.get(*args, **kwargs)
+
+
+async def post(database, *args, **kwargs):
+	async with HttpClient(database) as client:
+		return await client.post(*args, **kwargs)
+
+
+async def fetch_nodeinfo(database, *args, **kwargs):
+	async with HttpClient(database) as client:
+		return await client.fetch_nodeinfo(*args, **kwargs)

relay/http_debug.py

@@ -1,68 +0,0 @@
-import logging
-import aiohttp
-
-from collections import defaultdict
-
-
-STATS = {
-    'requests': defaultdict(int),
-    'response_codes': defaultdict(int),
-    'response_codes_per_domain': defaultdict(lambda: defaultdict(int)),
-    'delivery_codes': defaultdict(int),
-    'delivery_codes_per_domain': defaultdict(lambda: defaultdict(int)),
-    'exceptions': defaultdict(int),
-    'exceptions_per_domain': defaultdict(lambda: defaultdict(int)),
-    'delivery_exceptions': defaultdict(int),
-    'delivery_exceptions_per_domain': defaultdict(lambda: defaultdict(int))
-}
-
-
-async def on_request_start(session, trace_config_ctx, params):
-    global STATS
-
-    logging.debug("HTTP START [%r], [%r]", session, params)
-
-    STATS['requests'][params.url.host] += 1
-
-
-async def on_request_end(session, trace_config_ctx, params):
-    global STATS
-
-    logging.debug("HTTP END [%r], [%r]", session, params)
-
-    host = params.url.host
-    status = params.response.status
-
-    STATS['response_codes'][status] += 1
-    STATS['response_codes_per_domain'][host][status] += 1
-
-    if params.method == 'POST':
-        STATS['delivery_codes'][status] += 1
-        STATS['delivery_codes_per_domain'][host][status] += 1
-
-
-async def on_request_exception(session, trace_config_ctx, params):
-    global STATS
-
-    logging.debug("HTTP EXCEPTION [%r], [%r]", session, params)
-
-    host = params.url.host
-    exception = repr(params.exception)
-
-    STATS['exceptions'][exception] += 1
-    STATS['exceptions_per_domain'][host][exception] += 1
-
-    if params.method == 'POST':
-        STATS['delivery_exceptions'][exception] += 1
-        STATS['delivery_exceptions_per_domain'][host][exception] += 1
-
-
-def http_debug():
-    if logging.DEBUG >= logging.root.level:
-        return
-
-    trace_config = aiohttp.TraceConfig()
-    trace_config.on_request_start.append(on_request_start)
-    trace_config.on_request_end.append(on_request_end)
-    trace_config.on_request_exception.append(on_request_exception)
-    return [trace_config]

relay/manage.py

@@ -1,17 +1,19 @@
 import Crypto
 import asyncio
 import click
-import json
 import logging
-import os
 import platform
 
-from aiohttp.web import AppRunner, TCPSite
-from cachetools import LRUCache
+from urllib.parse import urlparse
 
-from . import app, misc, views, __version__
-from .config import DotDict, RelayConfig, relay_software_names
-from .database import RelayDatabase
+from . import misc, __version__
+from . import http_client as http
+from .application import Application
+from .config import RELAY_SOFTWARE
+
+
+app = None
+CONFIG_IGNORE = {'blocked_software', 'blocked_instances', 'whitelist'}
 
 
 @click.group('cli', context_settings={'show_default': True}, invoke_without_command=True)
@@ -19,32 +21,100 @@ from .database import RelayDatabase
 @click.version_option(version=__version__, prog_name='ActivityRelay')
 @click.pass_context
 def cli(ctx, config):
-	app['is_docker'] = bool(os.environ.get('DOCKER_RUNNING'))
-	app['config'] = RelayConfig(config, app['is_docker'])
-
-	if not app['config'].load():
-		app['config'].save()
-
-	app['database'] = RelayDatabase(app['config'])
-	app['database'].load()
-
-	app['cache'] = DotDict()
-	app['semaphore'] = asyncio.Semaphore(app['config']['push_limit'])
-
-	for key in app['config'].cachekeys:
-		app['cache'][key] = LRUCache(app['config'][key])
+	global app
+	app = Application(config)
 
 	if not ctx.invoked_subcommand:
-		if app['config'].host.endswith('example.com'):
-			relay_setup.callback()
+		if app.config.host.endswith('example.com'):
+			cli_setup.callback()
 
 		else:
-			relay_run.callback()
+			cli_run.callback()
+
+
+@cli.command('setup')
+def cli_setup():
+	'Generate a new config'
+
+	while True:
+		app.config.host = click.prompt('What domain will the relay be hosted on?', default=app.config.host)
+
+		if not app.config.host.endswith('example.com'):
+			break
+
+		click.echo('The domain must not be example.com')
+
+	if not app.config.is_docker:
+		app.config.listen = click.prompt('Which address should the relay listen on?', default=app.config.listen)
+
+		while True:
+			app.config.port = click.prompt('What TCP port should the relay listen on?', default=app.config.port, type=int)
+			break
+
+	app.config.save()
+
+	if not app.config.is_docker and click.confirm('Relay all setup! Would you like to run it now?'):
+		cli_run.callback()
+
+
+@cli.command('run')
+def cli_run():
+	'Run the relay'
+
+	if app.config.host.endswith('example.com'):
+		return click.echo('Relay is not set up. Please edit your relay config or run "activityrelay setup".')
+
+	vers_split = platform.python_version().split('.')
+	pip_command = 'pip3 uninstall pycrypto && pip3 install pycryptodome'
+
+	if Crypto.__version__ == '2.6.1':
+		if int(vers_split[1]) > 7:
+			click.echo('Error: PyCrypto is broken on Python 3.8+. Please replace it with pycryptodome before running again. Exiting...')
+			return click.echo(pip_command)
+
+		else:
+			click.echo('Warning: PyCrypto is old and should be replaced with pycryptodome')
+			return click.echo(pip_command)
+
+	if not misc.check_open_port(app.config.listen, app.config.port):
+		return click.echo(f'Error: A server is already running on port {app.config.port}')
+
+	app.run()
+
+
+# todo: add config default command for resetting config key
+@cli.group('config')
+def cli_config():
+	'Manage the relay config'
+	pass
+
+
+@cli_config.command('list')
+def cli_config_list():
+	'List the current relay config'
+
+	click.echo('Relay Config:')
+
+	for key, value in app.config.items():
+		if key not in CONFIG_IGNORE:
+			key = f'{key}:'.ljust(20)
+			click.echo(f'- {key} {value}')
+
+
+@cli_config.command('set')
+@click.argument('key')
+@click.argument('value')
+def cli_config_set(key, value):
+	'Set a config value'
+
+	app.config[key] = value
+	app.config.save()
+
+	print(f'{key}: {app.config[key]}')
 
 
 @cli.group('inbox')
-@click.pass_context
-def cli_inbox(ctx):
+def cli_inbox():
 	'Manage the inboxes in the database'
 	pass
 
@@ -55,7 +125,7 @@ def cli_inbox_list():
 
 	click.echo('Connected to the following instances or relays:')
 
-	for inbox in app['database'].inboxes:
+	for inbox in app.database.inboxes:
 		click.echo(f'- {inbox}')
 
 
@@ -64,29 +134,34 @@ def cli_inbox_list():
 def cli_inbox_follow(actor):
 	'Follow an actor (Relay must be running)'
 
-	config = app['config']
-	database = app['database']
-
-	if config.is_banned(actor):
+	if app.config.is_banned(actor):
 		return click.echo(f'Error: Refusing to follow banned actor: {actor}')
 
 	if not actor.startswith('http'):
+		domain = actor
 		actor = f'https://{actor}/actor'
 
-	if database.get_inbox(actor):
-		return click.echo(f'Error: Already following actor: {actor}')
+	else:
+		domain = urlparse(actor).hostname
 
-	actor_data = run_in_loop(misc.request, actor, sign_headers=True)
+	try:
+		inbox_data = app.database['relay-list'][domain]
+		inbox = inbox_data['inbox']
 
-	if not actor_data:
-		return click.echo(f'Error: Failed to fetch actor: {actor}')
+	except KeyError:
+		actor_data = asyncio.run(http.get(app.database, actor, sign_headers=True))
 
-	inbox = misc.get_actor_inbox(actor_data)
+		if not actor_data:
+			return click.echo(f'Failed to fetch actor: {actor}')
 
-	database.add_inbox(inbox)
-	database.save()
+		inbox = actor_data.shared_inbox
 
-	run_in_loop(misc.follow_remote_actor, actor)
+	message = misc.Message.new_follow(
+		host = app.config.host,
+		actor = actor
+	)
+
+	asyncio.run(http.post(app.database, inbox, message))
 	click.echo(f'Sent follow message to actor: {actor}')
 
 
@@ -95,18 +170,36 @@ def cli_inbox_follow(actor):
 def cli_inbox_unfollow(actor):
 	'Unfollow an actor (Relay must be running)'
 
-	database = app['database']
-
 	if not actor.startswith('http'):
+		domain = actor
 		actor = f'https://{actor}/actor'
 
-	if not database.get_inbox(actor):
-		return click.echo(f'Error: Not following actor: {actor}')
+	else:
+		domain = urlparse(actor).hostname
 
-	database.del_inbox(actor)
-	database.save()
+	try:
+		inbox_data = app.database['relay-list'][domain]
+		inbox = inbox_data['inbox']
+		message = misc.Message.new_unfollow(
+			host = app.config.host,
+			actor = actor,
+			follow = inbox_data['followid']
+		)
 
-	run_in_loop(misc.unfollow_remote_actor, actor)
+	except KeyError:
+		actor_data = asyncio.run(http.get(app.database, actor, sign_headers=True))
+		inbox = actor_data.shared_inbox
+		message = misc.Message.new_unfollow(
+			host = app.config.host,
+			actor = actor,
+			follow = {
+				'type': 'Follow',
+				'object': actor,
+				'actor': f'https://{app.config.host}/actor'
+			}
+		)
+
+	asyncio.run(http.post(app.database, inbox, message))
 	click.echo(f'Sent unfollow message to: {actor}')
 
 
@@ -115,22 +208,18 @@ def cli_inbox_unfollow(actor):
 def cli_inbox_add(inbox):
 	'Add an inbox to the database'
 
-	database = app['database']
-	config = app['config']
-
 	if not inbox.startswith('http'):
 		inbox = f'https://{inbox}/inbox'
 
-	if database.get_inbox(inbox):
-		click.echo(f'Error: Inbox already in database: {inbox}')
-		return
+	if app.config.is_banned(inbox):
+		return click.echo(f'Error: Refusing to add banned inbox: {inbox}')
 
-	if config.is_banned(inbox):
-		click.echo(f'Error: Refusing to add banned inbox: {inbox}')
-		return
+	if app.database.get_inbox(inbox):
+		return click.echo(f'Error: Inbox already in database: {inbox}')
+
+	app.database.add_inbox(inbox)
+	app.database.save()
 
-	database.add_inbox(inbox)
-	database.save()
 	click.echo(f'Added inbox to the database: {inbox}')
 
 
@@ -139,15 +228,16 @@ def cli_inbox_add(inbox):
 def cli_inbox_remove(inbox):
 	'Remove an inbox from the database'
 
-	database = app['database']
-	dbinbox = database.get_inbox(inbox)
+	try:
+		dbinbox = app.database.get_inbox(inbox, fail=True)
 
-	if not dbinbox:
+	except KeyError:
 		click.echo(f'Error: Inbox does not exist: {inbox}')
 		return
 
-	database.del_inbox(dbinbox)
-	database.save()
+	app.database.del_inbox(dbinbox['domain'])
+	app.database.save()
+
 	click.echo(f'Removed inbox from the database: {inbox}')
 
 
@@ -163,7 +253,7 @@ def cli_instance_list():
 
 	click.echo('Banned instances or relays:')
 
-	for domain in app['config'].blocked_instances:
+	for domain in app.config.blocked_instances:
 		click.echo(f'- {domain}')
 
 
@@ -172,16 +262,14 @@ def cli_instance_list():
 def cli_instance_ban(target):
 	'Ban an instance and remove the associated inbox if it exists'
 
-	config = app['config']
-	database = app['database']
-	inbox = database.get_inbox(target)
+	if target.startswith('http'):
+		target = urlparse(target).hostname
 
-	if config.ban_instance(target):
-		config.save()
+	if app.config.ban_instance(target):
+		app.config.save()
 
-		if inbox:
-			database.del_inbox(inbox)
-			database.save()
+		if app.database.del_inbox(target):
+			app.database.save()
 
 		click.echo(f'Banned instance: {target}')
 		return
@@ -194,10 +282,8 @@ def cli_instance_ban(target):
 def cli_instance_unban(target):
 	'Unban an instance'
 
-	config = app['config']
-
-	if config.unban_instance(target):
-		config.save()
+	if app.config.unban_instance(target):
+		app.config.save()
 
 		click.echo(f'Unbanned instance: {target}')
 		return
@@ -217,7 +303,7 @@ def cli_software_list():
 
 	click.echo('Banned software:')
 
-	for software in app['config'].blocked_software:
+	for software in app.config.blocked_software:
 		click.echo(f'- {software}')
 
 
@@ -229,25 +315,23 @@ def cli_software_list():
 def cli_software_ban(name, fetch_nodeinfo):
 	'Ban software. Use RELAYS for NAME to ban relays'
 
-	config = app['config']
-
 	if name == 'RELAYS':
-		for name in relay_software_names:
-			config.ban_software(name)
+		for name in RELAY_SOFTWARE:
+			app.config.ban_software(name)
 
-		config.save()
+		app.config.save()
 		return click.echo('Banned all relay software')
 
 	if fetch_nodeinfo:
-		software = run_in_loop(fetch_nodeinfo, name)
+		nodeinfo = asyncio.run(http.fetch_nodeinfo(app.database, name))
 
-		if not software:
+		if not nodeinfo:
 			click.echo(f'Failed to fetch software name from domain: {name}')
 
-		name = software
+		name = nodeinfo.sw_name
 
-	if config.ban_software(name):
-		config.save()
+	if app.config.ban_software(name):
+		app.config.save()
 		return click.echo(f'Banned software: {name}')
 
 	click.echo(f'Software already banned: {name}')
@@ -261,31 +345,28 @@ def cli_software_ban(name, fetch_nodeinfo):
 def cli_software_unban(name, fetch_nodeinfo):
 	'Ban software. Use RELAYS for NAME to unban relays'
 
-	config = app['config']
-
 	if name == 'RELAYS':
-		for name in relay_software_names:
-			config.unban_software(name)
+		for name in RELAY_SOFTWARE:
+			app.config.unban_software(name)
 
-		config.save()
+		app.config.save()
 		return click.echo('Unbanned all relay software')
 
 	if fetch_nodeinfo:
-		software = run_in_loop(fetch_nodeinfo, name)
+		nodeinfo = asyncio.run(http.fetch_nodeinfo(app.database, name))
 
-		if not software:
+		if not nodeinfo:
 			click.echo(f'Failed to fetch software name from domain: {name}')
 
-		name = software
+		name = nodeinfo.sw_name
 
-	if config.unban_software(name):
-		config.save()
+	if app.config.unban_software(name):
+		app.config.save()
 		return click.echo(f'Unbanned software: {name}')
 
 	click.echo(f'Software wasn\'t banned: {name}')
 
 
-
 @cli.group('whitelist')
 def cli_whitelist():
 	'Manage the instance whitelist'
@@ -294,9 +375,11 @@ def cli_whitelist():
 
 @cli_whitelist.command('list')
 def cli_whitelist_list():
+	'List all the instances in the whitelist'
+
 	click.echo('Current whitelisted domains')
 
-	for domain in app['config'].whitelist:
+	for domain in app.config.whitelist:
 		click.echo(f'- {domain}')
 
 
@@ -305,12 +388,10 @@ def cli_whitelist_list():
 def cli_whitelist_add(instance):
 	'Add an instance to the whitelist'
 
-	config = app['config']
-
-	if not config.add_whitelist(instance):
+	if not app.config.add_whitelist(instance):
 		return click.echo(f'Instance already in the whitelist: {instance}')
 
-	config.save()
+	app.config.save()
 	click.echo(f'Instance added to the whitelist: {instance}')
 
 
@@ -319,106 +400,24 @@ def cli_whitelist_add(instance):
 def cli_whitelist_remove(instance):
 	'Remove an instance from the whitelist'
 
-	config = app['config']
-	database = app['database']
-	inbox = database.get_inbox(instance)
-
-	if not config.del_whitelist(instance):
+	if not app.config.del_whitelist(instance):
 		return click.echo(f'Instance not in the whitelist: {instance}')
 
-	config.save()
+	app.config.save()
 
-	if inbox and config.whitelist_enabled:
-		database.del_inbox(inbox)
-		database.save()
+	if app.config.whitelist_enabled:
+		if app.database.del_inbox(instance):
+			app.database.save()
 
 	click.echo(f'Removed instance from the whitelist: {instance}')
 
 
-@cli.command('setup')
-def relay_setup():
-	'Generate a new config'
+@cli_whitelist.command('import')
+def cli_whitelist_import():
+	'Add all current inboxes to the whitelist'
 
-	config = app['config']
-
-	while True:
-		config.host = click.prompt('What domain will the relay be hosted on?', default=config.host)
-
-		if not config.host.endswith('example.com'):
-			break
-
-		click.echo('The domain must not be example.com')
-
-	config.listen = click.prompt('Which address should the relay listen on?', default=config.listen)
-
-	while True:
-		config.port = click.prompt('What TCP port should the relay listen on?', default=config.port, type=int)
-		break
-
-	config.save()
-
-	if not app['is_docker'] and click.confirm('Relay all setup! Would you like to run it now?'):
-		relay_run.callback()
-
-
-@cli.command('run')
-def relay_run():
-	'Run the relay'
-
-	config = app['config']
-
-	if config.host.endswith('example.com'):
-		return click.echo('Relay is not set up. Please edit your relay config or run "activityrelay setup".')
-
-	vers_split = platform.python_version().split('.')
-	pip_command = 'pip3 uninstall pycrypto && pip3 install pycryptodome'
-
-	if Crypto.__version__ == '2.6.1':
-		if int(vers_split[1]) > 7:
-			click.echo('Error: PyCrypto is broken on Python 3.8+. Please replace it with pycryptodome before running again. Exiting...')
-			return click.echo(pip_command)
-
-		else:
-			click.echo('Warning: PyCrypto is old and should be replaced with pycryptodome')
-			return click.echo(pip_command)
-
-	if not misc.check_open_port(config.listen, config.port):
-		return click.echo(f'Error: A server is already running on port {config.port}')
-
-	# web pages
-	app.router.add_get('/', views.home)
-
-	# endpoints
-	app.router.add_post('/actor', views.inbox)
-	app.router.add_post('/inbox', views.inbox)
-	app.router.add_get('/actor', views.actor)
-	app.router.add_get('/nodeinfo/2.0.json', views.nodeinfo_2_0)
-	app.router.add_get('/.well-known/nodeinfo', views.nodeinfo_wellknown)
-	app.router.add_get('/.well-known/webfinger', views.webfinger)
-
-	if logging.DEBUG >= logging.root.level:
-		app.router.add_get('/stats', views.stats)
-
-	loop = asyncio.new_event_loop()
-	asyncio.set_event_loop(loop)
-	asyncio.ensure_future(handle_start_webserver(), loop=loop)
-	loop.run_forever()
-
-
-def run_in_loop(func, *args, **kwargs):
-	loop = asyncio.new_event_loop()
-	return loop.run_until_complete(func(*args, **kwargs))
-
-
-async def handle_start_webserver():
-	config = app['config']
-	runner = AppRunner(app, access_log_format='%{X-Forwarded-For}i "%r" %s %b "%{Referer}i" "%{User-Agent}i"')
-
-	logging.info(f'Starting webserver at {config.host} ({config.listen}:{config.port})')
-	await runner.setup()
-
-	site = TCPSite(runner, config.listen, config.port)
-	await site.start()
+	for domain in app.database.hostnames:
+		cli_whitelist_add.callback(domain)
 
 
 def main():

relay/misc.py

@@ -1,32 +1,69 @@
+import aputils
 import asyncio
 import base64
 import json
 import logging
 import socket
 import traceback
+import uuid
 
-from Crypto.Hash import SHA, SHA256, SHA512
-from Crypto.PublicKey import RSA
-from Crypto.Signature import PKCS1_v1_5
-from aiohttp import ClientSession
+from aiohttp.hdrs import METH_ALL as METHODS
+from aiohttp.web import Response as AiohttpResponse, View as AiohttpView
 from datetime import datetime
 from json.decoder import JSONDecodeError
 from urllib.parse import urlparse
 from uuid import uuid4
 
-from . import app
-from .http_debug import http_debug
 
+app = None
 
-HASHES = {
-	'sha1': SHA,
-	'sha256': SHA256,
-	'sha512': SHA512
+MIMETYPES = {
+	'activity': 'application/activity+json',
+	'html': 'text/html',
+	'json': 'application/json',
+	'text': 'text/plain'
+}
+
+NODEINFO_NS = {
+	'20': 'http://nodeinfo.diaspora.software/ns/schema/2.0',
+	'21': 'http://nodeinfo.diaspora.software/ns/schema/2.1'
 }
 
 
-def build_signing_string(headers, used_headers):
-	return '\n'.join(map(lambda x: ': '.join([x.lower(), headers[x]]), used_headers))
+def set_app(new_app):
+	global app
+	app = new_app
+
+
+def boolean(value):
+	if isinstance(value, str):
+		if value.lower() in ['on', 'y', 'yes', 'true', 'enable', 'enabled', '1']:
+			return True
+
+		elif value.lower() in ['off', 'n', 'no', 'false', 'disable', 'disable', '0']:
+			return False
+
+		else:
+			raise TypeError(f'Cannot parse string "{value}" as a boolean')
+
+	elif isinstance(value, int):
+		if value == 1:
+			return True
+
+		elif value == 0:
+			return False
+
+		else:
+			raise ValueError('Integer value must be 1 or 0')
+
+	elif value == None:
+		return False
+
+	try:
+		return value.__bool__()
+
+	except AttributeError:
+		raise TypeError(f'Cannot convert object of type "{clsname(value)}"')
 
 
 def check_open_port(host, port):
@@ -41,279 +78,263 @@ def check_open_port(host, port):
 			return False
 
 
-def create_signature_header(headers):
-	headers = {k.lower(): v for k, v in headers.items()}
-	used_headers = headers.keys()
-	sigstring = build_signing_string(headers, used_headers)
+class DotDict(dict):
+	def __init__(self, _data, **kwargs):
+		dict.__init__(self)
 
-	sig = {
-		'keyId': app['config'].keyid,
-		'algorithm': 'rsa-sha256',
-		'headers': ' '.join(used_headers),
-		'signature': sign_signing_string(sigstring, app['database'].PRIVKEY)
-	}
-
-	chunks = ['{}="{}"'.format(k, v) for k, v in sig.items()]
-	return ','.join(chunks)
+		self.update(_data, **kwargs)
 
 
-def distill_object_id(activity):
-	logging.debug(f'>> determining object ID for {activity["object"]}')
-
-	try:
-		return activity['object']['id']
-
-	except TypeError:
-		return activity['object']
-
-
-def distill_inboxes(actor, object_id):
-	database = app['database']
-	origin_hostname = urlparse(object_id).hostname
-	actor_inbox = get_actor_inbox(actor)
-	targets = []
-
-	for inbox in database.inboxes:
-		if inbox != actor_inbox or urlparse(inbox).hostname != origin_hostname:
-			targets.append(inbox)
-
-	return targets
-
-
-def generate_body_digest(body):
-	bodyhash = app['cache'].digests.get(body)
-
-	if bodyhash:
-		return bodyhash
-
-	h = SHA256.new(body.encode('utf-8'))
-	bodyhash = base64.b64encode(h.digest()).decode('utf-8')
-	app['cache'].digests[body] = bodyhash
-
-	return bodyhash
-
-
-def get_actor_inbox(actor):
-	return actor.get('endpoints', {}).get('sharedInbox', actor['inbox'])
-
-
-def sign_signing_string(sigstring, key):
-	pkcs = PKCS1_v1_5.new(key)
-	h = SHA256.new()
-	h.update(sigstring.encode('ascii'))
-	sigdata = pkcs.sign(h)
-
-	return base64.b64encode(sigdata).decode('utf-8')
-
-
-def split_signature(sig):
-	default = {"headers": "date"}
-
-	sig = sig.strip().split(',')
-
-	for chunk in sig:
-		k, _, v = chunk.partition('=')
-		v = v.strip('\"')
-		default[k] = v
-
-	default['headers'] = default['headers'].split()
-	return default
-
-
-async def fetch_actor_key(actor):
-	actor_data = await request(actor)
-
-	if not actor_data:
-		return None
-
-	try:
-		return RSA.importKey(actor_data['publicKey']['publicKeyPem'])
-
-	except Exception as e:
-		logging.debug(f'Exception occured while fetching actor key: {e}')
-
-
-async def fetch_nodeinfo(domain):
-	nodeinfo_url = None
-
-	wk_nodeinfo = await request(f'https://{domain}/.well-known/nodeinfo', sign_headers=False, activity=False)
-
-	if not wk_nodeinfo:
-		return
-
-	for link in wk_nodeinfo.get('links', ''):
-		if link['rel'] == 'http://nodeinfo.diaspora.software/ns/schema/2.0':
-			nodeinfo_url = link['href']
-			break
-
-	if not nodeinfo_url:
-		return
-
-	nodeinfo_data = await request(nodeinfo_url, sign_headers=False, activity=False)
-
-	try:
-		return nodeinfo_data['software']['name']
-
-	except KeyError:
-		return False
-
-
-async def follow_remote_actor(actor_uri):
-	config = app['config']
-
-	actor = await request(actor_uri)
-	inbox = get_actor_inbox(actor)
-
-	if not actor:
-		logging.error(f'failed to fetch actor at: {actor_uri}')
-		return
-
-	logging.verbose(f'sending follow request: {actor_uri}')
-
-	message = {
-		"@context": "https://www.w3.org/ns/activitystreams",
-		"type": "Follow",
-		"to": [actor['id']],
-		"object": actor['id'],
-		"id": f"https://{config.host}/activities/{uuid4()}",
-		"actor": f"https://{config.host}/actor"
-	}
-
-	await request(inbox, message)
-
-
-async def unfollow_remote_actor(actor_uri):
-	config = app['config']
-
-	actor = await request(actor_uri)
-
-	if not actor:
-		logging.error(f'failed to fetch actor: {actor_uri}')
-		return
-
-	inbox = get_actor_inbox(actor)
-	logging.verbose(f'sending unfollow request to inbox: {inbox}')
-
-	message = {
-		"@context": "https://www.w3.org/ns/activitystreams",
-		"type": "Undo",
-		"to": [actor_uri],
-		"object": {
-			"type": "Follow",
-			"object": actor_uri,
-			"actor": actor_uri,
-			"id": f"https://{config.host}/activities/{uuid4()}"
-		},
-		"id": f"https://{config.host}/activities/{uuid4()}",
-		"actor": f"https://{config.host}/actor"
-	}
-
-	await request(inbox, message)
-
-
-async def request(uri, data=None, force=False, sign_headers=True, activity=True):
-	## If a get request and not force, try to use the cache first
-	if not data and not force:
+	def __getattr__(self, k):
 		try:
-			return app['cache'].json[uri]
+			return self[k]
 
 		except KeyError:
-			pass
+			raise AttributeError(f'{self.__class__.__name__} object has no attribute {k}') from None
 
-	url = urlparse(uri)
-	method = 'POST' if data else 'GET'
-	headers = {'User-Agent': 'ActivityRelay'}
-	mimetype = 'application/activity+json' if activity else 'application/json'
 
-	## Set the content type for a POST
-	if data and 'Content-Type' not in headers:
-		headers['Content-Type'] = mimetype
+	def __setattr__(self, k, v):
+		if k.startswith('_'):
+			super().__setattr__(k, v)
 
-	## Set the accepted content type for a GET
-	elif not data and 'Accept' not in headers:
-		headers['Accept'] = mimetype
+		else:
+			self[k] = v
 
-	if sign_headers:
-		signing_headers = {
-			'(request-target)': f'{method.lower()} {url.path}',
-			'Date': datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT'),
-			'Host': url.netloc
+
+	def __setitem__(self, k, v):
+		if type(v) == dict:
+			v = DotDict(v)
+
+		super().__setitem__(k, v)
+
+
+	def __delattr__(self, k):
+		try:
+			dict.__delitem__(self, k)
+
+		except KeyError:
+			raise AttributeError(f'{self.__class__.__name__} object has no attribute {k}') from None
+
+
+	@classmethod
+	def new_from_json(cls, data):
+		if not data:
+			raise JSONDecodeError('Empty body', data, 1)
+
+		try:
+			return cls(json.loads(data))
+
+		except ValueError:
+			raise JSONDecodeError('Invalid body', data, 1)
+
+
+	@classmethod
+	def new_from_signature(cls, sig):
+		data = cls({})
+
+		for chunk in sig.strip().split(','):
+			key, value = chunk.split('=', 1)
+			value = value.strip('\"')
+
+			if key == 'headers':
+				value = value.split()
+
+			data[key.lower()] = value
+
+		return data
+
+
+	def to_json(self, indent=None):
+		return json.dumps(self, indent=indent)
+
+
+	def update(self, _data, **kwargs):
+		if isinstance(_data, dict):
+			for key, value in _data.items():
+				self[key] = value
+
+		elif isinstance(_data, (list, tuple, set)):
+			for key, value in _data:
+				self[key] = value
+
+		for key, value in kwargs.items():
+			self[key] = value
+
+
+class Message(DotDict):
+	@classmethod
+	def new_actor(cls, host, pubkey, description=None):
+		return cls({
+			'@context': 'https://www.w3.org/ns/activitystreams',
+			'id': f'https://{host}/actor',
+			'type': 'Application',
+			'preferredUsername': 'relay',
+			'name': 'ActivityRelay',
+			'summary': description or 'ActivityRelay bot',
+			'followers': f'https://{host}/followers',
+			'following': f'https://{host}/following',
+			'inbox': f'https://{host}/inbox',
+			'url': f'https://{host}/inbox',
+			'endpoints': {
+				'sharedInbox': f'https://{host}/inbox'
+			},
+			'publicKey': {
+				'id': f'https://{host}/actor#main-key',
+				'owner': f'https://{host}/actor',
+				'publicKeyPem': pubkey
+			}
+		})
+
+
+	@classmethod
+	def new_announce(cls, host, object):
+		return cls({
+			'@context': 'https://www.w3.org/ns/activitystreams',
+			'id': f'https://{host}/activities/{uuid.uuid4()}',
+			'type': 'Announce',
+			'to': [f'https://{host}/followers'],
+			'actor': f'https://{host}/actor',
+			'object': object
+		})
+
+
+	@classmethod
+	def new_follow(cls, host, actor):
+		return cls({
+			'@context': 'https://www.w3.org/ns/activitystreams',
+			'type': 'Follow',
+			'to': [actor],
+			'object': actor,
+			'id': f'https://{host}/activities/{uuid.uuid4()}',
+			'actor': f'https://{host}/actor'
+		})
+
+
+	@classmethod
+	def new_unfollow(cls, host, actor, follow):
+		return cls({
+			'@context': 'https://www.w3.org/ns/activitystreams',
+			'id': f'https://{host}/activities/{uuid.uuid4()}',
+			'type': 'Undo',
+			'to': [actor],
+			'actor': f'https://{host}/actor',
+			'object': follow
+		})
+
+
+	@classmethod
+	def new_response(cls, host, actor, followid, accept):
+		return cls({
+			'@context': 'https://www.w3.org/ns/activitystreams',
+			'id': f'https://{host}/activities/{uuid.uuid4()}',
+			'type': 'Accept' if accept else 'Reject',
+			'to': [actor],
+			'actor': f'https://{host}/actor',
+			'object': {
+				'id': followid,
+				'type': 'Follow',
+				'object': f'https://{host}/actor',
+				'actor': actor
+			}
+		})
+
+
+	# misc properties
+	@property
+	def domain(self):
+		return urlparse(self.id).hostname
+
+
+	# actor properties
+	@property
+	def shared_inbox(self):
+		return self.get('endpoints', {}).get('sharedInbox', self.inbox)
+
+
+	# activity properties
+	@property
+	def actorid(self):
+		if isinstance(self.actor, dict):
+			return self.actor.id
+
+		return self.actor
+
+
+	@property
+	def objectid(self):
+		if isinstance(self.object, dict):
+			return self.object.id
+
+		return self.object
+
+
+	@property
+	def signer(self):
+		return aputils.Signer.new_from_actor(self)
+
+
+class Response(AiohttpResponse):
+	@classmethod
+	def new(cls, body='', status=200, headers=None, ctype='text'):
+		kwargs = {
+			'status': status,
+			'headers': headers,
+			'content_type': MIMETYPES[ctype]
 		}
 
-		if data:
-			assert isinstance(data, dict)
+		if isinstance(body, bytes):
+			kwargs['body'] = body
 
-			action = data.get('type')
-			data = json.dumps(data)
-			signing_headers.update({
-				'Digest': f'SHA-256={generate_body_digest(data)}',
-				'Content-Length': str(len(data.encode('utf-8')))
-			})
+		elif isinstance(body, dict) and ctype in {'json', 'activity'}:
+			kwargs['text'] = json.dumps(body)
 
-		signing_headers['Signature'] = create_signature_header(signing_headers)
+		else:
+			kwargs['text'] = body
 
-		del signing_headers['(request-target)']
-		del signing_headers['Host']
-
-		headers.update(signing_headers)
-
-	try:
-		# json_serializer=DotDict maybe?
-		async with ClientSession(trace_configs=http_debug()) as session, app['semaphore']:
-			async with session.request(method, uri, headers=headers, data=data) as resp:
-				## aiohttp has been known to leak if the response hasn't been read,
-				## so we're just gonna read the request no matter what
-				resp_data = await resp.read()
-				resp_payload = json.loads(resp_data.decode('utf-8'))
-
-				if resp.status not in [200, 202]:
-					if not data:
-						logging.verbose(f'Received error when requesting {uri}: {resp.status} {resp_payload}')
-						return
-
-					logging.verbose(f'Received error when sending {action} to {uri}: {resp.status} {resp_payload}')
-					return
-
-				logging.debug(f'{uri} >> resp {resp_payload}')
-
-				app['cache'].json[uri] = resp_payload
-				return resp_payload
-
-	except JSONDecodeError:
-		return
-
-	except Exception:
-		traceback.print_exc()
+		return cls(**kwargs)
 
 
-async def validate_signature(actor, http_request):
-	pubkey = await fetch_actor_key(actor)
+	@classmethod
+	def new_error(cls, status, body, ctype='text'):
+		if ctype == 'json':
+			body = json.dumps({'status': status, 'error': body})
 
-	if not pubkey:
-		return False
+		return cls.new(body=body, status=status, ctype=ctype)
 
-	logging.debug(f'actor key: {pubkey}')
 
-	headers = {key.lower(): value for key, value in http_request.headers.items()}
-	headers['(request-target)'] = ' '.join([http_request.method.lower(), http_request.path])
+	@property
+	def location(self):
+		return self.headers.get('Location')
 
-	sig = split_signature(headers['signature'])
-	logging.debug(f'sigdata: {sig}')
 
-	sigstring = build_signing_string(headers, sig['headers'])
-	logging.debug(f'sigstring: {sigstring}')
+	@location.setter
+	def location(self, value):
+		self.headers['Location'] = value
 
-	sign_alg, _, hash_alg = sig['algorithm'].partition('-')
-	logging.debug(f'sign alg: {sign_alg}, hash alg: {hash_alg}')
 
-	sigdata = base64.b64decode(sig['signature'])
+class View(AiohttpView):
+	async def _iter(self):
+		if self.request.method not in METHODS:
+			self._raise_allowed_methods()
 
-	pkcs = PKCS1_v1_5.new(pubkey)
-	h = HASHES[hash_alg].new()
-	h.update(sigstring.encode('ascii'))
-	result = pkcs.verify(h, sigdata)
+		method = getattr(self, self.request.method.lower(), None)
 
-	http_request['validated'] = result
+		if method is None:
+			self._raise_allowed_methods()
 
-	logging.debug(f'validates? {result}')
-	return result
+		return await method(**self.request.match_info)
+
+
+	@property
+	def app(self):
+		return self._request.app
+
+
+	@property
+	def config(self):
+		return self.app.config
+
+
+	@property
+	def database(self):
+		return self.app.database

relay/processors.py

@@ -1,110 +1,138 @@
 import asyncio
 import logging
 
+from cachetools import LRUCache
 from uuid import uuid4
 
-from . import app, misc
+from .misc import Message
 
 
-async def handle_relay(actor, data, request):
-	cache = app['cache'].objects
-	object_id = misc.distill_object_id(data)
+cache = LRUCache(1024)
 
-	if object_id in cache:
-		logging.verbose(f'already relayed {object_id} as {cache[object_id]}')
+
+def person_check(actor, software):
+	## pleroma and akkoma may use Person for the actor type for some reason
+	if software in {'akkoma', 'pleroma'} and actor.id == f'https://{actor.domain}/relay':
+		return False
+
+	## make sure the actor is an application
+	if actor.type != 'Application':
+		return True
+
+
+async def handle_relay(request):
+	if request.message.objectid in cache:
+		logging.verbose(f'already relayed {request.message.objectid}')
 		return
 
-	logging.verbose(f'Relaying post from {actor["id"]}')
-
-	activity_id = f"https://{request.host}/activities/{uuid4()}"
-
-	message = {
-		"@context": "https://www.w3.org/ns/activitystreams",
-		"type": "Announce",
-		"to": [f"https://{request.host}/followers"],
-		"actor": f"https://{request.host}/actor",
-		"object": object_id,
-		"id": activity_id
-	}
+	message = Message.new_announce(
+		host = request.config.host,
+		object = request.message.objectid
+	)
 
+	cache[request.message.objectid] = message.id
 	logging.debug(f'>> relay: {message}')
 
-	inboxes = misc.distill_inboxes(actor, object_id)
-	futures = [misc.request(inbox, data=message) for inbox in inboxes]
+	inboxes = request.database.distill_inboxes(request.message)
 
-	asyncio.ensure_future(asyncio.gather(*futures))
-	cache[object_id] = activity_id
+	for inbox in inboxes:
+		request.app.push_message(inbox, message)
 
 
-async def handle_forward(actor, data, request):
-	cache = app['cache'].objects
-	object_id = misc.distill_object_id(data)
-
-	if object_id in cache:
-		logging.verbose(f'already forwarded {object_id}')
+async def handle_forward(request):
+	if request.message.id in cache:
+		logging.verbose(f'already forwarded {request.message.id}')
 		return
 
-	logging.verbose(f'Forwarding post from {actor["id"]}')
-	logging.debug(f'>> Relay {data}')
+	message = Message.new_announce(
+		host = request.config.host,
+		object = request.message
+	)
 
-	inboxes = misc.distill_inboxes(actor, object_id)
+	cache[request.message.id] = message.id
+	logging.debug(f'>> forward: {message}')
 
-	futures = [misc.request(inbox, data=data) for inbox in inboxes]
-	asyncio.ensure_future(asyncio.gather(*futures))
+	inboxes = request.database.distill_inboxes(request.message)
 
-	cache[object_id] = object_id
+	for inbox in inboxes:
+		request.app.push_message(inbox, message)
 
 
-async def handle_follow(actor, data, request):
-	config = app['config']
-	database = app['database']
+async def handle_follow(request):
+	nodeinfo = await request.app.client.fetch_nodeinfo(request.actor.domain)
+	software = nodeinfo.sw_name if nodeinfo else None
 
-	inbox = misc.get_actor_inbox(actor)
+	## reject if software used by actor is banned
+	if request.config.is_banned_software(software):
+		request.app.push_message(
+			request.actor.shared_inbox,
+			Message.new_response(
+				host = request.config.host,
+				actor = request.actor.id,
+				followid = request.message.id,
+				accept = False
+			)
+		)
 
-	if inbox not in database.inboxes:
-		database.add_inbox(inbox)
-		database.save()
-		asyncio.ensure_future(misc.follow_remote_actor(actor['id']))
+		return logging.verbose(f'Rejected follow from actor for using specific software: actor={request.actor.id}, software={software}')
 
-	message = {
-		"@context": "https://www.w3.org/ns/activitystreams",
-		"type": "Accept",
-		"to": [actor["id"]],
-		"actor": config.actor,
+	## reject if the actor is not an instance actor
+	if person_check(request.actor, software):
+		request.app.push_message(
+			request.actor.shared_inbox,
+			Message.new_response(
+				host = request.config.host,
+				actor = request.actor.id,
+				followid = request.message.id,
+				accept = False
+			)
+		)
 
-		# this is wrong per litepub, but mastodon < 2.4 is not compliant with that profile.
-		"object": {
-			"type": "Follow",
-			"id": data["id"],
-			"object": config.actor,
-			"actor": actor["id"]
-		},
+		return logging.verbose(f'Non-application actor tried to follow: {request.actor.id}')
 
-		"id": f"https://{request.host}/activities/{uuid4()}",
-	}
+	request.database.add_inbox(request.actor.shared_inbox, request.message.id, software)
+	request.database.save()
 
-	asyncio.ensure_future(misc.request(inbox, message))
+	request.app.push_message(
+		request.actor.shared_inbox,
+		Message.new_response(
+			host = request.config.host,
+			actor = request.actor.id,
+			followid = request.message.id,
+			accept = True
+		)
+	)
+
+	# Are Akkoma and Pleroma the only two that expect a follow back?
+	# Ignoring only Mastodon for now
+	if software != 'mastodon':
+		request.app.push_message(
+			request.actor.shared_inbox,
+			Message.new_follow(
+				host = request.config.host,
+				actor = request.actor.id
+			)
+		)
 
 
-async def handle_undo(actor, data, request):
-	## If the activity being undone is an Announce, forward it insteead
-	if data['object']['type'] == 'Announce':
-		await handle_forward(actor, data, request)
+async def handle_undo(request):
+	## If the object is not a Follow, forward it
+	if request.message.object.type != 'Follow':
+		return await handle_forward(request)
+
+	if not request.database.del_inbox(request.actor.domain, request.message.id):
 		return
 
-	elif data['object']['type'] != 'Follow':
-		return
+	request.database.save()
 
-	database = app['database']
-	inbox = database.get_inbox(actor['id'])
-
-	if not inbox:
-		return
-
-	database.del_inbox(inbox)
-	database.save()
-
-	await misc.unfollow_remote_actor(actor['id'])
+	request.app.push_message(
+		request.actor.shared_inbox,
+		Message.new_unfollow(
+			host = request.config.host,
+			actor = request.actor.id,
+			follow = request.message
+		)
+	)
 
 
 processors = {
@@ -117,9 +145,16 @@ processors = {
 }
 
 
-async def run_processor(request, data, actor):
-	if data['type'] not in processors:
+async def run_processor(request):
+	if request.message.type not in processors:
 		return
 
-	logging.verbose(f'New activity from actor: {actor["id"]} {data["type"]}')
-	return await processors[data['type']](actor, data, request)
+	if request.instance and not request.instance.get('software'):
+		nodeinfo = await request.app.client.fetch_nodeinfo(request.instance['domain'])
+
+		if nodeinfo:
+			request.instance['software'] = nodeinfo.sw_name
+			request.database.save()
+
+	logging.verbose(f'New "{request.message.type}" from actor: {request.actor.id}')
+	return await processors[request.message.type](request)

relay/views.py

@@ -1,28 +1,47 @@
+import aputils
+import asyncio
 import logging
 import subprocess
 import traceback
 
-from aiohttp.web import HTTPForbidden, HTTPUnauthorized, Response, json_response
-from urllib.parse import urlparse
+from pathlib import Path
 
-from . import __version__, app, misc
-from .http_debug import STATS
+from . import __version__, misc
+from .misc import DotDict, Message, Response
 from .processors import run_processor
 
 
-try:
-	commit_label = subprocess.check_output(["git", "rev-parse", "HEAD"]).strip().decode('ascii')
-	version = f'{__version__} {commit_label}'
-
-except:
-	version = __version__
+routes = []
+version = __version__
 
 
+if Path(__file__).parent.parent.joinpath('.git').exists():
+	try:
+		commit_label = subprocess.check_output(["git", "rev-parse", "HEAD"]).strip().decode('ascii')
+		version = f'{__version__} {commit_label}'
+
+	except:
+		pass
+
+
+def register_route(method, path):
+	def wrapper(func):
+		routes.append([method, path, func])
+		return func
+
+	return wrapper
+
+
+@register_route('GET', '/')
 async def home(request):
-	targets = '<br>'.join(app['database'].hostnames)
-	text = """
+	targets = '<br>'.join(request.database.hostnames)
+	note = request.config.note
+	count = len(request.database.hostnames)
+	host = request.config.host
+
+	text = f"""
 <html><head>
-<title>ActivityPub Relay at {host}</title>
+<title>SEDI中繼器</title>
 <style>
 p {{ color: #FFFFFF; font-family: monospace, arial; font-size: 100%; }}
 body {{ background-color: #000000; }}
@@ -34,170 +53,142 @@ a:hover {{ color: #8AF; }}
 <body>
 <p>This is an Activity Relay for fediverse instances.</p>
 <p>{note}</p>
-<p>You may subscribe to this relay with the address: <a href="https://{host}/actor">https://{host}/actor</a></p>
-<p>To host your own relay, you may download the code at this address: <a href="https://git.pleroma.social/pleroma/relay">https://git.pleroma.social/pleroma/relay</a></p>
-<br><p>List of {count} registered instances:<br>{targets}</p>
-</body></html>""".format(host=request.host, note=app['config'].note, targets=targets, count=len(app['database'].inboxes))
+<p>Misskey及Mastodon站長請訂閱這個地址：<a href="https://{host}/inbox">https://{host}/inbox</a></p>
+<p>Pleroma及Friendica站長請訂閱這個地址：<a href="https://{host}/actor">https://{host}/actor</a></p>
+<p>原始碼：<a href="https://git.seediqbale.xyz/pch_xyz/sedi-relay">https://git.seediqbale.xyz/pch_xyz/sedi-relay</a></p>
+<p>請我喝杯咖啡：<a href="https://buymeacoffee.com/SEDI">https://buymeacoffee.com/SEDI</a></p>
+<p>activityrelay v0.2.4</p>
+<br><p>有 {count} 個實例訂閱中：<br>{targets}</p>
+</body></html>"""
 
-	return Response(
-		status = 200,
-		content_type = 'text/html',
-		charset = 'utf-8',
-		text = text
+	return Response.new(text, ctype='html')
+
+
+@register_route('GET', '/inbox')
+@register_route('GET', '/actor')
+async def actor(request):
+	data = Message.new_actor(
+		host = request.config.host, 
+		pubkey = request.database.signer.pubkey
 	)
 
-
-async def actor(request):
-	database = app['database']
-
-	data = {
-		"@context": "https://www.w3.org/ns/activitystreams",
-		"endpoints": {
-			"sharedInbox": f"https://{request.host}/inbox"
-		},
-		"followers": f"https://{request.host}/followers",
-		"following": f"https://{request.host}/following",
-		"inbox": f"https://{request.host}/inbox",
-		"name": "ActivityRelay",
-		"type": "Application",
-		"id": f"https://{request.host}/actor",
-		"publicKey": {
-			"id": f"https://{request.host}/actor#main-key",
-			"owner": f"https://{request.host}/actor",
-			"publicKeyPem": database.pubkey
-		},
-		"summary": "ActivityRelay bot",
-		"preferredUsername": "relay",
-		"url": f"https://{request.host}/actor"
-	}
-
-	return json_response(data, content_type='application/activity+json')
+	return Response.new(data, ctype='activity')
 
 
+@register_route('POST', '/inbox')
+@register_route('POST', '/actor')
 async def inbox(request):
-	config = app['config']
-	database = app['database']
+	config = request.config
+	database = request.database
 
 	## reject if missing signature header
-	if 'signature' not in request.headers:
+	if not request.signature:
 		logging.verbose('Actor missing signature header')
 		raise HTTPUnauthorized(body='missing signature')
 
-	## read message and get actor id and domain
 	try:
-		data = await request.json()
-		actor_id = data['actor']
-		actor_domain = urlparse(actor_id).hostname
+		request['message'] = await request.json(loads=Message.new_from_json)
 
-	except KeyError:
-		logging.verbose('actor not in data')
-		raise HTTPUnauthorized(body='no actor in message')
+		## reject if there is no message
+		if not request.message:
+			logging.verbose('empty message')
+			return Response.new_error(400, 'missing message', 'json')
+
+		## reject if there is no actor in the message
+		if 'actor' not in request.message:
+			logging.verbose('actor not in message')
+			return Response.new_error(400, 'no actor in message', 'json')
 
-	## reject if there is no actor in the message
 	except:
+		## this code should hopefully never get called
 		traceback.print_exc()
 		logging.verbose('Failed to parse inbox message')
-		raise HTTPUnauthorized(body='failed to parse message')
+		return Response.new_error(400, 'failed to parse message', 'json')
 
-	actor = await misc.request(actor_id)
+	request['actor'] = await request.app.client.get(request.signature.keyid, sign_headers=True)
 
 	## reject if actor is empty
-	if not actor:
-		logging.verbose(f'Failed to fetch actor: {actor_id}')
-		raise HTTPUnauthorized('failed to fetch actor')
+	if not request.actor:
+		## ld signatures aren't handled atm, so just ignore it
+		if request['message'].type == 'Delete':
+			logging.verbose(f'Instance sent a delete which cannot be handled')
+			return Response.new(status=202)
+
+		logging.verbose(f'Failed to fetch actor: {request.signature.keyid}')
+		return Response.new_error(400, 'failed to fetch actor', 'json')
+
+	request['instance'] = request.database.get_inbox(request['actor'].inbox)
 
 	## reject if the actor isn't whitelisted while the whiltelist is enabled
-	elif config.whitelist_enabled and not config.is_whitelisted(actor_id):
-		logging.verbose(f'Rejected actor for not being in the whitelist: {actor_id}')
-		raise HTTPForbidden(body='access denied')
+	if config.whitelist_enabled and not config.is_whitelisted(request.actor.domain):
+		logging.verbose(f'Rejected actor for not being in the whitelist: {request.actor.id}')
+		return Response.new_error(403, 'access denied', 'json')
 
 	## reject if actor is banned
-	if app['config'].is_banned(actor_id):
-		logging.verbose(f'Ignored request from banned actor: {actor_id}')
-		raise HTTPForbidden(body='access denied')
-
-	## reject if software used by actor is banned
-	if len(config.blocked_software):
-		software = await misc.fetch_nodeinfo(actor_domain)
-
-		if config.is_banned_software(software):
-			logging.verbose(f'Rejected actor for using specific software: {software}')
-			raise HTTPForbidden(body='access denied')
+	if request.config.is_banned(request.actor.domain):
+		logging.verbose(f'Ignored request from banned actor: {actor.id}')
+		return Response.new_error(403, 'access denied', 'json')
 
 	## reject if the signature is invalid
-	if not (await misc.validate_signature(actor_id, request)):
-		logging.verbose(f'signature validation failed for: {actor_id}')
-		raise HTTPUnauthorized(body='signature check failed, signature did not match key')
+	try:
+		await request.actor.signer.validate_aiohttp_request(request)
+
+	except aputils.SignatureValidationError as e:
+		logging.verbose(f'signature validation failed for: {actor.id}')
+		logging.debug(str(e))
+		return Response.new_error(401, str(e), 'json')
 
 	## reject if activity type isn't 'Follow' and the actor isn't following
-	if data['type'] != 'Follow' and not database.get_inbox(actor_domain):
-		logging.verbose(f'Rejected actor for trying to post while not following: {actor_id}')
-		raise HTTPUnauthorized(body='access denied')
+	if request.message.type != 'Follow' and not database.get_inbox(request.actor.domain):
+		logging.verbose(f'Rejected actor for trying to post while not following: {request.actor.id}')
+		return Response.new_error(401, 'access denied', 'json')
 
-	logging.debug(f">> payload {data}")
+	logging.debug(f">> payload {request.message.to_json(4)}")
 
-	await run_processor(request, data, actor)
-	return Response(body=b'{}', content_type='application/activity+json')
+	asyncio.ensure_future(run_processor(request))
+	return Response.new(status=202)
 
 
+@register_route('GET', '/.well-known/webfinger')
 async def webfinger(request):
-	config = app['config']
-	subject = request.query['resource']
+	try:
+		subject = request.query['resource']
 
-	if subject != f'acct:relay@{request.host}':
-		return json_response({'error': 'user not found'}, status=404)
+	except KeyError:
+		return Response.new_error(400, 'missing \'resource\' query key', 'json')
 
-	data = {
-		'subject': subject,
-		'aliases': [config.actor],
-		'links': [
-			{'href': config.actor, 'rel': 'self', 'type': 'application/activity+json'},
-			{'href': config.actor, 'rel': 'self', 'type': 'application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"'}
-		]
-	}
+	if subject != f'acct:relay@{request.config.host}':
+		return Response.new_error(404, 'user not found', 'json')
 
-	return json_response(data)
+	data = aputils.Webfinger.new(
+		handle = 'relay',
+		domain = request.config.host,
+		actor = request.config.actor
+	)
+
+	return Response.new(data, ctype='json')
 
 
-async def nodeinfo_2_0(request):
-	data = {
-		# XXX - is this valid for a relay?
-		'openRegistrations': True,
-		'protocols': ['activitypub'],
-		'services': {
-			'inbound': [],
-			'outbound': []
-		},
-		'software': {
-			'name': 'activityrelay',
-			'version': version
-		},
-		'usage': {
-			'localPosts': 0,
-			'users': {
-				'total': 1
-			}
-		},
-		'metadata': {
-			'peers': app['database'].hostnames
-		},
-		'version': '2.0'
-	}
+@register_route('GET', '/nodeinfo/{version:\d.\d\.json}')
+async def nodeinfo(request):
+	niversion = request.match_info['version'][:3]
 
-	return json_response(data)
+	data = dict(
+		name = 'activityrelay',
+		version = version,
+		protocols = ['activitypub'],
+		open_regs = not request.config.whitelist_enabled,
+		users = 1,
+		metadata = {'peers': request.database.hostnames}
+	)
+
+	if niversion == '2.1':
+		data['repo'] = 'https://git.pleroma.social/pleroma/relay'
+
+	return Response.new(aputils.Nodeinfo.new(**data), ctype='json')
 
 
+@register_route('GET', '/.well-known/nodeinfo')
 async def nodeinfo_wellknown(request):
-	data = {
-		'links': [
-			{
-				'rel': 'http://nodeinfo.diaspora.software/ns/schema/2.0',
-				'href': f'https://{request.host}/nodeinfo/2.0.json'
-			}
-		]
-	}
-	return json_response(data)
-
-
-async def stats(request):
-    return json_response(STATS)
+	data = aputils.WellKnownNodeinfo.new_template(request.config.host)
+	return Response.new(data, ctype='json')

requirements.txt

@@ -1 +1,5 @@
-.
+aiohttp>=3.8.0
+aputils@https://git.barkshark.xyz/barkshark/aputils/archive/0.1.3.tar.gz
+cachetools>=5.2.0
+click>=8.1.2
+pyyaml>=6.0

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = relay
-version = 0.2.2
+version = attr: relay.__version__
 description = Generic LitePub relay (works with all LitePub consumers and Mastodon)
 long_description = file: README.md
 long_description_content_type = text/markdown; charset=UTF-8
@@ -22,13 +22,12 @@ project_urls =
 [options]
 zip_safe = False
 packages = find:
-install_requires =
-    aiohttp >= 3.8.0
-    cachetools >= 5.0.0
-    click >= 8.1.2
-    pycryptodome >= 3.14.1
-    PyYAML >= 5.0.0
-python_requires = >=3.6
+install_requires = file: requirements.txt
+python_requires = >=3.7
+
+[options.extras_require]
+dev =
+	pyinstaller >= 5.6.0
 
 [options.entry_points]
 console_scripts =