sedi-relay/relay/config.py

246 lines
4.4 KiB
Python
Raw Normal View History

2022-05-06 07:04:51 +00:00
import json
2022-11-27 00:59:20 +00:00
import os
2022-05-06 07:04:51 +00:00
import yaml
2022-11-27 00:59:20 +00:00
from functools import cached_property
2022-05-06 07:04:51 +00:00
from pathlib import Path
from urllib.parse import urlparse
2022-11-20 11:14:37 +00:00
from .misc import DotDict, boolean
2022-11-06 06:11:54 +00:00
2022-05-06 07:04:51 +00:00
RELAY_SOFTWARE = [
2022-11-27 04:41:57 +00:00
'activityrelay', # https://git.pleroma.social/pleroma/relay
'aoderelay', # https://git.asonix.dog/asonix/relay
'feditools-relay' # https://git.ptzo.gdn/feditools/relay
2022-05-06 07:04:51 +00:00
]
APKEYS = [
'host',
'whitelist_enabled',
'blocked_software',
'blocked_instances',
'whitelist'
]
2022-05-06 07:04:51 +00:00
class RelayConfig(DotDict):
2022-11-27 00:59:20 +00:00
def __init__(self, path):
2022-11-10 17:38:08 +00:00
DotDict.__init__(self, {})
2022-11-27 00:59:20 +00:00
if self.is_docker:
path = '/data/config.yaml'
2022-05-06 07:04:51 +00:00
self._path = Path(path).expanduser()
2022-11-10 17:38:08 +00:00
self.reset()
2022-05-06 07:04:51 +00:00
def __setitem__(self, key, value):
if key in ['blocked_instances', 'blocked_software', 'whitelist']:
assert isinstance(value, (list, set, tuple))
elif key in ['port', 'workers', 'json_cache', 'timeout']:
2022-11-20 11:14:37 +00:00
if not isinstance(value, int):
value = int(value)
2022-05-06 07:04:51 +00:00
elif key == 'whitelist_enabled':
2022-11-20 11:14:37 +00:00
if not isinstance(value, bool):
value = boolean(value)
2022-05-06 07:04:51 +00:00
super().__setitem__(key, value)
@property
def db(self):
return Path(self['db']).expanduser().resolve()
@property
def path(self):
return self._path
@property
def actor(self):
return f'https://{self.host}/actor'
@property
def inbox(self):
return f'https://{self.host}/inbox'
@property
def keyid(self):
return f'{self.actor}#main-key'
2022-11-27 00:59:20 +00:00
@cached_property
def is_docker(self):
return bool(os.environ.get('DOCKER_RUNNING'))
2022-11-10 17:38:08 +00:00
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,
2022-11-20 10:50:14 +00:00
'workers': 0,
2022-11-10 17:38:08 +00:00
'host': 'relay.example.com',
'whitelist_enabled': False,
2022-11-10 17:38:08 +00:00
'blocked_software': [],
'blocked_instances': [],
'whitelist': []
2022-11-10 17:38:08 +00:00
})
2022-11-27 00:59:20 +00:00
2022-05-06 07:04:51 +00:00
def ban_instance(self, instance):
if instance.startswith('http'):
instance = urlparse(instance).hostname
if self.is_banned(instance):
return False
self.blocked_instances.append(instance)
return True
def unban_instance(self, instance):
if instance.startswith('http'):
instance = urlparse(instance).hostname
try:
self.blocked_instances.remove(instance)
return True
except:
return False
def ban_software(self, software):
if self.is_banned_software(software):
return False
self.blocked_software.append(software)
return True
def unban_software(self, software):
try:
self.blocked_software.remove(software)
return True
except:
return False
def add_whitelist(self, instance):
if instance.startswith('http'):
instance = urlparse(instance).hostname
if self.is_whitelisted(instance):
return False
self.whitelist.append(instance)
return True
def del_whitelist(self, instance):
if instance.startswith('http'):
instance = urlparse(instance).hostname
try:
self.whitelist.remove(instance)
return True
except:
return False
def is_banned(self, instance):
if instance.startswith('http'):
instance = urlparse(instance).hostname
return instance in self.blocked_instances
def is_banned_software(self, software):
if not software:
return False
return software.lower() in self.blocked_software
def is_whitelisted(self, instance):
if instance.startswith('http'):
instance = urlparse(instance).hostname
return instance in self.whitelist
def load(self):
2022-11-10 17:38:08 +00:00
self.reset()
2022-05-06 07:04:51 +00:00
options = {}
try:
options['Loader'] = yaml.FullLoader
except AttributeError:
pass
try:
with open(self.path) as fd:
config = yaml.load(fd, **options)
except FileNotFoundError:
return False
if not config:
return False
for key, value in config.items():
if key in ['ap']:
2022-05-06 07:04:51 +00:00
for k, v in value.items():
if k not in self:
continue
self[k] = v
2022-11-20 11:14:37 +00:00
continue
2022-05-06 07:04:51 +00:00
elif key not in self:
continue
self[key] = value
if self.host.endswith('example.com'):
return False
return True
def save(self):
config = {
# just turning config.db into a string is good enough for now
'db': str(self.db),
2022-05-06 07:04:51 +00:00
'listen': self.listen,
'port': self.port,
'note': self.note,
'push_limit': self.push_limit,
2022-11-20 10:50:14 +00:00
'workers': self.workers,
'json_cache': self.json_cache,
'timeout': self.timeout,
'ap': {key: self[key] for key in APKEYS}
2022-05-06 07:04:51 +00:00
}
with open(self._path, 'w') as fd:
yaml.dump(config, fd, sort_keys=False)
return config