diff --git a/relay/views/frontend.py b/relay/views/frontend.py index 448b834..2215d5e 100644 --- a/relay/views/frontend.py +++ b/relay/views/frontend.py @@ -153,104 +153,6 @@ class AdminInstances(View): return Response.new(data, ctype = 'html') - async def post(self, request: Request) -> Response: - post = await request.post() - data: dict[str, str] = {key: value for key, value in post.items()} # type: ignore - - if not data.get('actor') and not data.get('domain'): - return await self.get(request, error = 'Missing actor and/or domain') - - if not data.get('domain'): - data['domain'] = urlparse(data['actor']).netloc - - if not data.get('software'): - nodeinfo = await self.client.fetch_nodeinfo(data['domain']) - - if nodeinfo is None: - return await self.get(request, error = 'Failed to fetch nodeinfo') - - data['software'] = nodeinfo.sw_name - - if not data.get('actor') and data['software'] in ACTOR_FORMATS: - data['actor'] = ACTOR_FORMATS[data['software']].format(domain = data['domain']) - - if not data.get('inbox') and data['actor']: - actor: Message | None = await self.client.get(data['actor'], True, Message) - - if actor is None: - return await self.get(request, error = 'Failed to fetch actor') - - data['inbox'] = actor.shared_inbox - - with self.database.session(True) as conn: - conn.put_inbox(**data) - - return await self.get(request, message = "Added new inbox") - - -@register_route('/admin/instances/delete/{domain}') -class AdminInstancesDelete(View): - async def get(self, request: Request, domain: str) -> Response: - with self.database.session(True) as conn: - if not conn.get_inbox(domain): - return await AdminInstances(request).get(request, error = 'Instance not found') - - conn.del_inbox(domain) - - return await AdminInstances(request).get(request, message = 'Removed instance') - - -@register_route('/admin/instances/approve/{domain}') -class AdminInstancesApprove(View): - async def get(self, request: Request, domain: str) -> Response: - try: - with self.database.session(True) as conn: - instance = conn.put_request_response(domain, True) - - except KeyError: - return await AdminInstances(request).get(request, error = 'Instance not found') - - message = Message.new_response( - host = self.config.domain, - actor = instance['actor'], - followid = instance['followid'], - accept = True - ) - - self.app.push_message(instance['inbox'], message, instance) - - if instance['software'] != 'mastodon': - message = Message.new_follow( - host = self.config.domain, - actor = instance['actor'] - ) - - self.app.push_message(instance['inbox'], message, instance) - - return await AdminInstances(request).get(request, message = 'Request accepted') - - -@register_route('/admin/instances/deny/{domain}') -class AdminInstancesDeny(View): - async def get(self, request: Request, domain: str) -> Response: - try: - with self.database.session(True) as conn: - instance = conn.put_request_response(domain, False) - - except KeyError: - return await AdminInstances(request).get(request, error = 'Instance not found') - - message = Message.new_response( - host = self.config.domain, - actor = instance['actor'], - followid = instance['followid'], - accept = False - ) - - self.app.push_message(instance['inbox'], message, instance) - return await AdminInstances(request).get(request, message = 'Request denied') - - @register_route('/admin/whitelist') class AdminWhitelist(View): async def get(self, @@ -273,34 +175,6 @@ class AdminWhitelist(View): return Response.new(data, ctype = 'html') - async def post(self, request: Request) -> Response: - data = await request.post() - - if not data['domain']: - return await self.get(request, error = 'Missing domain') - - with self.database.session(True) as conn: - if conn.get_domain_whitelist(data['domain']): - return await self.get(request, error = "Domain already in whitelist") - - conn.put_domain_whitelist(data['domain']) - - return await self.get(request, message = "Added/updated domain ban") - - -@register_route('/admin/whitelist/delete/{domain}') -class AdminWhitlistDelete(View): - async def get(self, request: Request, domain: str) -> Response: - with self.database.session() as conn: - if not conn.get_domain_whitelist(domain): - msg = 'Whitelisted domain not found' - return await AdminWhitelist.run("GET", request, error = msg) - - conn.del_domain_whitelist(domain) - - return await AdminWhitelist.run("GET", request, message = 'Removed domain from whitelist') - - @register_route('/admin/domain_bans') class AdminDomainBans(View): async def get(self, @@ -323,42 +197,6 @@ class AdminDomainBans(View): return Response.new(data, ctype = 'html') - async def post(self, request: Request) -> Response: - data = await request.post() - - if not data['domain']: - return await self.get(request, error = 'Missing domain') - - with self.database.session(True) as conn: - if conn.get_domain_ban(data['domain']): - conn.update_domain_ban( - data['domain'], - data.get('reason'), - data.get('note') - ) - - else: - conn.put_domain_ban( - data['domain'], - data.get('reason'), - data.get('note') - ) - - return await self.get(request, message = "Added/updated domain ban") - - -@register_route('/admin/domain_bans/delete/{domain}') -class AdminDomainBansDelete(View): - async def get(self, request: Request, domain: str) -> Response: - with self.database.session() as conn: - if not conn.get_domain_ban(domain): - return await AdminDomainBans.run("GET", request, error = 'Domain ban not found') - - conn.del_domain_ban(domain) - - return await AdminDomainBans.run("GET", request, message = 'Unbanned domain') - - @register_route('/admin/software_bans') class AdminSoftwareBans(View): async def get(self, @@ -381,42 +219,6 @@ class AdminSoftwareBans(View): return Response.new(data, ctype = 'html') - async def post(self, request: Request) -> Response: - data = await request.post() - - if not data['name']: - return await self.get(request, error = 'Missing name') - - with self.database.session(True) as conn: - if conn.get_software_ban(data['name']): - conn.update_software_ban( - data['name'], - data.get('reason'), - data.get('note') - ) - - else: - conn.put_software_ban( - data['name'], - data.get('reason'), - data.get('note') - ) - - return await self.get(request, message = "Added/updated software ban") - - -@register_route('/admin/software_bans/delete/{name}') -class AdminSoftwareBansDelete(View): - async def get(self, request: Request, name: str) -> Response: - with self.database.session() as conn: - if not conn.get_software_ban(name): - return await AdminSoftwareBans.run("GET", request, error = 'Software ban not found') - - conn.del_software_ban(name) - - return await AdminSoftwareBans.run("GET", request, message = 'Unbanned software') - - @register_route('/admin/users') class AdminUsers(View): async def get(self, @@ -439,37 +241,6 @@ class AdminUsers(View): return Response.new(data, ctype = 'html') - async def post(self, request: Request) -> Response: - data = await request.post() - required_fields = {'username', 'password', 'password2'} - - if not all(data.get(field) for field in required_fields): - return await self.get(request, error = 'Missing username and/or password') - - if data['password'] != data['password2']: - return await self.get(request, error = 'Passwords do not match') - - with self.database.session(True) as conn: - if conn.get_user(data['username']): - return await self.get(request, error = "User already exists") - - conn.put_user(data['username'], data['password'], data['handle']) - - return await self.get(request, message = "Added user") - - -@register_route('/admin/users/delete/{name}') -class AdminUsersDelete(View): - async def get(self, request: Request, name: str) -> Response: - with self.database.session() as conn: - if not conn.get_user(name): - return await AdminUsers.run("GET", request, error = 'User not found') - - conn.del_user(name) - - return await AdminUsers.run("GET", request, message = 'User deleted') - - @register_route('/admin/config') class AdminConfig(View): async def get(self, request: Request, message: str | None = None) -> Response: @@ -483,23 +254,6 @@ class AdminConfig(View): return Response.new(data, ctype = 'html') - async def post(self, request: Request) -> Response: - form = dict(await request.post()) - data = ConfigData() - - for key in ConfigData.USER_KEYS(): - data.set(key, form.get(key.replace('_', '-'))) - - with self.database.session(True) as conn: - for key, value in data.to_dict().items(): - if key in ConfigData.SYSTEM_KEYS(): - continue - - conn.put_config(key, value) - - return await self.get(request, message = 'Updated config') - - @register_route('/style.css') class StyleCss(View): async def get(self, request: Request) -> Response: