import json import yaml from pathlib import Path from urllib.parse import urlparse 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', 'blocked_software', 'blocked_instances', 'whitelist', 'whitelist_enabled' } cachekeys = { 'json', 'objects', 'digests' } def __init__(self, path): super().__init__({ 'db': 'relay.jsonld', 'listen': '0.0.0.0', 'port': 8080, 'note': 'Make a note about your instance here.', 'push_limit': 512, 'host': 'example.com', 'blocked_software': [], 'blocked_instances': [], 'whitelist': [], 'whitelist_enabled': False, 'json': 1024, 'objects': 1024, 'digests': 1024 }) self._path = Path(path).expanduser().resolve() def __setitem__(self, key, value): if key in ['blocked_instances', 'blocked_software', 'whitelist']: assert isinstance(value, (list, set, tuple)) elif key in ['port', 'json', 'objects', 'digests']: assert isinstance(value, (int)) elif key == 'whitelist_enabled': assert isinstance(value, bool) 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' def is_banned(self, inbox): return urlparse(inbox).hostname in self.blocked_instances def is_whitelisted(self, inbox): return urlparse(inbox).hostname in self.whitelist def load(self): 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', 'cache']: for k, v in value.items(): if k not in self: continue self[k] = v elif key not in self: continue self[key] = value if self.host == 'example.com': return False return True def save(self): config = { 'db': self.db, 'listen': self.listen, 'port': self.port, 'note': self.note, 'push_limit': self.push_limit, 'ap': {key: self[key] for key in self.apkeys}, 'cache': {key: self[key] for key in self.cachekeys} } with open(self._path, 'w') as fd: yaml.dump(config, fd, sort_keys=False) return config