removed unused endpoints

This commit is contained in:
Izalia Mae 2024-03-15 19:52:12 -04:00
parent 3114e48371
commit 8b738dd365

View file

@ -153,104 +153,6 @@ class AdminInstances(View):
return Response.new(data, ctype = 'html') 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') @register_route('/admin/whitelist')
class AdminWhitelist(View): class AdminWhitelist(View):
async def get(self, async def get(self,
@ -273,34 +175,6 @@ class AdminWhitelist(View):
return Response.new(data, ctype = 'html') 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') @register_route('/admin/domain_bans')
class AdminDomainBans(View): class AdminDomainBans(View):
async def get(self, async def get(self,
@ -323,42 +197,6 @@ class AdminDomainBans(View):
return Response.new(data, ctype = 'html') 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') @register_route('/admin/software_bans')
class AdminSoftwareBans(View): class AdminSoftwareBans(View):
async def get(self, async def get(self,
@ -381,42 +219,6 @@ class AdminSoftwareBans(View):
return Response.new(data, ctype = 'html') 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') @register_route('/admin/users')
class AdminUsers(View): class AdminUsers(View):
async def get(self, async def get(self,
@ -439,37 +241,6 @@ class AdminUsers(View):
return Response.new(data, ctype = 'html') 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') @register_route('/admin/config')
class AdminConfig(View): class AdminConfig(View):
async def get(self, request: Request, message: str | None = None) -> Response: async def get(self, request: Request, message: str | None = None) -> Response:
@ -483,23 +254,6 @@ class AdminConfig(View):
return Response.new(data, ctype = 'html') 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') @register_route('/style.css')
class StyleCss(View): class StyleCss(View):
async def get(self, request: Request) -> Response: async def get(self, request: Request) -> Response: