Use of Mastodonplus.py's admin_ip_blocks endpoint

master
spla 3 months ago
parent a849179aa9
commit 4c4ebaef77
  1. 1
      requirements.txt
  2. 252
      spamcheck.py

@ -1,2 +1,3 @@
psycopg2-binary
requests
Mastodonplus.py

@ -1,5 +1,6 @@
import datetime
from datetime import date, datetime, timedelta
#from mastodon import Mastodon
from mastodon import Mastodon, MastodonNetworkError, MastodonNotFoundError, MastodonInternalServerError
import time
import os
import json
@ -8,15 +9,17 @@ import os.path
import operator
import psycopg2
from psycopg2 import sql
import requests
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
class Spamcheck:
name = "Spamcheck for Mastodon social servers"
name = "Spamcheck for Mastodon social server"
def __init__(self, mastodon_hostname=None, mastodon_db=None, mastodon_db_user=None, spamcheck_db=None, spamcheck_db_user=None):
self.config_file = 'config/config.txt'
self.secrets_file = "secrets/secrets.txt"
is_setup = self.__check_setup(self)
@ -27,7 +30,10 @@ class Spamcheck:
self.mastodon_db_user = self.__get_parameter("mastodon_db_user", self.config_file)
self.spamcheck_db = self.__get_parameter("spamcheck_db", self.config_file)
self.spamcheck_db_user = self.__get_parameter("spamcheck_db_user", self.config_file)
self.__uc_client_id = self.__get_parameter("uc_client_id", self.secrets_file)
self.__uc_client_secret = self.__get_parameter("uc_client_secret", self.secrets_file)
self.__uc_access_token = self.__get_parameter("uc_access_token", self.secrets_file)
else:
self.mastodon_hostname, self.mastodon_db, self.mastodon_db_user, self.spamcheck_db, self.spamcheck_db_user = self.__setup(self)
@ -39,7 +45,7 @@ class Spamcheck:
self.__createdb(self)
def new_registers(self, created_at_lst=[], id_lst=[], email_lst=[], ip_lst=[]):
try:
conn = None
@ -48,7 +54,7 @@ class Spamcheck:
cur = conn.cursor()
cur.execute("select users.created_at, users.id, users.email, users.sign_up_ip from users where users.created_at > now() - interval '7 days'")
cur.execute("select users.created_at, users.id, users.email, users.sign_up_ip from users where users.created_at > now() - interval '70 days'")
rows = cur.fetchall()
@ -266,6 +272,184 @@ class Spamcheck:
conn.close()
def get_tor(self):
tor_list = []
try:
conn = None
conn = psycopg2.connect(database = self.spamcheck_db, user = self.spamcheck_db_user, password = "", host = "/var/run/postgresql", port = "5432")
cur = conn.cursor()
cur.execute("select ip from spamcheck where tor_exit_node")
rows = cur.fetchall()
for row in rows:
tor_list.append(row[0])
cur.close()
return tor_list
except (Exception, psycopg2.DatabaseError) as error:
print (error)
finally:
if conn is not None:
conn.close()
def ip_blocks(self, ip):
'''
severity:
Limit sign-ups : 5000
Block sign-ups : 5500
Block access : 9999
'''
data = {
'ip': ip,
'severity': 5500,
'comment': 'Spam from this Tor exit node',
'expires_in': 8600,
}
endpoint = f'https://{self.mastodon_hostname}/api/v1/admin/ip_blocks'
response = self.api_request('POST', endpoint, data)
if response.ok:
data = response.json()
else:
pass
def api_request(self, method, endpoint, data={}):
session = requests.Session()
response = None
try:
kwargs = dict(data=data)
response = session.request(method, url = endpoint, headers = headers, **kwargs)
except Exception as e:
raise MastodonNetworkError(f"Could not complete request: {e}")
if response is None:
raise MastodonIllegalArgumentError("Illegal request.")
if not response.ok:
try:
if isinstance(response, dict) and 'error' in response:
error_msg = response['error']
elif isinstance(response, str):
error_msg = response
else:
error_msg = None
except ValueError:
error_msg = None
if response.status_code == 404:
ex_type = MastodonNotFoundError
if not error_msg:
error_msg = 'Endpoint not found.'
# this is for compatibility with older versions
# which raised MastodonAPIError('Endpoint not found.')
# on any 404
elif response.status_code == 401:
ex_type = MastodonUnauthorizedError
elif response.status_code == 422:
return response
elif response.status_code == 500:
ex_type = MastodonInternalServerError
elif response.status_code == 502:
ex_type = MastodonBadGatewayError
elif response.status_code == 503:
ex_type = MastodonServiceUnavailableError
elif response.status_code == 504:
ex_type = MastodonGatewayTimeoutError
elif response.status_code >= 500 and \
response.status_code <= 511:
ex_type = MastodonServerError
else:
ex_type = MastodonAPIError
raise ex_type(
'Mastodon API returned error',
response.status_code,
response.reason,
error_msg)
else:
return response
def blocked_ip_list(self):
blocked_ip_list = []
temp_ip_list = {}
i = 0
while True:
if i == 0:
temp_ip_list[i] = mastodon.admin_ip_blocks_list(limit=200)
if temp_ip_list[i] == []:
return blocked_ip_list
else:
temp_ip_list[i] = mastodon.fetch_next(temp_ip_list[i-1]._pagination_next)
dict_len = len(temp_ip_list[i])
ii = 0
while ii - dict_len:
blocked_ip_list.append(temp_ip_list[i][ii])
ii += 1
if len(temp_ip_list[i]) < 200:
return blocked_ip_list
i += 1
def ip_blocked(self, ip, blocked_list):
is_blocked = False
for block_ip_item in blocked_list:
if ip in block_ip_item.ip:
print(f'\n{ip} is already blocked')
is_blocked = True
return (is_blocked)
@staticmethod
def __check_ip(self, ip):
@ -437,6 +621,25 @@ class Spamcheck:
conn.close()
def log_in(self):
uc_client_id = self.__get_parameter("uc_client_id", self.secrets_file)
uc_client_secret = self.__get_parameter("uc_client_secret", self.secrets_file)
uc_access_token = self.__get_parameter("uc_access_token", self.secrets_file)
self.mastodon_hostname = self.__get_parameter("mastodon_hostname", self.config_file)
mastodon = Mastodon(
client_id = uc_client_id,
client_secret = uc_client_secret,
access_token = uc_access_token,
api_base_url = 'https://' + self.mastodon_hostname,
)
headers={ 'Authorization': 'Bearer %s'%uc_access_token }
return (mastodon, self.mastodon_hostname, headers)
@staticmethod
def __get_parameter(parameter, file_path ):
@ -454,7 +657,9 @@ class Spamcheck:
if __name__ == '__main__':
spamcheck = Spamcheck()
mastodon, mastodon_hostname, headers = spamcheck.log_in()
created_at_lst, id_lst, email_lst, ip_lst = spamcheck.new_registers()
spamcheck.save_registers(created_at_lst, id_lst, email_lst, ip_lst)
@ -462,3 +667,38 @@ if __name__ == '__main__':
spamcheck_datetime_lst, spamcheck_registers_lst = spamcheck.get_totals()
spamcheck.write_totals(spamcheck_datetime_lst, spamcheck_registers_lst)
tor_list = spamcheck.get_tor()
print(f'\nTor IPs: {len(tor_list)}')
blocked_list = spamcheck.blocked_ip_list()
print(f'\nalready blocked IPs: {len(blocked_list)}')
for ip in tor_list:
is_blocked = spamcheck.ip_blocked(ip, blocked_list)
if not is_blocked:
print(f'\n** blocking IP {ip}')
severity = 'sign_up_requires_approval'
comment = 'Tor exit node'
expires_in = 86400
result = mastodon.admin_ip_blocks_create(ip, severity, comment, expires_in)
blocked_list.append(result)
print(f'\nalready blocked IPs: {len(blocked_list)}')
print(f'\nRate limit: {mastodon.ratelimit_remaining} of {mastodon.ratelimit_limit}\nNext reset: {datetime.fromtimestamp(mastodon.ratelimit_reset).isoformat()}\n')
if mastodon.ratelimit_remaining < 10:
print(f'Rate limit is lower than 10! Sleeping by 5 minutes...')
time.sleep(300)
time.sleep(1)

Loading…
Cancel
Save