fediverse-blocklist-tool/mastodon_blocklist_deploy/models.py

162 lines
7.2 KiB
Python
Raw Permalink Normal View History

2023-01-11 19:13:27 +00:00
import logging
2023-01-09 10:48:44 +00:00
import requests
class Instance:
def __init__(self, instance_dict):
"""If obfuscate, reject_media or reject_reports are not specified default to False"""
self.obfuscate = False
self.reject_media = False
self.reject_reports = False
self.id = None
"""Remote blocks and local blocks are parsed differently"""
try:
instance_dict["id"]
self.parse_remote_block(instance_dict)
except KeyError:
self.parse_local_block(instance_dict)
def __str__(self):
2023-01-09 12:10:40 +00:00
return f"{self.domain}: {self.severity}"
def __eq__(self, other):
return self.domain == other.domain and self.severity == other.severity and self.reject_media == other.reject_media and self.reject_reports == other.reject_reports and self.obfuscate == other.obfuscate
def status_str(self):
2023-01-12 06:42:23 +00:00
return f"{self.severity}\nReject reports: {self.reject_reports}\nReject media: {self.reject_media}\nObfuscate: {self.obfuscate}"
@property
def exportable_dict(self):
keys = ["domain", "severity", "public_comment", "private_comment", "obfuscate", "reject_media", "reject_reports"]
exportable = {}
for key in keys:
exportable[key] = getattr(self, key)
return exportable
def parse_remote_block(self, instance_dict):
self.domain = instance_dict["domain"]
self.id = instance_dict["id"]
self.severity = instance_dict["severity"]
self.public_comment = instance_dict["public_comment"]
self.private_comment = instance_dict["private_comment"]
2023-01-09 10:48:44 +00:00
self.obfuscate = instance_dict["obfuscate"]
self.reject_media = instance_dict["reject_media"]
self.reject_reports = instance_dict["reject_reports"]
def parse_local_block(self, instance_dict):
try:
self.name = instance_dict["name"]
except KeyError:
pass
self.domain = instance_dict["domain"]
self.severity = instance_dict["severity"]
self.public_comment = instance_dict["public_comment"]
self.private_comment = instance_dict["private_comment"]
2023-01-09 10:48:44 +00:00
try:
self.obfuscate = instance_dict["obfuscate"]
except KeyError:
pass
try:
self.reject_media = instance_dict["reject_media"]
except KeyError:
pass
try:
self.reject_reports = instance_dict["reject_reports"]
except KeyError:
pass
2023-01-09 10:48:44 +00:00
def apply(self, server, token, block_id=None):
2023-01-12 06:46:11 +00:00
"""Applies instance block on the remote server"""
2023-01-09 10:48:44 +00:00
headers = {
f'Authorization': f'Bearer {token}',
}
data = {"domain": self.domain,
"severity": self.severity,
"reject_media": str(self.reject_media).lower(),
"reject_reports": str(self.reject_reports).lower(),
"private_comment": str(self.private_comment).lower(),
2023-01-09 10:48:44 +00:00
"public_comment": self.public_comment,
"obfuscate": str(self.obfuscate).lower()}
2023-01-09 10:48:44 +00:00
"""If no id is given add a new block, else update the existing block"""
if block_id is None:
response = requests.post(f'https://{server}/api/v1/admin/domain_blocks', data=data, headers=headers)
else:
2023-01-11 19:40:05 +00:00
response = requests.put(f'https://{server}/api/v1/admin/domain_blocks/{block_id}', data=data,
headers=headers)
2023-01-09 10:48:44 +00:00
if response.status_code != 200:
raise ConnectionError(f"Could not apply block ({response.status_code}: {response.reason})")
2023-01-11 19:40:05 +00:00
2023-01-11 18:19:14 +00:00
def delete(self, server: str, token: str):
2023-01-12 06:46:11 +00:00
"""Deletes the instance from the blocklist on the remote server"""
2023-01-11 18:19:14 +00:00
headers = {
f'Authorization': f'Bearer {token}',
}
response = requests.delete(f'https://{server}/api/v1/admin/domain_blocks/{self.id}', headers=headers)
if response.status_code != 200:
raise ConnectionError(f"Could not apply block ({response.status_code}: {response.reason})")
@staticmethod
2023-01-09 10:48:44 +00:00
def list_diffs(local_blocklist, remote_blocklist):
2023-01-12 06:46:11 +00:00
"""Compares the local and remote blocklist and returns a list of differences"""
diffs = []
for local_instance in local_blocklist:
instance_found = False
for idx, remote_instance in enumerate(remote_blocklist):
if local_instance.domain == remote_instance.domain:
instance_found = True
if local_instance == remote_instance:
pass
else:
"""If the local block is different from the remote block, add it to the diff"""
diffs.append({"local": local_instance, "remote": remote_instance})
"""Remove the remote instance from the list so we later have a list of remote instances we don't
have locally"""
del remote_blocklist[idx]
"""If the local instance is not in the remote blocklist, add it to the diff"""
if not instance_found:
diffs.append({"local": local_instance, "remote": None})
for remote_instance in remote_blocklist:
diffs.append({"local": None, "remote": remote_instance})
return diffs
@staticmethod
2023-01-11 19:21:01 +00:00
def apply_blocks_from_diff(diffs, server, token, no_delete: bool):
2023-01-12 06:46:11 +00:00
"""Uses a diff (list of difference in local an remote instance) to apply instance blocks"""
2023-01-09 10:48:44 +00:00
for diff in diffs:
2023-01-11 19:39:46 +00:00
if diff["local"] is None:
if not no_delete:
"""Delete the block on the remote server"""
diff['remote'].delete(server, token)
logging.info(f"Deleted {diff['remote'].domain} from blocklist")
2023-01-09 10:48:44 +00:00
elif diff["remote"] is None:
2023-01-11 19:13:27 +00:00
"""Add the block on the remote server"""
2023-01-09 10:48:44 +00:00
diff["local"].apply(server, token)
2023-01-11 20:27:20 +00:00
logging.info(f"Added {diff['local'].domain} to blocklist")
2023-01-09 10:48:44 +00:00
else:
2023-01-11 19:13:27 +00:00
"""Update the block on the remote server"""
2023-01-09 10:48:44 +00:00
diff["local"].apply(server, token, block_id=diff["remote"].id)
2023-01-11 20:27:20 +00:00
logging.info(f"Updated {diff['local'].domain} in blocklist")
2023-01-09 10:48:44 +00:00
@staticmethod
def show_diffs(local_blocklist, remote_blocklist):
2023-01-12 06:46:11 +00:00
"""Shows a table in the CLI comparing the local and remote blocklist"""
from rich.table import Table
from rich.console import Console
table = Table(title="Differences", expand=True, show_lines=True)
table.add_column("Domain", style="cyan")
table.add_column("Local status", style="green")
2023-01-12 06:42:23 +00:00
table.add_column("Current remote status", style="magenta")
2023-01-09 10:48:44 +00:00
diffs = Instance.list_diffs(local_blocklist, remote_blocklist)
for diff in diffs:
if diff["local"] is None:
table.add_row(diff["remote"].domain, None, diff["remote"].status_str())
elif diff["remote"] is None:
table.add_row(diff["local"].domain, diff["local"].status_str(), None)
else:
table.add_row(diff["local"].domain, diff["local"].status_str(), diff["remote"].status_str())
console = Console()
2023-01-09 10:48:44 +00:00
console.print(table)