pch_xyz/sedi-relay
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Compare commits

base: pch_xyz:master
Branches Tags
pch_xyz:master
pch_xyz:pch_xyz-patch-1
pch_xyz:sqldatabase
pch_xyz:0.2.4
pch_xyz:0.2.3
pch_xyz:0.2.2
pch_xyz:0.2.1
pch_xyz:0.2.0
pch_xyz:0.1.0
..
compare: pch_xyz:0.2.2
Branches Tags
pch_xyz:master
pch_xyz:pch_xyz-patch-1
pch_xyz:sqldatabase
pch_xyz:0.2.4
pch_xyz:0.2.3
pch_xyz:0.2.2
pch_xyz:0.2.1
pch_xyz:0.2.0
pch_xyz:0.1.0

No commits in common. "master" and "0.2.2" have entirely different histories.
master ... 0.2.2

18 changed files with 969 additions and 1503 deletions
Show stats Expand all files Collapse all files

28
docs/commands.md
View file

@ -24,27 +24,6 @@ Run the setup wizard to configure your relay.
activityrelay setup 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 ## Inbox
Manage the list of subscribed instances. Manage the list of subscribed instances.
@ -118,13 +97,6 @@ Remove a domain from the whitelist.
activityrelay whitelist remove <domain> activityrelay whitelist remove <domain>
### Import
Add all current inboxes to the whitelist
activityrelay whitelist import
## Instance ## Instance
Manage the instance ban list. Manage the instance ban list.

55
docs/configuration.md
View file

