create Application class

relay/__init__.py

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

relay/application.py

@@ -0,0 +1,119 @@
+import asyncio
+import logging
+import os
+import signal
+
+from aiohttp import web
+from cachetools import LRUCache
+from datetime import datetime, timedelta
+
+from .config import RelayConfig
+from .database import RelayDatabase
+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['is_docker'] = bool(os.environ.get('DOCKER_RUNNING'))
+		self['config'] = RelayConfig(cfgpath, self['is_docker'])
+
+		if not self['config'].load():
+			self['config'].save()
+
+		self['database'] = RelayDatabase(self['config'])
+		self['database'].load()
+
+		self['cache'] = DotDict({key: LRUCache(maxsize=self['config'][key]) for key in self['config'].cachekeys})
+		self['semaphore'] = asyncio.Semaphore(self['config'].push_limit)
+
+		self.set_signal_handler()
+		set_app(self)
+
+
+	@property
+	def cache(self):
+		return self['cache']
+
+
+	@property
+	def config(self):
+		return self['config']
+
+
+	@property
+	def database(self):
+		return self['database']
+
+
+	@property
+	def is_docker(self):
+		return self['is_docker']
+
+
+	@property
+	def semaphore(self):
+		return self['semaphore']
+
+
+	@property
+	def uptime(self):
+		if not self['starttime']:
+			return timedelta(seconds=0)
+
+		uptime = datetime.now() - self['starttime']
+
+		return timedelta(seconds=uptime.seconds)
+
+
+	def set_signal_handler(self):
+		signal.signal(signal.SIGHUP, self.stop)
+		signal.signal(signal.SIGINT, self.stop)
+		signal.signal(signal.SIGQUIT, self.stop)
+		signal.signal(signal.SIGTERM, self.stop)
+
+
+	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:
+			if route[1] == '/stats' and logging.DEBUG < logging.root.level:
+				continue
+
+			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
+
+		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

relay/manage.py

@@ -1,17 +1,15 @@
 import Crypto
 import asyncio
 import click
-import json
 import logging
-import os
 import platform
 
-from aiohttp.web import AppRunner, TCPSite
+from . import misc, __version__
-from cachetools import LRUCache
+from .application import Application
+from .config import relay_software_names
 
-from . import app, misc, views, __version__
+
-from .config import DotDict, RelayConfig, relay_software_names
+app = None
-from .database import RelayDatabase
 
 
 @click.group('cli', context_settings={'show_default': True}, invoke_without_command=True)
@@ -19,23 +17,11 @@ 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'))
+	global app
-	app['config'] = RelayConfig(config, app['is_docker'])
+	app = Application(config)
-
-	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])
 
 	if not ctx.invoked_subcommand:
-		if app['config'].host.endswith('example.com'):
+		if app.config.host.endswith('example.com'):
 			relay_setup.callback()
 
 		else:
@@ -55,7 +41,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,16 +50,13 @@ def cli_inbox_list():
 def cli_inbox_follow(actor):
 	'Follow an actor (Relay must be running)'
 
-	config = app['config']
+	if app.config.is_banned(actor):
-	database = app['database']
-
-	if config.is_banned(actor):
 		return click.echo(f'Error: Refusing to follow banned actor: {actor}')
 
 	if not actor.startswith('http'):
 		actor = f'https://{actor}/actor'
 
-	if database.get_inbox(actor):
+	if app.database.get_inbox(actor):
 		return click.echo(f'Error: Already following actor: {actor}')
 
 	actor_data = asyncio.run(misc.request(actor, sign_headers=True))
@@ -81,8 +64,8 @@ def cli_inbox_follow(actor):
 	if not actor_data:
 		return click.echo(f'Error: Failed to fetch actor: {actor}')
 
-	database.add_inbox(actor_data.shared_inbox)
+	app.database.add_inbox(actor_data.shared_inbox)
-	database.save()
+	app.database.save()
 
 	asyncio.run(misc.follow_remote_actor(actor))
 	click.echo(f'Sent follow message to actor: {actor}')
@@ -93,13 +76,11 @@ 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'):
 		actor = f'https://{actor}/actor'
 
-	if database.del_inbox(actor):
+	if app.database.del_inbox(actor):
-		database.save()
+		app.database.save()
 		asyncio.run(misc.unfollow_remote_actor(actor))
 		return click.echo(f'Sent unfollow message to: {actor}')
 
@@ -111,17 +92,14 @@ 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 config.is_banned(inbox):
+	if app.config.is_banned(inbox):
 		return click.echo(f'Error: Refusing to add banned inbox: {inbox}')
 
