117 lines
No EOL
3.7 KiB
Python
117 lines
No EOL
3.7 KiB
Python
import requests, json, argparse, os
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
parser = argparse.ArgumentParser()
|
|
|
|
sp = parser.add_subparsers(dest="list_or_push")
|
|
list_sp = sp.add_parser("list")
|
|
push_sp = sp.add_parser("push")
|
|
clear_sp = sp.add_parser("clear")
|
|
|
|
list_sp.add_argument("-o", "--output")
|
|
push_sp.add_argument("input")
|
|
|
|
list_grp = list_sp.add_mutually_exclusive_group()
|
|
list_grp.add_argument("-a", "--allowed", action="store_true")
|
|
list_grp.add_argument("-b", "--blocked", action="store_true")
|
|
|
|
push_grp = push_sp.add_mutually_exclusive_group()
|
|
push_grp.add_argument("-a", "--allowed", action="store_true")
|
|
push_grp.add_argument("-b", "--blocked", action="store_true")
|
|
|
|
clear_grp = clear_sp.add_mutually_exclusive_group()
|
|
clear_grp.add_argument("-a", "--allowed", action="store_true")
|
|
clear_grp.add_argument("-b", "--blocked", action="store_true")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.list_or_push == "list":
|
|
instance_base_proto = os.getenv("INSTANCE_BASE_PROTO")
|
|
instance_base_url = os.getenv("INSTANCE_BASE_URL")
|
|
instance_full_url = instance_base_proto + "://" + instance_base_url + "/api/iceshrimp/admin/instances/"
|
|
auth_token = os.getenv("AUTH_TOKEN")
|
|
|
|
if args.blocked:
|
|
instance_full_url += "blocked"
|
|
ab = "blocked"
|
|
elif args.allowed:
|
|
instance_full_url += "allowed"
|
|
ab = "allowed"
|
|
else:
|
|
print("specify -a or -b")
|
|
exit()
|
|
|
|
headers = {"authorization": auth_token}
|
|
req = requests.get(instance_full_url, headers=headers)
|
|
|
|
json_dumped = json.loads(req.text)
|
|
|
|
if len(json_dumped) == 0:
|
|
print("no instances are " + ab)
|
|
exit()
|
|
|
|
if args.output:
|
|
x = open(args.output, "w+")
|
|
x.writelines(json.dumps(json_dumped))
|
|
else:
|
|
for instance in json_dumped:
|
|
print(instance)
|
|
elif args.list_or_push == "push":
|
|
instance_base_proto = os.getenv("INSTANCE_BASE_PROTO")
|
|
instance_base_url = os.getenv("INSTANCE_BASE_URL")
|
|
instance_full_url = instance_base_proto + "://" + instance_base_url + "/api/iceshrimp/admin/instances/"
|
|
auth_token = os.getenv("AUTH_TOKEN")
|
|
|
|
if args.blocked:
|
|
ab = "block"
|
|
elif args.allowed:
|
|
ab = "allow"
|
|
else:
|
|
print("specify -a or -b")
|
|
exit()
|
|
|
|
x = open(args.input, "r")
|
|
x_data = x.read()
|
|
|
|
x_json = json.loads(x_data)
|
|
|
|
headers = {"authorization": auth_token}
|
|
ctr = 0
|
|
for instance in x_json:
|
|
ctr += 1
|
|
print(f"({ctr}/{len(x_json)}) {ab}ing {instance["host"]} for {instance["reason"]}")
|
|
instance["imported"] = instance.pop("isImported")
|
|
url = instance_full_url + instance["host"] + "/" + ab
|
|
instance.pop("host")
|
|
req = requests.post(url, headers=headers, data=instance)
|
|
elif args.list_or_push == "clear":
|
|
instance_base_proto = os.getenv("INSTANCE_BASE_PROTO")
|
|
instance_base_url = os.getenv("INSTANCE_BASE_URL")
|
|
instance_full_url = instance_base_proto + "://" + instance_base_url + "/api/iceshrimp/admin/instances/"
|
|
auth_token = os.getenv("AUTH_TOKEN")
|
|
|
|
if args.blocked:
|
|
instance_full_url += "blocked"
|
|
ab = "blocked"
|
|
ab2 = "unblock"
|
|
elif args.allowed:
|
|
instance_full_url += "allowed"
|
|
ab = "allowed"
|
|
ab2 = "disallow"
|
|
else:
|
|
print("specify -a or -b")
|
|
exit()
|
|
|
|
headers = {"authorization": auth_token}
|
|
req = requests.get(instance_full_url, headers=headers)
|
|
|
|
json_dumped = json.loads(req.text)
|
|
|
|
ctr = 0
|
|
for instance in json_dumped:
|
|
ctr += 1
|
|
print(f"({ctr}/{len(json_dumped)}) {ab2}ing {instance["host"]}")
|
|
instance_full_url = instance_base_proto + "://" + instance_base_url + "/api/iceshrimp/admin/instances/" + instance["host"] + "/" + ab2
|
|
req = requests.post(instance_full_url, headers=headers) |