@ -1,8 +1,6 @@
# Configuration # Configuration
## General ## DB
### DB
The path to the database. It contains the relay actor private key and all subscribed 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. instances. If the path is not absolute, it is relative to the working directory.
@ -10,7 +8,7 @@ instances. If the path is not absolute, it is relative to the working directory.
db: relay.jsonld db: relay.jsonld
### Listener ## Listener
The address and port the relay will listen on. If the reverse proxy (nginx, apache, caddy, etc) 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` is running on the same host, it is recommended to change `listen` to `localhost`
@ -19,41 +17,22 @@ is running on the same host, it is recommended to change `listen` to `localhost`
port: 8080 port: 8080
### Note ## Note
A small blurb to describe your relay instance. This will show up on the relay's home page. 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." 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 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 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. 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_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 ## AP
Various ActivityPub-related settings Various ActivityPub-related settings
@ -103,3 +82,29 @@ setting this to the below list will block all other relays and prevent relay cha
- aoderelay - aoderelay
- social.seattle.wa.us-relay - social.seattle.wa.us-relay
- unciarelay - 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

4
docs/installation.md
View file

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

50
relay.spec
View file

@ -1,50 +0,0 @@
# -*- 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,
)

22
relay.yaml.example
View file

@ -9,19 +9,13 @@ port: 8080
# Note # Note
note: "Make a note about your instance here." note: "Make a note about your instance here."
# Number of worker threads to start. If 0, use asyncio futures instead of threads. # maximum number of inbox posts to do at once
workers: 0 post_limit: 512
# 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: ap:
# This is used for generating activitypub messages, as well as instructions for # 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. # linking AP identities. it should be an SSL-enabled domain reachable by https.
host: 'relay.example.com' host: 'relay.example.com'
blocked_instances: blocked_instances:
@ -41,3 +35,9 @@ ap:
#- 'aoderelay' #- 'aoderelay'
#- 'social.seattle.wa.us-relay' #- 'social.seattle.wa.us-relay'
#- 'unciarelay' #- 'unciarelay'
# cache limits as number of items. only change this if you know what you're doing
cache:
objects: 1024
actors: 1024
digests: 1024

7
relay/__init__.py
View file

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

2
relay/__main__.py
View file

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

216
relay/application.py
View file

@ -1,216 +0,0 @@
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))

141
relay/config.py
View file

@ -1,51 +1,109 @@
import json import json
import os
import yaml import yaml
from functools import cached_property
from pathlib import Path from pathlib import Path
from urllib.parse import urlparse from urllib.parse import urlparse
from .misc import DotDict, boolean
relay_software_names = [
RELAY_SOFTWARE = [ 'activityrelay',
'activityrelay', # https://git.pleroma.social/pleroma/relay 'aoderelay',
'aoderelay', # https://git.asonix.dog/asonix/relay 'social.seattle.wa.us-relay',
'feditools-relay' # https://git.ptzo.gdn/feditools/relay 'unciarelay'
] ]
APKEYS = [
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', 'host',
'whitelist_enabled', 'whitelist_enabled',
'blocked_software', 'blocked_software',
'blocked_instances', 'blocked_instances',
'whitelist' 'whitelist'
] }
cachekeys = {
'json',
'objects',
'digests'
}
class RelayConfig(DotDict): def __init__(self, path, is_docker):
def __init__(self, path): if is_docker:
DotDict.__init__(self, {}) path = '/data/relay.yaml'
if self.is_docker:
path = '/data/config.yaml'
self._isdocker = is_docker
self._path = Path(path).expanduser() self._path = Path(path).expanduser()
self.reset()
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
})
def __setitem__(self, key, value): def __setitem__(self, key, value):
if self._isdocker and key in ['db', 'listen', 'port']:
return
if key in ['blocked_instances', 'blocked_software', 'whitelist']: if key in ['blocked_instances', 'blocked_software', 'whitelist']:
assert isinstance(value, (list, set, tuple)) assert isinstance(value, (list, set, tuple))
elif key in ['port', 'workers', 'json_cache', 'timeout']: elif key in ['port', 'json', 'objects', 'digests']:
if not isinstance(value, int): assert isinstance(value, (int))
value = int(value)
elif key == 'whitelist_enabled': elif key == 'whitelist_enabled':
if not isinstance(value, bool): assert isinstance(value, bool)
value = boolean(value)
super().__setitem__(key, value) super().__setitem__(key, value)
@ -75,30 +133,6 @@ class RelayConfig(DotDict):
return f'{self.actor}#main-key' 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): def ban_instance(self, instance):
if instance.startswith('http'): if instance.startswith('http'):
instance = urlparse(instance).hostname instance = urlparse(instance).hostname
@ -184,8 +218,6 @@ class RelayConfig(DotDict):
def load(self): def load(self):
self.reset()
options = {} options = {}
try: try:
@ -205,15 +237,13 @@ class RelayConfig(DotDict):
return False return False
for key, value in config.items(): for key, value in config.items():
if key in ['ap']: if key in ['ap', 'cache']:
for k, v in value.items(): for k, v in value.items():
if k not in self: if k not in self:
continue continue
self[k] = v self[k] = v
continue
elif key not in self: elif key not in self:
continue continue
@ -227,16 +257,13 @@ class RelayConfig(DotDict):
def save(self): def save(self):
config = { config = {
# just turning config.db into a string is good enough for now 'db': self['db'],
'db': str(self.db),
'listen': self.listen, 'listen': self.listen,
'port': self.port, 'port': self.port,
'note': self.note, 'note': self.note,
'push_limit': self.push_limit, 'push_limit': self.push_limit,
'workers': self.workers, 'ap': {key: self[key] for key in self.apkeys},
'json_cache': self.json_cache, 'cache': {key: self[key] for key in self.cachekeys}
'timeout': self.timeout,
'ap': {key: self[key] for key in APKEYS}
} }
with open(self._path, 'w') as fd: with open(self._path, 'w') as fd:

206
relay/database.py
View file

@ -1,33 +1,50 @@
import aputils
import asyncio
import json import json
import logging import logging
import traceback import traceback
from Crypto.PublicKey import RSA
from urllib.parse import urlparse from urllib.parse import urlparse
class RelayDatabase(dict): class RelayDatabase:
def __init__(self, config): def __init__(self, config):
dict.__init__(self, {
'relay-list': {},
'private-key': None,
'follow-requests': {},
'version': 1
})
self.config = config self.config = config
self.signer = None 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
@property @property
def hostnames(self): def hostnames(self):
return tuple(self['relay-list'].keys()) return [urlparse(inbox).hostname for inbox in self.inboxes]
@property @property
def inboxes(self): def inboxes(self):
return tuple(data['inbox'] for data in self['relay-list'].values()) 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')
def load(self): def load(self):
@ -35,36 +52,14 @@ class RelayDatabase(dict):
try: try:
with self.config.db.open() as fd: with self.config.db.open() as fd:
data = json.load(fd) self.data = json.load(fd)
self['version'] = data.get('version', None) key = self.data.pop('actorKeys', None)
self['private-key'] = data.get('private-key')
if self['version'] == None: if key:
self['version'] = 1 self.data['private-key'] = key.get('privateKey')
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 new_db = False
except FileNotFoundError: except FileNotFoundError:
@ -74,13 +69,20 @@ class RelayDatabase(dict):
if self.config.db.stat().st_size > 0: if self.config.db.stat().st_size > 0:
raise e from None raise e from None
if not self['private-key']: 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:
logging.info("No actor keys present, generating 4096-bit RSA keypair.") logging.info("No actor keys present, generating 4096-bit RSA keypair.")
self.signer = aputils.Signer.new(self.config.keyid, size=4096) self.generate_key()
self['private-key'] = self.signer.export()
else: else:
self.signer = aputils.Signer(self['private-key'], self.config.keyid) self.PRIVKEY = RSA.importKey(self.privkey)
self.save() self.save()
return not new_db return not new_db
@ -88,110 +90,34 @@ class RelayDatabase(dict):
def save(self): def save(self):
with self.config.db.open('w') as fd: with self.config.db.open('w') as fd:
json.dump(self, fd, indent=4) data = {
'relay-list': self.inboxes,
'private-key': self.privkey
}
json.dump(data, fd, indent=4)
def get_inbox(self, domain, fail=False): def get_inbox(self, domain):
if domain.startswith('http'): if domain.startswith('http'):
domain = urlparse(domain).hostname domain = urlparse(domain).hostname
inbox = self['relay-list'].get(domain) for inbox in self.inboxes:
if domain == urlparse(inbox).hostname:
if inbox:
return inbox return inbox
if fail:
raise KeyError(domain) def add_inbox(self, inbox):
assert inbox.startswith('https')
assert not self.get_inbox(inbox)
self.data['relay-list'].append(inbox)
def add_inbox(self, inbox, followid=None, software=None): def del_inbox(self, inbox_url):
assert inbox.startswith('https'), 'Inbox must be a url' inbox = self.get_inbox(inbox_url)
domain = urlparse(inbox).hostname
instance = self.get_inbox(domain)
if instance: if not inbox:
if followid: raise KeyError(inbox_url)
instance['followid'] = followid
if software: self.data['relay-list'].remove(inbox)
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, domain, followid=None, fail=False):
data = self.get_inbox(domain, fail=False)
if not data:
if fail:
raise KeyError(domain)
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']

222
relay/http_client.py
View file

@ -1,222 +0,0 @@
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)

68
relay/http_debug.py Normal file
View file

@ -0,0 +1,68 @@
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]

391
relay/manage.py
View file

@ -1,19 +1,17 @@
import Crypto import Crypto
import asyncio import asyncio
import click import click
import json
import logging import logging
import os
import platform import platform
from urllib.parse import urlparse from aiohttp.web import AppRunner, TCPSite
from cachetools import LRUCache
from . import misc, __version__ from . import app, misc, views, __version__
from . import http_client as http from .config import DotDict, RelayConfig, relay_software_names
from .application import Application from .database import RelayDatabase
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) @click.group('cli', context_settings={'show_default': True}, invoke_without_command=True)
@ -21,100 +19,32 @@ CONFIG_IGNORE = {'blocked_software', 'blocked_instances', 'whitelist'}
@click.version_option(version=__version__, prog_name='ActivityRelay') @click.version_option(version=__version__, prog_name='ActivityRelay')
@click.pass_context @click.pass_context
def cli(ctx, config): def cli(ctx, config):
global app app['is_docker'] = bool(os.environ.get('DOCKER_RUNNING'))
app = Application(config) 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])
if not ctx.invoked_subcommand: if not ctx.invoked_subcommand:
if app.config.host.endswith('example.com'): if app['config'].host.endswith('example.com'):
cli_setup.callback() relay_setup.callback()
else: else:
cli_run.callback() relay_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') @cli.group('inbox')
def cli_inbox(): @click.pass_context
def cli_inbox(ctx):
'Manage the inboxes in the database' 'Manage the inboxes in the database'
pass pass
@ -125,7 +55,7 @@ def cli_inbox_list():
click.echo('Connected to the following instances or relays:') 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}') click.echo(f'- {inbox}')
@ -134,34 +64,29 @@ def cli_inbox_list():
def cli_inbox_follow(actor): def cli_inbox_follow(actor):
'Follow an actor (Relay must be running)' 'Follow an actor (Relay must be running)'
if app.config.is_banned(actor): config = app['config']
database = app['database']
if config.is_banned(actor):
return click.echo(f'Error: Refusing to follow banned actor: {actor}') return click.echo(f'Error: Refusing to follow banned actor: {actor}')
if not actor.startswith('http'): if not actor.startswith('http'):
domain = actor
actor = f'https://{actor}/actor' actor = f'https://{actor}/actor'
else: if database.get_inbox(actor):
domain = urlparse(actor).hostname return click.echo(f'Error: Already following actor: {actor}')
try: actor_data = run_in_loop(misc.request, actor, sign_headers=True)
inbox_data = app.database['relay-list'][domain]
inbox = inbox_data['inbox']
except KeyError:
actor_data = asyncio.run(http.get(app.database, actor, sign_headers=True))
if not actor_data: if not actor_data:
return click.echo(f'Failed to fetch actor: {actor}') return click.echo(f'Error: Failed to fetch actor: {actor}')
inbox = actor_data.shared_inbox inbox = misc.get_actor_inbox(actor_data)
message = misc.Message.new_follow( database.add_inbox(inbox)
host = app.config.host, database.save()
actor = actor
)
asyncio.run(http.post(app.database, inbox, message)) run_in_loop(misc.follow_remote_actor, actor)
click.echo(f'Sent follow message to actor: {actor}') click.echo(f'Sent follow message to actor: {actor}')
@ -170,36 +95,18 @@ def cli_inbox_follow(actor):
def cli_inbox_unfollow(actor): def cli_inbox_unfollow(actor):
'Unfollow an actor (Relay must be running)' 'Unfollow an actor (Relay must be running)'
database = app['database']
if not actor.startswith('http'): if not actor.startswith('http'):
domain = actor
actor = f'https://{actor}/actor' actor = f'https://{actor}/actor'
else: if not database.get_inbox(actor):
domain = urlparse(actor).hostname return click.echo(f'Error: Not following actor: {actor}')
try: database.del_inbox(actor)
inbox_data = app.database['relay-list'][domain] database.save()
inbox = inbox_data['inbox']
message = misc.Message.new_unfollow(
host = app.config.host,
actor = actor,
follow = inbox_data['followid']
)
except KeyError: run_in_loop(misc.unfollow_remote_actor, actor)
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}') click.echo(f'Sent unfollow message to: {actor}')
@ -208,18 +115,22 @@ def cli_inbox_unfollow(actor):
def cli_inbox_add(inbox): def cli_inbox_add(inbox):
'Add an inbox to the database' 'Add an inbox to the database'
database = app['database']
config = app['config']
if not inbox.startswith('http'): if not inbox.startswith('http'):
inbox = f'https://{inbox}/inbox' inbox = f'https://{inbox}/inbox'
if app.config.is_banned(inbox): if database.get_inbox(inbox):
return click.echo(f'Error: Refusing to add banned inbox: {inbox}') click.echo(f'Error: Inbox already in database: {inbox}')
return
if app.database.get_inbox(inbox): if config.is_banned(inbox):
return click.echo(f'Error: Inbox already in database: {inbox}') click.echo(f'Error: Refusing to add banned inbox: {inbox}')
return
app.database.add_inbox(inbox)
app.database.save()
database.add_inbox(inbox)
database.save()
click.echo(f'Added inbox to the database: {inbox}') click.echo(f'Added inbox to the database: {inbox}')
@ -228,16 +139,15 @@ def cli_inbox_add(inbox):
def cli_inbox_remove(inbox): def cli_inbox_remove(inbox):
'Remove an inbox from the database' 'Remove an inbox from the database'
try: database = app['database']
dbinbox = app.database.get_inbox(inbox, fail=True) dbinbox = database.get_inbox(inbox)
except KeyError: if not dbinbox:
click.echo(f'Error: Inbox does not exist: {inbox}') click.echo(f'Error: Inbox does not exist: {inbox}')
return return
app.database.del_inbox(dbinbox['domain']) database.del_inbox(dbinbox)
app.database.save() database.save()
click.echo(f'Removed inbox from the database: {inbox}') click.echo(f'Removed inbox from the database: {inbox}')
@ -253,7 +163,7 @@ def cli_instance_list():
click.echo('Banned instances or relays:') click.echo('Banned instances or relays:')
for domain in app.config.blocked_instances: for domain in app['config'].blocked_instances:
click.echo(f'- {domain}') click.echo(f'- {domain}')
@ -262,14 +172,16 @@ def cli_instance_list():
def cli_instance_ban(target): def cli_instance_ban(target):
'Ban an instance and remove the associated inbox if it exists' 'Ban an instance and remove the associated inbox if it exists'
if target.startswith('http'): config = app['config']
target = urlparse(target).hostname database = app['database']
inbox = database.get_inbox(target)
if app.config.ban_instance(target): if config.ban_instance(target):
app.config.save() config.save()
if app.database.del_inbox(target): if inbox:
app.database.save() database.del_inbox(inbox)
database.save()
click.echo(f'Banned instance: {target}') click.echo(f'Banned instance: {target}')
return return
@ -282,8 +194,10 @@ def cli_instance_ban(target):
def cli_instance_unban(target): def cli_instance_unban(target):
'Unban an instance' 'Unban an instance'
if app.config.unban_instance(target): config = app['config']
app.config.save()
if config.unban_instance(target):
config.save()
click.echo(f'Unbanned instance: {target}') click.echo(f'Unbanned instance: {target}')
return return
@ -303,7 +217,7 @@ def cli_software_list():
click.echo('Banned software:') click.echo('Banned software:')
for software in app.config.blocked_software: for software in app['config'].blocked_software:
click.echo(f'- {software}') click.echo(f'- {software}')
@ -315,23 +229,25 @@ def cli_software_list():
def cli_software_ban(name, fetch_nodeinfo): def cli_software_ban(name, fetch_nodeinfo):
'Ban software. Use RELAYS for NAME to ban relays' 'Ban software. Use RELAYS for NAME to ban relays'
if name == 'RELAYS': config = app['config']
for name in RELAY_SOFTWARE:
app.config.ban_software(name)
app.config.save() if name == 'RELAYS':
for name in relay_software_names:
config.ban_software(name)
config.save()
return click.echo('Banned all relay software') return click.echo('Banned all relay software')
if fetch_nodeinfo: if fetch_nodeinfo:
nodeinfo = asyncio.run(http.fetch_nodeinfo(app.database, name)) software = run_in_loop(fetch_nodeinfo, name)
if not nodeinfo: if not software:
click.echo(f'Failed to fetch software name from domain: {name}') click.echo(f'Failed to fetch software name from domain: {name}')
name = nodeinfo.sw_name name = software
if app.config.ban_software(name): if config.ban_software(name):
app.config.save() config.save()
return click.echo(f'Banned software: {name}') return click.echo(f'Banned software: {name}')
click.echo(f'Software already banned: {name}') click.echo(f'Software already banned: {name}')
@ -345,28 +261,31 @@ def cli_software_ban(name, fetch_nodeinfo):
def cli_software_unban(name, fetch_nodeinfo): def cli_software_unban(name, fetch_nodeinfo):
'Ban software. Use RELAYS for NAME to unban relays' 'Ban software. Use RELAYS for NAME to unban relays'
if name == 'RELAYS': config = app['config']
for name in RELAY_SOFTWARE:
app.config.unban_software(name)
app.config.save() if name == 'RELAYS':
for name in relay_software_names:
config.unban_software(name)
config.save()
return click.echo('Unbanned all relay software') return click.echo('Unbanned all relay software')
if fetch_nodeinfo: if fetch_nodeinfo:
nodeinfo = asyncio.run(http.fetch_nodeinfo(app.database, name)) software = run_in_loop(fetch_nodeinfo, name)
if not nodeinfo: if not software:
click.echo(f'Failed to fetch software name from domain: {name}') click.echo(f'Failed to fetch software name from domain: {name}')
name = nodeinfo.sw_name name = software
if app.config.unban_software(name): if config.unban_software(name):
app.config.save() config.save()
return click.echo(f'Unbanned software: {name}') return click.echo(f'Unbanned software: {name}')
click.echo(f'Software wasn\'t banned: {name}') click.echo(f'Software wasn\'t banned: {name}')
@cli.group('whitelist') @cli.group('whitelist')
def cli_whitelist(): def cli_whitelist():
'Manage the instance whitelist' 'Manage the instance whitelist'
@ -375,11 +294,9 @@ def cli_whitelist():
@cli_whitelist.command('list') @cli_whitelist.command('list')
def cli_whitelist_list(): def cli_whitelist_list():
'List all the instances in the whitelist'
click.echo('Current whitelisted domains') click.echo('Current whitelisted domains')
for domain in app.config.whitelist: for domain in app['config'].whitelist:
click.echo(f'- {domain}') click.echo(f'- {domain}')
@ -388,10 +305,12 @@ def cli_whitelist_list():
def cli_whitelist_add(instance): def cli_whitelist_add(instance):
'Add an instance to the whitelist' 'Add an instance to the whitelist'
if not app.config.add_whitelist(instance): config = app['config']
if not config.add_whitelist(instance):
return click.echo(f'Instance already in the whitelist: {instance}') return click.echo(f'Instance already in the whitelist: {instance}')
app.config.save() config.save()
click.echo(f'Instance added to the whitelist: {instance}') click.echo(f'Instance added to the whitelist: {instance}')
@ -400,24 +319,106 @@ def cli_whitelist_add(instance):
def cli_whitelist_remove(instance): def cli_whitelist_remove(instance):
'Remove an instance from the whitelist' 'Remove an instance from the whitelist'
if not app.config.del_whitelist(instance): config = app['config']
database = app['database']
inbox = database.get_inbox(instance)
if not config.del_whitelist(instance):
return click.echo(f'Instance not in the whitelist: {instance}') return click.echo(f'Instance not in the whitelist: {instance}')
app.config.save() config.save()
if app.config.whitelist_enabled: if inbox and config.whitelist_enabled:
if app.database.del_inbox(instance): database.del_inbox(inbox)
app.database.save() database.save()
click.echo(f'Removed instance from the whitelist: {instance}') click.echo(f'Removed instance from the whitelist: {instance}')
@cli_whitelist.command('import') @cli.command('setup')
def cli_whitelist_import(): def relay_setup():
'Add all current inboxes to the whitelist' 'Generate a new config'
for domain in app.database.hostnames: config = app['config']
cli_whitelist_add.callback(domain)
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()
def main(): def main():

531
relay/misc.py
View file

@ -1,69 +1,32 @@
import aputils
import asyncio import asyncio
import base64 import base64
import json import json
import logging import logging
import socket import socket
import traceback import traceback
import uuid
from aiohttp.hdrs import METH_ALL as METHODS from Crypto.Hash import SHA, SHA256, SHA512
from aiohttp.web import Response as AiohttpResponse, View as AiohttpView from Crypto.PublicKey import RSA
from Crypto.Signature import PKCS1_v1_5
from aiohttp import ClientSession
from datetime import datetime from datetime import datetime
from json.decoder import JSONDecodeError from json.decoder import JSONDecodeError
from urllib.parse import urlparse from urllib.parse import urlparse
from uuid import uuid4 from uuid import uuid4
from . import app
from .http_debug import http_debug
app = None
MIMETYPES = { HASHES = {
'activity': 'application/activity+json', 'sha1': SHA,
'html': 'text/html', 'sha256': SHA256,
'json': 'application/json', 'sha512': SHA512
'text': 'text/plain'
}
NODEINFO_NS = {
'20': 'http://nodeinfo.diaspora.software/ns/schema/2.0',
'21': 'http://nodeinfo.diaspora.software/ns/schema/2.1'
} }
def set_app(new_app): def build_signing_string(headers, used_headers):
global app return '\n'.join(map(lambda x: ': '.join([x.lower(), headers[x]]), used_headers))
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): def check_open_port(host, port):
@ -78,263 +41,279 @@ def check_open_port(host, port):
return False return False
class DotDict(dict): def create_signature_header(headers):
def __init__(self, _data, **kwargs): headers = {k.lower(): v for k, v in headers.items()}
dict.__init__(self) used_headers = headers.keys()
sigstring = build_signing_string(headers, used_headers)
self.update(_data, **kwargs) 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)
def __getattr__(self, k): def distill_object_id(activity):
logging.debug(f'>> determining object ID for {activity["object"]}')
try: try:
return self[k] 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: except KeyError:
raise AttributeError(f'{self.__class__.__name__} object has no attribute {k}') from None return False
def __setattr__(self, k, v): async def follow_remote_actor(actor_uri):
if k.startswith('_'): config = app['config']
super().__setattr__(k, v)
else: actor = await request(actor_uri)
self[k] = v 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)
def __setitem__(self, k, v): async def unfollow_remote_actor(actor_uri):
if type(v) == dict: config = app['config']
v = DotDict(v)
super().__setitem__(k, v) actor = await request(actor_uri)
if not actor:
logging.error(f'failed to fetch actor: {actor_uri}')
return
def __delattr__(self, k): inbox = get_actor_inbox(actor)
try: logging.verbose(f'sending unfollow request to inbox: {inbox}')
dict.__delitem__(self, k)
except KeyError: message = {
raise AttributeError(f'{self.__class__.__name__} object has no attribute {k}') from None "@context": "https://www.w3.org/ns/activitystreams",
"type": "Undo",
"to": [actor_uri],
@classmethod "object": {
def new_from_json(cls, data): "type": "Follow",
if not data: "object": actor_uri,
raise JSONDecodeError('Empty body', data, 1) "actor": actor_uri,
"id": f"https://{config.host}/activities/{uuid4()}"
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://{config.host}/activities/{uuid4()}",
'id': f'https://{host}/actor#main-key', "actor": f"https://{config.host}/actor"
'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 isinstance(body, bytes): await request(inbox, message)
kwargs['body'] = body
elif isinstance(body, dict) and ctype in {'json', 'activity'}:
kwargs['text'] = json.dumps(body)
else:
kwargs['text'] = body
return cls(**kwargs)
@classmethod async def request(uri, data=None, force=False, sign_headers=True, activity=True):
def new_error(cls, status, body, ctype='text'): ## If a get request and not force, try to use the cache first
if ctype == 'json': if not data and not force:
body = json.dumps({'status': status, 'error': body}) try:
return app['cache'].json[uri]
return cls.new(body=body, status=status, ctype=ctype) except KeyError:
pass
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
## Set the accepted content type for a GET
elif not data and 'Accept' not in headers:
headers['Accept'] = mimetype
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
}
if data:
assert isinstance(data, dict)
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')))
})
signing_headers['Signature'] = create_signature_header(signing_headers)
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()
@property async def validate_signature(actor, http_request):
def location(self): pubkey = await fetch_actor_key(actor)
return self.headers.get('Location')
if not pubkey:
return False
@location.setter logging.debug(f'actor key: {pubkey}')
def location(self, value):
self.headers['Location'] = value
headers = {key.lower(): value for key, value in http_request.headers.items()}
headers['(request-target)'] = ' '.join([http_request.method.lower(), http_request.path])
class View(AiohttpView): sig = split_signature(headers['signature'])
async def _iter(self): logging.debug(f'sigdata: {sig}')
if self.request.method not in METHODS:
self._raise_allowed_methods()
method = getattr(self, self.request.method.lower(), None) sigstring = build_signing_string(headers, sig['headers'])
logging.debug(f'sigstring: {sigstring}')
if method is None: sign_alg, _, hash_alg = sig['algorithm'].partition('-')
self._raise_allowed_methods() logging.debug(f'sign alg: {sign_alg}, hash alg: {hash_alg}')
return await method(**self.request.match_info) sigdata = base64.b64decode(sig['signature'])
pkcs = PKCS1_v1_5.new(pubkey)
h = HASHES[hash_alg].new()
h.update(sigstring.encode('ascii'))
result = pkcs.verify(h, sigdata)
@property http_request['validated'] = result
def app(self):
return self._request.app
logging.debug(f'validates? {result}')
@property return result
def config(self):
return self.app.config
@property
def database(self):
return self.app.database

189
relay/processors.py
View file

@ -1,138 +1,110 @@
import asyncio import asyncio
import logging import logging
from cachetools import LRUCache
from uuid import uuid4 from uuid import uuid4
from .misc import Message from . import app, misc
cache = LRUCache(1024) async def handle_relay(actor, data, request):
cache = app['cache'].objects
object_id = misc.distill_object_id(data)
if object_id in cache:
def person_check(actor, software): logging.verbose(f'already relayed {object_id} as {cache[object_id]}')
## 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 return
message = Message.new_announce( logging.verbose(f'Relaying post from {actor["id"]}')
host = request.config.host,
object = request.message.objectid 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
}
cache[request.message.objectid] = message.id
logging.debug(f'>> relay: {message}') logging.debug(f'>> relay: {message}')
inboxes = request.database.distill_inboxes(request.message) inboxes = misc.distill_inboxes(actor, object_id)
futures = [misc.request(inbox, data=message) for inbox in inboxes]
for inbox in inboxes: asyncio.ensure_future(asyncio.gather(*futures))
request.app.push_message(inbox, message) cache[object_id] = activity_id
async def handle_forward(request): async def handle_forward(actor, data, request):
if request.message.id in cache: cache = app['cache'].objects
logging.verbose(f'already forwarded {request.message.id}') object_id = misc.distill_object_id(data)
if object_id in cache:
logging.verbose(f'already forwarded {object_id}')
return return
message = Message.new_announce( logging.verbose(f'Forwarding post from {actor["id"]}')
host = request.config.host, logging.debug(f'>> Relay {data}')
object = request.message
)
cache[request.message.id] = message.id inboxes = misc.distill_inboxes(actor, object_id)
logging.debug(f'>> forward: {message}')
inboxes = request.database.distill_inboxes(request.message) futures = [misc.request(inbox, data=data) for inbox in inboxes]
asyncio.ensure_future(asyncio.gather(*futures))
for inbox in inboxes: cache[object_id] = object_id
request.app.push_message(inbox, message)
async def handle_follow(request): async def handle_follow(actor, data, request):
nodeinfo = await request.app.client.fetch_nodeinfo(request.actor.domain) config = app['config']
software = nodeinfo.sw_name if nodeinfo else None database = app['database']
## reject if software used by actor is banned inbox = misc.get_actor_inbox(actor)
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
)
)
return logging.verbose(f'Rejected follow from actor for using specific software: actor={request.actor.id}, software={software}') if inbox not in database.inboxes:
database.add_inbox(inbox)
database.save()
asyncio.ensure_future(misc.follow_remote_actor(actor['id']))
## reject if the actor is not an instance actor message = {
if person_check(request.actor, software): "@context": "https://www.w3.org/ns/activitystreams",
request.app.push_message( "type": "Accept",
request.actor.shared_inbox, "to": [actor["id"]],
Message.new_response( "actor": config.actor,
host = request.config.host,
actor = request.actor.id,
followid = request.message.id,
accept = False
)
)
return logging.verbose(f'Non-application actor tried to follow: {request.actor.id}') # 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"]
},
request.database.add_inbox(request.actor.shared_inbox, request.message.id, software) "id": f"https://{request.host}/activities/{uuid4()}",
request.database.save() }
request.app.push_message( asyncio.ensure_future(misc.request(inbox, 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(request): async def handle_undo(actor, data, request):
## If the object is not a Follow, forward it ## If the activity being undone is an Announce, forward it insteead
if request.message.object.type != 'Follow': if data['object']['type'] == 'Announce':
return await handle_forward(request) await handle_forward(actor, data, request)
if not request.database.del_inbox(request.actor.domain, request.message.id):
return return
request.database.save() elif data['object']['type'] != 'Follow':
return
request.app.push_message( database = app['database']
request.actor.shared_inbox, inbox = database.get_inbox(actor['id'])
Message.new_unfollow(
host = request.config.host, if not inbox:
actor = request.actor.id, return
follow = request.message
) database.del_inbox(inbox)
) database.save()
await misc.unfollow_remote_actor(actor['id'])
processors = { processors = {
@ -145,16 +117,9 @@ processors = {
} }
async def run_processor(request): async def run_processor(request, data, actor):
if request.message.type not in processors: if data['type'] not in processors:
return return
if request.instance and not request.instance.get('software'): logging.verbose(f'New activity from actor: {actor["id"]} {data["type"]}')
nodeinfo = await request.app.client.fetch_nodeinfo(request.instance['domain']) return await processors[data['type']](actor, data, request)
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)

291
relay/views.py
View file

@ -1,47 +1,28 @@
import aputils
import asyncio
import logging import logging
import subprocess import subprocess
import traceback import traceback
from pathlib import Path from aiohttp.web import HTTPForbidden, HTTPUnauthorized, Response, json_response
from urllib.parse import urlparse
from . import __version__, misc from . import __version__, app, misc
from .misc import DotDict, Message, Response from .http_debug import STATS
from .processors import run_processor from .processors import run_processor
routes = []
version = __version__
if Path(__file__).parent.parent.joinpath('.git').exists():
try: try:
commit_label = subprocess.check_output(["git", "rev-parse", "HEAD"]).strip().decode('ascii') commit_label = subprocess.check_output(["git", "rev-parse", "HEAD"]).strip().decode('ascii')
version = f'{__version__} {commit_label}' version = f'{__version__} {commit_label}'
except: except:
pass version = __version__
def register_route(method, path):
def wrapper(func):
routes.append([method, path, func])
return func
return wrapper
@register_route('GET', '/')
async def home(request): async def home(request):
targets = '<br>'.join(request.database.hostnames) targets = '<br>'.join(app['database'].hostnames)
note = request.config.note text = """
count = len(request.database.hostnames)
host = request.config.host
text = f"""
<html><head> <html><head>
<title>SEDI中繼器</title> <title>ActivityPub Relay at {host}</title>
<style> <style>
p {{ color: #FFFFFF; font-family: monospace, arial; font-size: 100%; }} p {{ color: #FFFFFF; font-family: monospace, arial; font-size: 100%; }}
body {{ background-color: #000000; }} body {{ background-color: #000000; }}
@ -53,142 +34,170 @@ a:hover {{ color: #8AF; }}
<body> <body>
<p>This is an Activity Relay for fediverse instances.</p> <p>This is an Activity Relay for fediverse instances.</p>
<p>{note}</p> <p>{note}</p>
<p>Misskey及Mastodon站長請訂閱這個地址：<a href="https://{host}/inbox">https://{host}/inbox</a></p> <p>You may subscribe to this relay with the address: <a href="https://{host}/actor">https://{host}/actor</a></p>
<p>Pleroma及Friendica站長請訂閱這個地址：<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>
<p>原始碼：<a href="https://git.seediqbale.xyz/pch_xyz/sedi-relay">https://git.seediqbale.xyz/pch_xyz/sedi-relay</a></p> <br><p>List of {count} registered instances:<br>{targets}</p>
<p>請我喝杯咖啡：<a href="https://buymeacoffee.com/SEDI">https://buymeacoffee.com/SEDI</a></p> </body></html>""".format(host=request.host, note=app['config'].note, targets=targets, count=len(app['database'].inboxes))
<p>activityrelay v0.2.4</p>
<br><p>有 {count} 個實例訂閱中：<br>{targets}</p>
</body></html>"""
return Response.new(text, ctype='html') return Response(
status = 200,
content_type = 'text/html',
@register_route('GET', '/inbox') charset = 'utf-8',
@register_route('GET', '/actor') text = text
async def actor(request):
data = Message.new_actor(
host = request.config.host,
pubkey = request.database.signer.pubkey
) )
return Response.new(data, ctype='activity')
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')
@register_route('POST', '/inbox')
@register_route('POST', '/actor')
async def inbox(request): async def inbox(request):
config = request.config config = app['config']
database = request.database database = app['database']
## reject if missing signature header ## reject if missing signature header
if not request.signature: if 'signature' not in request.headers:
logging.verbose('Actor missing signature header') logging.verbose('Actor missing signature header')
raise HTTPUnauthorized(body='missing signature') raise HTTPUnauthorized(body='missing signature')
## read message and get actor id and domain
try: try:
request['message'] = await request.json(loads=Message.new_from_json) data = await request.json()
actor_id = data['actor']
## reject if there is no message actor_domain = urlparse(actor_id).hostname
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')
except:
## this code should hopefully never get called
traceback.print_exc()
logging.verbose('Failed to parse inbox message')
return Response.new_error(400, 'failed to parse message', 'json')
request['actor'] = await request.app.client.get(request.signature.keyid, sign_headers=True)
## reject if actor is empty
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
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 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
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 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 {request.message.to_json(4)}")
asyncio.ensure_future(run_processor(request))
return Response.new(status=202)
@register_route('GET', '/.well-known/webfinger')
async def webfinger(request):
try:
subject = request.query['resource']
except KeyError: except KeyError:
return Response.new_error(400, 'missing \'resource\' query key', 'json') logging.verbose('actor not in data')
raise HTTPUnauthorized(body='no actor in message')
if subject != f'acct:relay@{request.config.host}': ## reject if there is no actor in the message
return Response.new_error(404, 'user not found', 'json') except:
traceback.print_exc()
logging.verbose('Failed to parse inbox message')
raise HTTPUnauthorized(body='failed to parse message')
data = aputils.Webfinger.new( actor = await misc.request(actor_id)
handle = 'relay',
domain = request.config.host,
actor = request.config.actor
)
return Response.new(data, ctype='json') ## reject if actor is empty
if not actor:
logging.verbose(f'Failed to fetch actor: {actor_id}')
raise HTTPUnauthorized('failed to fetch actor')
## 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')
## 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')
## 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')
## 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')
logging.debug(f">> payload {data}")
await run_processor(request, data, actor)
return Response(body=b'{}', content_type='application/activity+json')
@register_route('GET', '/nodeinfo/{version:\d.\d\.json}') async def webfinger(request):
async def nodeinfo(request): config = app['config']
niversion = request.match_info['version'][:3] subject = request.query['resource']
data = dict( if subject != f'acct:relay@{request.host}':
name = 'activityrelay', return json_response({'error': 'user not found'}, status=404)
version = version,
protocols = ['activitypub'],
open_regs = not request.config.whitelist_enabled,
users = 1,
metadata = {'peers': request.database.hostnames}
)
if niversion == '2.1': data = {
data['repo'] = 'https://git.pleroma.social/pleroma/relay' '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\"'}
]
}
return Response.new(aputils.Nodeinfo.new(**data), ctype='json') return json_response(data)
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'
}
return json_response(data)
@register_route('GET', '/.well-known/nodeinfo')
async def nodeinfo_wellknown(request): async def nodeinfo_wellknown(request):
data = aputils.WellKnownNodeinfo.new_template(request.config.host) data = {
return Response.new(data, ctype='json') '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)

6
requirements.txt
View file

@ -1,5 +1 @@
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

15
setup.cfg
View file

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