-	if database.add_inbox(inbox):
+	if app.database.add_inbox(inbox):
-		database.save()
+		app.database.save()
 		return click.echo(f'Added inbox to the database: {inbox}')
 
 	click.echo(f'Error: Inbox already in database: {inbox}')
@@ -132,17 +110,15 @@ def cli_inbox_add(inbox):
 def cli_inbox_remove(inbox):
 	'Remove an inbox from the database'
 
-	database = app['database']
-
 	try:
-		dbinbox = database.get_inbox(inbox, fail=True)
+		dbinbox = app.database.get_inbox(inbox, fail=True)
 
 	except KeyError:
 		click.echo(f'Error: Inbox does not exist: {inbox}')
 		return
 
-	database.del_inbox(dbinbox['domain'])
+	app.database.del_inbox(dbinbox['domain'])
-	database.save()
+	app.database.save()
 
 	click.echo(f'Removed inbox from the database: {inbox}')
 
@@ -159,7 +135,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}')
 
 
@@ -168,17 +144,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']
-
 	if target.startswith('http'):
 		target = urlparse(target).hostname
 
-	if config.ban_instance(target):
+	if app.config.ban_instance(target):
-		config.save()
+		app.config.save()
 
-		if database.del_inbox(target):
+		if app.database.del_inbox(target):
-			database.save()
+			app.database.save()
 
 		click.echo(f'Banned instance: {target}')
 		return
@@ -191,10 +164,8 @@ def cli_instance_ban(target):
 def cli_instance_unban(target):
 	'Unban an instance'
 
-	config = app['config']
+	if app.config.unban_instance(target):
-
+		app.config.save()
-	if config.unban_instance(target):
-		config.save()
 
 		click.echo(f'Unbanned instance: {target}')
 		return
@@ -214,7 +185,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}')
 
 
@@ -226,13 +197,11 @@ 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)
+			app.config.ban_software(name)
 
-		config.save()
+		app.config.save()
 		return click.echo('Banned all relay software')
 
 	if fetch_nodeinfo:
@@ -244,7 +213,7 @@ def cli_software_ban(name, fetch_nodeinfo):
 		name = software
 
 	if config.ban_software(name):
-		config.save()
+		app.config.save()
 		return click.echo(f'Banned software: {name}')
 
 	click.echo(f'Software already banned: {name}')
@@ -258,11 +227,9 @@ 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)
+			app.config.unban_software(name)
 
 		config.save()
 		return click.echo('Unbanned all relay software')
@@ -275,8 +242,8 @@ def cli_software_unban(name, fetch_nodeinfo):
 
 		name = software
 
-	if config.unban_software(name):
+	if app.config.unban_software(name):
-		config.save()
+		app.config.save()
 		return click.echo(f'Unbanned software: {name}')
 
 	click.echo(f'Software wasn\'t banned: {name}')
@@ -293,7 +260,7 @@ def cli_whitelist():
 def cli_whitelist_list():
 	click.echo('Current whitelisted domains')
 
-	for domain in app['config'].whitelist:
+	for domain in app.config.whitelist:
 		click.echo(f'- {domain}')
 
 
@@ -302,12 +269,10 @@ def cli_whitelist_list():
 def cli_whitelist_add(instance):
 	'Add an instance to the whitelist'
 
-	config = app['config']
+	if not app.config.add_whitelist(instance):
-
-	if not 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}')
 
 
@@ -316,17 +281,14 @@ def cli_whitelist_add(instance):
 def cli_whitelist_remove(instance):
 	'Remove an instance from the whitelist'
 
-	config = app['config']
+	if not app.config.del_whitelist(instance):
-	database = app['database']
-
-	if not config.del_whitelist(instance):
 		return click.echo(f'Instance not in the whitelist: {instance}')
 
-	config.save()
+	app.config.save()
 
-	if config.whitelist_enabled:
+	if app.config.whitelist_enabled:
-		if database.del_inbox(inbox):
+		if app.database.del_inbox(inbox):
-			database.save()
+			app.database.save()
 
 	click.echo(f'Removed instance from the whitelist: {instance}')
 
@@ -335,23 +297,21 @@ def cli_whitelist_remove(instance):
 def relay_setup():
 	'Generate a new config'
 
-	config = app['config']
-
 	while True:
-		config.host = click.prompt('What domain will the relay be hosted on?', default=config.host)
+		app.config.host = click.prompt('What domain will the relay be hosted on?', default=app.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)
+	app.config.listen = click.prompt('Which address should the relay listen on?', default=app.config.listen)
 
 	while True:
-		config.port = click.prompt('What TCP port should the relay listen on?', default=config.port, type=int)
+		app.config.port = click.prompt('What TCP port should the relay listen on?', default=app.config.port, type=int)
 		break
 
-	config.save()
+	app.config.save()
 
 	if not app['is_docker'] and click.confirm('Relay all setup! Would you like to run it now?'):
 		relay_run.callback()
@@ -361,9 +321,7 @@ def relay_setup():
 def relay_run():
 	'Run the relay'
 
-	config = app['config']
+	if app.config.host.endswith('example.com'):
-
-	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('.')
@@ -378,38 +336,10 @@ def relay_run():
 			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):
+	if not misc.check_open_port(app.config.listen, app.config.port):
-		return click.echo(f'Error: A server is already running on port {config.port}')
+		return click.echo(f'Error: A server is already running on port {app.config.port}')
 
-	# web pages
+	app.run()
-	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()
-
-
-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()
 
 
 def main():

relay/misc.py

@@ -15,10 +15,10 @@ 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,
@@ -26,6 +26,11 @@ HASHES = {
 }
 
 
+def set_app(new_app):
+	global app
+	app = new_app
+
+
 def build_signing_string(headers, used_headers):
 	return '\n'.join(map(lambda x: ': '.join([x.lower(), headers[x]]), used_headers))
 
@@ -62,7 +67,7 @@ def distill_inboxes(actor, object_id):
 	database = app['database']
 
 	for inbox in database.inboxes:
-		if inbox != actor.shared_inbox or urlparse(inbox).hostname != urlparse(object_id).hostname:
+		if inbox != actor.shared_inbox and urlparse(inbox).hostname != urlparse(object_id).hostname:
 			yield inbox
 
 

relay/processors.py

@@ -3,21 +3,18 @@ import logging
 
 from uuid import uuid4
 
-from . import app, misc
+from . import misc
 
 
 async def handle_relay(request, actor, data, software):
-	cache = app['cache'].objects
+	if data.objectid in request.app.cache.objects:
-	config = app['config']
+		logging.verbose(f'already relayed {data.objectid}')
-
-	if data.objectid in cache:
-		logging.verbose(f'already relayed {data.objectid} as {cache[data.objectid]}')
 		return
 
 	logging.verbose(f'Relaying post from {data.actorid}')
 
 	message = misc.Message.new_announce(
-		host = config.host,
+		host = request.app.config.host,
 		object = data.objectid
 	)
 
@@ -27,19 +24,16 @@ async def handle_relay(request, actor, data, software):
 	futures = [misc.request(inbox, data=message) for inbox in inboxes]
 
 	asyncio.ensure_future(asyncio.gather(*futures))
-	cache[data.objectid] = message.id
+	request.app.cache.objects[data.objectid] = message.id
 
 
 async def handle_forward(request, actor, data, software):
-	cache = app['cache'].objects
+	if data.id in request.app.cache.objects:
-	config = app['config']
-
-	if data.id in cache:
 		logging.verbose(f'already forwarded {data.id}')
 		return
 
 	message = misc.Message.new_announce(
-		host = config.host,
+		host = request.app.config.host,
 		object = data
 	)
 
@@ -50,22 +44,19 @@ async def handle_forward(request, actor, data, software):
 	futures = [misc.request(inbox, data=message) for inbox in inboxes]
 
 	asyncio.ensure_future(asyncio.gather(*futures))
-	cache[data.id] = message.id
+	request.app.cache.objects[data.id] = message.id
 
 
 async def handle_follow(request, actor, data, software):
-	config = app['config']
+	if request.app.database.add_inbox(inbox, data.id):
-	database = app['database']
+		request.app.database.set_followid(actor.id, data.id)
 
-	if database.add_inbox(inbox, data.id):
+	request.app.database.save()
-		database.set_followid(actor.id, data.id)
-
-	database.save()
 
 	await misc.request(
 		actor.shared_inbox,
 		misc.Message.new_response(
-			host = config.host,
+			host = request.app.config.host,
 			actor = actor.id,
 			followid = data.id,
 			accept = True
@@ -78,7 +69,7 @@ async def handle_follow(request, actor, data, software):
 		misc.request(
 			actor.shared_inbox,
 			misc.Message.new_follow(
-				host = config.host,
+				host = request.app.config.host,
 				actor = actor.id
 			)
 		)
@@ -89,15 +80,13 @@ async def handle_undo(request, actor, data, software):
 	if data['object']['type'] != 'Follow':
 		return await handle_forward(request, actor, data, software)
 
-	database = app['database']
+	if not request.app.database.del_inbox(actor.domain, data.id):
-
-	if not database.del_inbox(actor.domain, data.id):
 		return
 
-	database.save()
+	request.app.database.save()
 
 	message = misc.Message.new_unfollow(
-		host = config.host,
+		host = request.app.config.host,
 		actor = actor.id,
 		follow = data
 	)

relay/views.py

@@ -2,14 +2,25 @@ import logging
 import subprocess
 import traceback
 
-from aiohttp.web import HTTPForbidden, HTTPUnauthorized, Response, json_response
+from aiohttp.web import HTTPForbidden, HTTPUnauthorized, Response, json_response, route
 
-from . import __version__, app, misc
+from . import __version__, misc
 from .http_debug import STATS
 from .misc import Message
 from .processors import run_processor
 
 
+routes = []
+
+
+def register_route(method, path):
+	def wrapper(func):
+		routes.append([method, path, func])
+		return func
+
+	return wrapper
+
+
 try:
 	commit_label = subprocess.check_output(["git", "rev-parse", "HEAD"]).strip().decode('ascii')
 	version = f'{__version__} {commit_label}'
@@ -18,9 +29,14 @@ except:
 	version = __version__
 
 
+@register_route('GET', '/')
 async def home(request):
-	targets = '<br>'.join(app['database'].hostnames)
+	targets = '<br>'.join(request.app.database.hostnames)
-	text = """
+	note = request.app.config.note
+	count = len(request.app.database.hostnames)
+	host = request.app.config.host
+
+	text = f"""
 <html><head>
 <title>ActivityPub Relay at {host}</title>
 <style>
@@ -37,7 +53,7 @@ a:hover {{ color: #8AF; }}
 <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'].hostnames))
+</body></html>"""
 
 	return Response(
 		status = 200,
@@ -47,21 +63,22 @@ a:hover {{ color: #8AF; }}
 	)
 
 
+@register_route('GET', '/inbox')
+@register_route('GET', '/actor')
 async def actor(request):
-	config = app['config']
-	database = app['database']
-
 	data = Message.new_actor(
-		host = config.host, 
+		host = request.app.config.host, 
-		pubkey = database.pubkey
+		pubkey = request.app.database.pubkey
 	)
 
 	return json_response(data, content_type='application/activity+json')
 
 
+@register_route('POST', '/inbox')
+@register_route('POST', '/actor')
 async def inbox(request):
-	config = app['config']
+	config = request.app.config
-	database = app['database']
+	database = request.app.database
 
 	## reject if missing signature header
 	if 'signature' not in request.headers:
@@ -98,7 +115,7 @@ async def inbox(request):
 		raise HTTPForbidden(body='access denied')
 
 	## reject if actor is banned
-	if app['config'].is_banned(data.domain):
+	if request.app['config'].is_banned(data.domain):
 		logging.verbose(f'Ignored request from banned actor: {data.actorid}')
 		raise HTTPForbidden(body='access denied')
 
@@ -126,25 +143,26 @@ async def inbox(request):
 	return Response(body=b'{}', content_type='application/activity+json')
 
 
+@register_route('GET', '/.well-known/webfinger')
 async def webfinger(request):
-	config = app['config']
 	subject = request.query['resource']
 
-	if subject != f'acct:relay@{request.host}':
+	if subject != f'acct:relay@{request.app.config.host}':
 		return json_response({'error': 'user not found'}, status=404)
 
 	data = {
 		'subject': subject,
-		'aliases': [config.actor],
+		'aliases': [request.app.config.actor],
 		'links': [
-			{'href': config.actor, 'rel': 'self', 'type': 'application/activity+json'},
+			{'href': request.app.config.actor, 'rel': 'self', 'type': 'application/activity+json'},
-			{'href': config.actor, 'rel': 'self', 'type': 'application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"'}
+			{'href': request.app.config.actor, 'rel': 'self', 'type': 'application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"'}
 		]
 	}
 
 	return json_response(data)
 
 
+@register_route('GET', '/nodeinfo/2.0.json')
 async def nodeinfo_2_0(request):
 	data = {
 		# XXX - is this valid for a relay?
@@ -165,7 +183,7 @@ async def nodeinfo_2_0(request):
 			}
 		},
 		'metadata': {
-			'peers': app['database'].hostnames
+			'peers': request.app.database.hostnames
 		},
 		'version': '2.0'
 	}
@@ -173,17 +191,19 @@ async def nodeinfo_2_0(request):
 	return json_response(data)
 
 
+@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'
+				'href': f'https://{request.app.config.host}/nodeinfo/2.0.json'
 			}
 		]
 	}
 	return json_response(data)
 
 
+@register_route('GET', '/stats')
 async def stats(request):
     return json_response(STATS)