127 lines
7.0 KiB
Python
127 lines
7.0 KiB
Python
import os
|
|
from flask import request
|
|
from flask_restx import Namespace, Resource, reqparse
|
|
from overseer.flask import cache, db
|
|
from overseer.observer import retrieve_suspicious_instances
|
|
from loguru import logger
|
|
from overseer.classes.instance import Instance
|
|
from overseer.database import functions as database
|
|
from overseer import exceptions as e
|
|
from overseer.utils import hash_api_key
|
|
from overseer.lemmy import pm_new_api_key
|
|
from pythorhead import Lemmy
|
|
|
|
api = Namespace('v1', 'API Version 1' )
|
|
|
|
from overseer.apis.models.v1 import Models
|
|
|
|
models = Models(api)
|
|
|
|
handle_bad_request = api.errorhandler(e.BadRequest)(e.handle_bad_requests)
|
|
handle_forbidden = api.errorhandler(e.Forbidden)(e.handle_bad_requests)
|
|
|
|
# Used to for the flask limiter, to limit requests per url paths
|
|
def get_request_path():
|
|
# logger.info(dir(request))
|
|
return f"{request.remote_addr}@{request.method}@{request.path}"
|
|
|
|
|
|
class Suspicions(Resource):
|
|
get_parser = reqparse.RequestParser()
|
|
get_parser.add_argument("Client-Agent", default="unknown:0:unknown", type=str, required=False, help="The client name and version.", location="headers")
|
|
get_parser.add_argument("activity_suspicion", required=False, default=20, type=int, help="How many users per local post+comment to consider suspicious", location="args")
|
|
get_parser.add_argument("csv", required=False, type=bool, help="Set to true to return just the domains as a csv. Mutually exclusive with domains", location="args")
|
|
get_parser.add_argument("domains", required=False, type=bool, help="Set to true to return just the domains as a list. Mutually exclusive with csv", location="args")
|
|
|
|
@api.expect(get_parser)
|
|
@logger.catch(reraise=True)
|
|
@cache.cached(timeout=10, query_string=True)
|
|
@api.marshal_with(models.response_model_model_Suspicions_get, code=200, description='Suspicious Instances', skip_none=True)
|
|
def get(self):
|
|
'''A List with the details of all suspicious instances
|
|
'''
|
|
self.args = self.get_parser.parse_args()
|
|
sus_instances = retrieve_suspicious_instances(self.args.activity_suspicion)
|
|
if self.args.csv:
|
|
return {"csv": ",".join([instance["domain"] for instance in sus_instances])},200
|
|
if self.args.domains:
|
|
return {"domains": [instance["domain"] for instance in sus_instances]},200
|
|
return {"instances": sus_instances},200
|
|
|
|
|
|
class Whitelist(Resource):
|
|
get_parser = reqparse.RequestParser()
|
|
get_parser.add_argument("Client-Agent", default="unknown:0:unknown", type=str, required=False, help="The client name and version.", location="headers")
|
|
get_parser.add_argument("endorsements", required=False, default=1, type=int, help="Limit to this amount of endorsements of more", location="args")
|
|
get_parser.add_argument("domain", required=False, type=str, help="Filter by instance domain", location="args")
|
|
get_parser.add_argument("csv", required=False, type=bool, help="Set to true to return just the domains as a csv. Mutually exclusive with domains", location="args")
|
|
get_parser.add_argument("domains", required=False, type=bool, help="Set to true to return just the domains as a list. Mutually exclusive with csv", location="args")
|
|
|
|
@api.expect(get_parser)
|
|
@cache.cached(timeout=10, query_string=True)
|
|
@api.marshal_with(models.response_model_model_Whitelist_get, code=200, description='Instances', skip_none=True)
|
|
def get(self):
|
|
'''A List with the details of all instances and their endorsements
|
|
'''
|
|
self.args = self.get_parser.parse_args()
|
|
instance_details = []
|
|
for instance in database.get_all_instances():
|
|
instance_details.append(instance.get_details())
|
|
if self.args.csv:
|
|
return {"csv": ",".join([instance["domain"] for instance in instance_details])},200
|
|
if self.args.domains:
|
|
return {"domains": [instance["domain"] for instance in instance_details]},200
|
|
return {"instances": instance_details},200
|
|
|
|
|
|
put_parser = reqparse.RequestParser()
|
|
put_parser.add_argument("Client-Agent", default="unknown:0:unknown", type=str, required=False, help="The client name and version.", location="headers")
|
|
put_parser.add_argument("domain", required=False, type=str, help="The instance domain. It MUST be alredy registered in https://overctrl.dbzer0.com", location="json")
|
|
put_parser.add_argument("guarantor", required=False, type=str, help="(Optiona) The domain of the guaranteeing instance. They will receive a PM to validate you", location="json")
|
|
|
|
|
|
@api.expect(put_parser)
|
|
@api.marshal_with(models.response_model_instances, code=200, description='Instances')
|
|
def put(self):
|
|
'''A List with the details of all instances and their endorsements
|
|
'''
|
|
self.args = self.put_parser.parse_args()
|
|
existing_instance = Instance.query.filter_by(domain=self.args.domain).first()
|
|
if existing_instance:
|
|
return existing_instance.get_details(),200
|
|
requested_lemmy = Lemmy(f"https://{self.args.domain}")
|
|
site = requested_lemmy.site.get()
|
|
api_key = pm_new_api_key(self.args.domain)
|
|
if not api_key:
|
|
raise e.BadRequest("Failed to generate API Key")
|
|
new_instance = Instance(
|
|
domain=self.args.domain,
|
|
api_key=hash_api_key(api_key),
|
|
open_registrations=site["site_view"]["local_site"]["registration_mode"] == "open",
|
|
email_verify=site["site_view"]["local_site"]["require_email_verification"],
|
|
software=requested_lemmy.nodeinfo['software']['name'],
|
|
)
|
|
new_instance.create()
|
|
return new_instance.get_details(),200
|
|
|
|
patch_parser = reqparse.RequestParser()
|
|
patch_parser.add_argument("Authorization: apikey", type=str, required=True, help="The sending instance's API key.", location='headers')
|
|
patch_parser.add_argument("Client-Agent", default="unknown:0:unknown", type=str, required=False, help="The client name and version.", location="headers")
|
|
patch_parser.add_argument("domain", required=False, type=str, help="The instance domain. It MUST be alredy registered in https://overctrl.dbzer0.com", location="json")
|
|
patch_parser.add_argument("guarantor", required=False, type=str, help="(Optiona) The domain of the guaranteeing instance. They will receive a PM to validate you", location="json")
|
|
|
|
|
|
@api.expect(put_parser)
|
|
@logger.catch(reraise=True)
|
|
@api.marshal_with(models.response_model_instances, code=200, description='Instances', skip_none=True)
|
|
def patch(self):
|
|
'''A List with the details of all instances and their endorsements
|
|
'''
|
|
self.args = self.patch_parser.parse_args()
|
|
self.apikey = self.args["Authorization: apikey"]
|
|
if not self.apikey:
|
|
raise e.BadRequest("You must provide the API key that was PM'd to your overctrl.dbzer0.com account")
|
|
existing_instance = Instance.query.filter_by(domain=self.args.domain).first()
|
|
if existing_instance:
|
|
return existing_instance.get_details,200
|