spla/blocksoft
1
0
Fork 0
Codi Incidències Sol·licituds Pull repo.project Llançaments Paquets Wiki Activitat

First functional release

Explorar la Font
This commit is contained in:
spla 2022-08-21 20:00:02 +02:00
pare 49e918a914
commit bc423b9e0d
S'han modificat 4 arxius amb 1080 adicions i 249 eliminacions
Mostrar estadístiques Descarregar el fitxer de pedaç Descarregar el fitxer de diferències Expand all files Collapse all files

22
README.md Normal file
Veure arxiu

@ -0,0 +1,22 @@
# BlockSoft
Get all the running software of the peers of your Mastodon server and block the software you want.
### Dependencies
- **Python 3**
- Postgresql server
- Mastodon server admin access token
### Usage:
Within Python Virtual Environment:
1. Run `pip install -r requirements.txt` to install needed Python libraries.
3. Run `python peers.py` to collect and store all needed data from your server peers..
4. Run `python blocksoft.py` to block the configured software..

493
blocksoft.py Normal file
Veure arxiu

@ -0,0 +1,493 @@
import os
import sys
from mastodon import Mastodon
from mastodon.Mastodon import MastodonNotFoundError, MastodonNetworkError, MastodonReadTimeout, MastodonAPIError, MastodonUnauthorizedError, MastodonIllegalArgumentError
import psycopg2
from psycopg2 import sql
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
import getpass
import fileinput,re
import requests
import pdb
class DomainBlocks():
name = 'Domain blocks for Mastodon'
def __init__(self, mastodon_hostname=None, domain_block_api=None, session=None):
self.domain_blocks_api = '/api/v1/admin/domain_blocks'
self.config_file = "config/config.txt"
self.secrets_file = "secrets/secrets.txt"
is_setup = self.__check_setup(self.secrets_file)
if is_setup:
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)
self.blocker_db, self.blocker_db_user = self.db_config()
self.mastodon, self.mastodon_hostname, self.headers = self.log_in()
else:
while(True):
logged_in, self.mastodon, self.mastodon_hostname = self.setup()
if not logged_in:
print("\nLog in failed! Try again.\n")
else:
break
db_setup = self.__check_dbsetup(self)
if not db_setup:
self.__createdb(self)
if session:
self.session = session
else:
self.session = requests.Session()
def get_servers(self, software):
servers_list = []
try:
conn = None
conn = psycopg2.connect(database = self.blocker_db, user = self.blocker_db_user, password = "", host = "/var/run/postgresql", port = "5432")
cur = conn.cursor()
cur.execute("select server from blocker where software=(%s)", (software,))
rows = cur.fetchall()
for row in rows:
servers_list.append(row[0])
cur.close()
except (Exception, psycopg2.DatabaseError) as error:
sys.exit(error)
finally:
if conn is not None:
conn.close()
return (servers_list)
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)
def domain_blocks_create(self, server):
data = {
'domain': server,
'severity': 'suspend',
'reject_media': 'true',
'reject_reports': 'true',
'obfuscate': 'true',
}
endpoint = f'https://{self.mastodon_hostname}/{self.domain_blocks_api}'
response = self.__api_request('POST', endpoint, data)
if response.ok:
print(f"Done, {server} is now suspended")
else:
pass
@staticmethod
def __check_setup(file_path):
is_setup = False
if not os.path.isfile(file_path):
print(f"File {file_path} not found, running setup.")
return
else:
is_setup = True
return is_setup
def setup(self):
logged_in = False
try:
self.mastodon_hostname = input("Enter Mastodon hostname (or 'q' to exit): ")
if self.mastodon_hostname == 'q':
sys.exit("Bye")
user_name = input("Bot user name, ex. john? ")
user_password = getpass.getpass("Bot password? ")
app_name = input("Bot App name? ")
self.blocker_db = input("Blocker's database name: ")
self.blocker_db_user = input("blocker's database user: ")
client_id, client_secret = Mastodon.create_app(
app_name,
scopes = ["read", "write", "admin:read", "admin:write"],
to_file="app_clientcred.txt",
api_base_url=self.mastodon_hostname
)
mastodon = Mastodon(client_id = "app_clientcred.txt", api_base_url = self.mastodon_hostname)
mastodon.log_in(
user_name,
user_password,
scopes = ["read", "write", "admin:read", "admin:write"],
to_file = "app_usercred.txt"
)
if os.path.isfile("app_usercred.txt"):
print(f"Log in succesful!")
logged_in = True
if not os.path.exists('secrets'):
os.makedirs('secrets')
if not os.path.exists(self.secrets_file):
with open(self.secrets_file, 'w'): pass
print(f"{self.secrets_file} created!")
with open(self.secrets_file, 'a') as the_file:
print(f"Writing secrets parameter names to {self.secrets_file}")
the_file.write('uc_client_id: \n'+'uc_client_secret: \n'+'uc_access_token: \n')
client_path = 'app_clientcred.txt'
with open(client_path) as fp:
line = fp.readline()
cnt = 1
while line:
if cnt == 1:
print(f"Writing client id to {self.secrets_file}")
self.__modify_file(self, self.secrets_file, "uc_client_id: ", value=line.rstrip())
elif cnt == 2:
print(f"Writing client secret to {self.secrets_file}")
self.__modify_file(self, self.secrets_file, "uc_client_secret: ", value=line.rstrip())
line = fp.readline()
cnt += 1
token_path = 'app_usercred.txt'
with open(token_path) as fp:
line = fp.readline()
print(f"Writing access token to {self.secrets_file}")
self.__modify_file(self, self.secrets_file, "uc_access_token: ", value=line.rstrip())
if os.path.exists("app_clientcred.txt"):
print("Removing app_clientcred.txt temp file..")
os.remove("app_clientcred.txt")
if os.path.exists("app_usercred.txt"):
print("Removing app_usercred.txt temp file..")
os.remove("app_usercred.txt")
self.__create_config(self)
self.__write_config(self)
print("Secrets setup done!\n")
with open(self.config_file, 'a') as the_file:
print(f"Writing Mastodon hostname parameter to {self.config_file}")
the_file.write(f'blocker_db: {self.blocker_db}\n')
the_file.write(f'blocker_db_user: {self.blocker_db_user}\n')
print("Blocker setup done!\n")
except MastodonIllegalArgumentError as i_error:
sys.stdout.write(f'\n{str(i_error)}\n')
except MastodonNetworkError as n_error:
sys.stdout.write(f'\n{str(n_error)}\n')
except MastodonReadTimeout as r_error:
sys.stdout.write(f'\n{str(r_error)}\n')
except MastodonAPIError as a_error:
sys.stdout.write(f'\n{str(a_error)}\n')
return (logged_in, mastodon, self.mastodon_hostname)
def __api_request(self, method, endpoint, data={}):
response = None
try:
kwargs = dict(data=data)
response = self.session.request(method, url = endpoint, headers = self.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 db_config(self):
self.blocker_db = self.__get_parameter("blocker_db", self.config_file)
self.blocker_db_user = self.__get_parameter("blocker_db_user", self.config_file)
return (self.blocker_db, self.blocker_db_user)
@staticmethod
def __check_dbsetup(self):
dbsetup = False
try:
conn = None
conn = psycopg2.connect(database = self.blocker_db, user = self.blocker_db_user, password = "", host = "/var/run/postgresql", port = "5432")
dbsetup = True
except (Exception, psycopg2.DatabaseError) as error:
print(error)
return dbsetup
@staticmethod
def __createdb(self):
conn = None
try:
conn = psycopg2.connect(dbname='postgres',
user=self.blocker_db_user, host='',
password='')
conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
cur = conn.cursor()
print(f"Creating database {self.blocker_db}. Please wait...")
cur.execute(sql.SQL("CREATE DATABASE {}").format(
sql.Identifier(self.blocker_db))
)
print(f"Database {self.blocker_db} created!\n")
self.__dbtables_schemes(self)
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
@staticmethod
def __dbtables_schemes(self):
table = "blocker"
sql = "create table "+table+" (server varchar(200), users int, updated_at timestamptz, software varchar(50), alive boolean, users_api varchar(50), version varchar(100), first_checked_at timestamptz, last_checked_at timestamptz, downs int)"
self.__create_table(self, table, sql)
table = "execution_time"
sql = "create table "+table+" (program varchar(30), start timestamptz, finish timestamptz)"
self.__create_table(self, table, sql)
@staticmethod
def __create_table(self, table, sql):
conn = None
try:
conn = psycopg2.connect(database = self.blocker_db, user = self.blocker_db_user, password = "", host = "/var/run/postgresql", port = "5432")
cur = conn.cursor()
print(f"Creating table {table}")
cur.execute(sql)
conn.commit()
print(f"Table {table} created!\n")
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
def __get_parameter(self, parameter, config_file):
if not os.path.isfile(config_file):
print(f"File {config_file} not found, exiting.")
sys.exit(0)
with open( config_file ) as f:
for line in f:
if line.startswith( parameter ):
return line.replace(parameter + ":", "").strip()
print(f"{config_file} Missing parameter {parameter}")
sys.exit(0)
@staticmethod
def __modify_file(self, file_name, pattern,value=""):
fh=fileinput.input(file_name,inplace=True)
for line in fh:
replacement=pattern + value
line=re.sub(pattern,replacement,line)
sys.stdout.write(line)
fh.close()
@staticmethod
def __create_config(self):
if not os.path.exists('config'):
os.makedirs('config')
if not os.path.exists(self.config_file):
print(self.config_file + " created!")
with open('config/config.txt', 'w'): pass
@staticmethod
def __write_config(self):
with open(self.config_file, 'a') as the_file:
the_file.write(f'mastodon_hostname: {self.mastodon_hostname}\n')
print(f"adding parameter 'mastodon_hostname' to {self.config_file}\n")
###############################################################################
# main
if __name__ == '__main__':
blocker = DomainBlocks()
software = 'birdsitelive'
servers_list = blocker.get_servers(software)
for server in servers_list:
blocker.domain_blocks_create(server)

565
peers.py Normal file
Veure arxiu

@ -0,0 +1,565 @@
import os
import sys
import time
from datetime import datetime
import requests
import urllib3
import json
import psycopg2
from psycopg2 import sql
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
#import ray
import pdb
#ray.init(num_cpus = 32) # Specify this system CPUs.
class Peers:
name = 'Mastodon server peers'
def __init__(self, mastodon_hostname=None, peers_api=None):
self.peers_api = '/api/v1/instance/peers'
self.config_path = "config/config.txt"
self.apis = ['/api/v1/instance?',
'/nodeinfo/2.0?',
'/nodeinfo/2.0.json?',
'/main/nodeinfo/2.0?',
'/api/statusnet/config?',
'/api/nodeinfo/2.0.json?',
'/api/nodeinfo?',
'/wp-json/nodeinfo/2.0?',
'/api/v1/instance/nodeinfo/2.0?',
'/.well-known/x-nodeinfo2?'
]
is_setup = self.__check_setup(self)
if is_setup:
self.mastodon_hostname = self.__get_parameter("mastodon_hostname", self.config_path)
self.blocker_db = self.__get_parameter("blocker_db", self.config_path)
self.blocker_db_user = self.__get_parameter("blocker_db_user", self.config_path)
else:
self.mastodon_hostname, self.blocker_db, self.blocker_db_user = self.__setup(self)
db_setup = self.__check_dbsetup(self)
if not db_setup:
self.__createdb(self)
def get_peers(self):
user_agent = {'User-agent': "fediverse's stats (fediverse@soc.catala.digital)"}
res = requests.get(f'https://{self.mastodon_hostname}{self.peers_api}', headers = user_agent, timeout=3)
peers = res.json()
return peers
def getsoft(self, server):
if server.find(".") == -1:
return
if server.find("@") != -1:
return
if server.find("/") != -1:
return
if server.find(":") != -1:
return
soft = ''
is_nodeinfo = False
url = 'https://' + server
user_agent = {'User-agent': "fediverse's stats (fediverse@soc.catala.digital)"}
try:
response = requests.get(url + '/.well-known/nodeinfo', headers = user_agent, timeout=3)
if response.status_code == 200:
try:
response_json = response.json()
nodeinfo = response_json['links'][0]['href'].replace(f'https://{server}','')
try:
nodeinfo_data = requests.get(url + nodeinfo, headers = user_agent, timeout=3)
if nodeinfo_data.status_code == 200:
nodeinfo_json = nodeinfo_data.json()
is_nodeinfo = True
else:
print(f"Server {server}'s nodeinfo not responding: error code {nodeinfo_data.status_code}")
except:
pass
except:
print(f'Server {server} not responding: error code {response.status_code}')
print('*********************************************************************')
pass
else:
for api in self.apis:
try:
response = requests.get(url + api, headers = user_agent, timeout=3)
if is_json(response.text):
nodeinfo_json = response.json()
if 'software' in nodeinfo_json:
nodeinfo = api
is_nodeinfo = True
break
elif 'title' in nodeinfo_json:
if nodeinfo_json['title'] == 'Zap':
nodeinfo = api
is_nodeinfo = True
soft = 'zap'
break
elif 'version' in nodeinfo_json:
nodeinfo = api
is_nodeinfo = True
break
except:
pass
except requests.exceptions.SSLError as errssl:
pass
except requests.exceptions.HTTPError as errh:
pass
except requests.exceptions.ConnectionError as errc:
pass
except requests.exceptions.ReadTimeout as to_err:
pass
except requests.exceptions.TooManyRedirects as tmr_err:
pass
except urllib3.exceptions.LocationParseError as lp_err:
pass
except requests.exceptions.InvalidURL as iu_err:
pass
except requests.exceptions.ChunkedEncodingError as chunk_err:
print(f'ChunkedEncodingError! {server}')
pass
except ray.exceptions.RaySystemError as ray_sys_error:
print(ray_sys_error)
pass
else:
if is_nodeinfo:
if nodeinfo != '/api/v1/instance?':
if nodeinfo != '/.well-known/x-nodeinfo2?':
try:
soft = nodeinfo_json['software']['name']
soft = soft.lower()
soft_version = nodeinfo_json['software']['version']
users = nodeinfo_json['usage']['users']['total']
alive = True
self.write_api(server, soft, users, alive, nodeinfo, soft_version)
print(f"Server {server} ({soft} {soft_version}) is alive!")
print('*********************************************************************')
return
except:
pass
else:
try:
soft = nodeinfo_json['server']['software']
soft = soft.lower()
soft_version = nodeinfo_json['server']['version']
users = nodeinfo_json['usage']['users']['total']
alive = True
if soft == 'socialhome':
self.write_api(server, soft, users, alive, nodeinfo, soft_version)
print('*********************************************************************')
print(f"Server {serve}r ({soft} {soft_version}) is alive!")
print('*********************************************************************')
return
except:
pass
if soft == '' and nodeinfo == "/api/v1/instance?":
soft = 'mastodon'
try:
users = nodeinfo_json['stats']['user_count']
if users > 1000000:
return
except:
users = 0
try:
soft_version = nodeinfo_json['version']
except:
soft_version = 'unknown'
alive = True
self.write_api(self, server, soft, users, alive, nodeinfo, soft_version)
print('*********************************************************************')
print(f"Server {server} ({soft}) is alive!")
elif soft == 'zap' and nodeinfo == "/api/v1/instance?":
soft = 'zap'
users = nodeinfo_json['stats']['user_count']
soft_version = nodeinfo_json['version']
alive = True
print(server, soft, users, alive, api)
print('*********************************************************************')
print(f"Server {server} ({soft}) is alive!")
else:
print(f'Server {server} is dead')
print('*********************************************************************')
def write_api(self, server, software, users, alive, api, soft_version):
insert_sql = "INSERT INTO blocker(server, updated_at, software, users, alive, users_api, version) VALUES(%s,%s,%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING"
conn = None
try:
conn = psycopg2.connect(database=self.blocker_db, user=self.blocker_db_user, password="", host="/var/run/postgresql", port="5432")
cur = conn.cursor()
print(f'Writing {server} nodeinfo data...')
cur.execute(insert_sql, (server, now, software, users, alive, api, soft_version))
cur.execute(
"UPDATE blocker SET updated_at=(%s), software=(%s), users=(%s), alive=(%s), users_api=(%s), version=(%s) where server=(%s)",
(now, software, users, alive, api, soft_version, server)
)
conn.commit()
cur.close()
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
def save_time(self, program, start, finish):
insert_sql = "INSERT INTO execution_time(program, start, finish) VALUES(%s,%s,%s) ON CONFLICT DO NOTHING"
conn = None
try:
conn = psycopg2.connect(database = self.blocker_db, user = self.blocker_db_user, password = "", host = "/var/run/postgresql", port = "5432")
cur = conn.cursor()
cur.execute(insert_sql, (program, start, finish,))
cur.execute("UPDATE execution_time SET start=(%s), finish=(%s) where program=(%s)", (start, finish, program))
conn.commit()
cur.close()
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
@staticmethod
def __check_setup(self):
is_setup = False
if not os.path.isfile(self.config_path):
print(f"File {self.config_path} not found, running setup.")
else:
is_setup = True
return is_setup
@staticmethod
def __setup(self):
if not os.path.exists('config'):
os.makedirs('config')
self.mastodon_hostname = input("Mastodon hostname: ")
self.blocker_db = input("Blocker database name: ")
self.blocker_db_user = input("Blocker database user: ")
if not os.path.exists(self.config_path):
with open(self.config_path, 'w'): pass
print(f"{self.config_path} created!")
with open(self.config_path, 'a') as the_file:
print("Writing Blocker parameters to " + self.config_path)
the_file.write(f'mastodon_hostname: {self.mastodon_hostname}\n')
the_file.write(f'blocker_db: {self.blocker_db}\n')
the_file.write(f'blocker_db_user: {self.blocker_db_user}\n')
return (self.mastodon_hostname, self.blocker_db, self.blocker_db_user)
@staticmethod
def __check_dbsetup(self):
dbsetup = False
try:
conn = None
conn = psycopg2.connect(database = self.blocker_db, user = self.blocker_db_user, password = "", host = "/var/run/postgresql", port = "5432")
dbsetup = True
except (Exception, psycopg2.DatabaseError) as error:
print(error)
return dbsetup
@staticmethod
def __createdb(self):
conn = None
try:
conn = psycopg2.connect(dbname='postgres',
user=self.blocker_db_user, host='',
password='')
conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
cur = conn.cursor()
print(f"Creating database {self.blocker_db}. Please wait...")
cur.execute(sql.SQL("CREATE DATABASE {}").format(
sql.Identifier(self.blocker_db))
)
print(f"Database {self.blocker_db} created!\n")
self.__dbtables_schemes(self)
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
@staticmethod
def __dbtables_schemes(self):
table = "blocker"
sql = f"""create table {table} (server varchar(200) PRIMARY KEY, users INT, updated_at timestamptz, software varchar(50), alive boolean, users_api varchar(50),
version varchar(100), first_checked_at timestamptz, last_checked_at timestamptz, downs int)"""
self.__create_table(self, table, sql)
table = "execution_time"
sql = "create table "+table+" (program varchar(30) PRIMARY KEY, start timestamptz, finish timestamptz)"
self.__create_table(self, table, sql)
@staticmethod
def __create_table(self, table, sql):
conn = None
try:
conn = psycopg2.connect(database = self.blocker_db, user = self.blocker_db_user, password = "", host = "/var/run/postgresql", port = "5432")
cur = conn.cursor()
print(f"Creating table {table}")
cur.execute(sql)
conn.commit()
print(f"Table {table} created!\n")
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
@staticmethod
def __get_parameter(parameter, file_path ):
with open( file_path ) as f:
for line in f:
if line.startswith( parameter ):
return line.replace(parameter + ":", "").strip()
print(f'{file_path} Missing parameter {parameter}')
sys.exit(0)
def write_server(server, federated_with):
insert_sql = "INSERT INTO world(server, federated_with, updated_at, saved_at, checked) VALUES(%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING"
conn = None
try:
conn = psycopg2.connect(database = fediverse_db, user = fediverse_db_user, password = "", host = "/var/run/postgresql", port = "5432")
cur = conn.cursor()
cur.execute(insert_sql, (server, federated_with, now, now, 'f'))
print(f'writing {server} to world database')
conn.commit()
cur.close()
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
###############################################################################
# main
if __name__ == '__main__':
obj = Peers()
peers = obj.get_peers()
print(f"{obj.mastodon_hostname}'s peers: {len(peers)}")
now = datetime.now()
start = datetime.now()
program = obj.name
finish = start
obj.save_time(program, start, finish)
for peer in peers:
obj.getsoft(peer)
#results = ray.get([obj.getsoft.remote(server) for server in peers])
finish = datetime.now()
print(f"duration = {finish - start}.\nprocessed servers: {len(peers)}")
obj.save_time(program, start, finish)

249
setup.py
Veure arxiu

@ -1,249 +0,0 @@
import getpass
from mastodon import Mastodon
from mastodon.Mastodon import MastodonMalformedEventError, MastodonNetworkError, MastodonReadTimeout, MastodonAPIError, MastodonIllegalArgumentError
import fileinput,re
import os
import sys
def create_dir():
if not os.path.exists('secrets'):
os.makedirs('secrets')
def create_file():
if not os.path.exists('secrets/secrets.txt'):
with open('secrets/secrets.txt', 'w'): pass
print(secrets_filepath + " created!")
def create_config():
if not os.path.exists('config'):
os.makedirs('config')
if not os.path.exists(config_filepath):
print(config_filepath + " created!")
with open('config/config.txt', 'w'): pass
def write_params():
with open(secrets_filepath, 'a') as the_file:
print("Writing secrets parameter names to " + secrets_filepath)
the_file.write('uc_client_id: \n'+'uc_client_secret: \n'+'uc_access_token: \n')
def write_config():
with open(config_filepath, 'a') as the_file:
the_file.write('mastodon_hostname: \n')
the_file.write('bot_username: \n')
print("adding parameters 'mastodon_hostname' & 'bot_username' to "+ config_filepath)
def read_client_lines(self):
client_path = 'app_clientcred.txt'
with open(client_path) as fp:
line = fp.readline()
cnt = 1
while line:
if cnt == 1:
print("Writing client id to " + secrets_filepath)
modify_file(secrets_filepath, "uc_client_id: ", value=line.rstrip())
elif cnt == 2:
print("Writing client secret to " + secrets_filepath)
modify_file(secrets_filepath, "uc_client_secret: ", value=line.rstrip())
line = fp.readline()
cnt += 1
def read_token_line(self):
token_path = 'app_usercred.txt'
with open(token_path) as fp:
line = fp.readline()
print("Writing access token to " + secrets_filepath)
modify_file(secrets_filepath, "uc_access_token: ", value=line.rstrip())
def read_config_line():
with open(config_filepath) as fp:
line = fp.readline()
modify_file(config_filepath, "mastodon_hostname: ", value=hostname)
modify_file(config_filepath, "bot_username: ", value=bot_username)
def remove_secrets():
if os.path.exists("secrets/secrets.txt"):
print("Removing secrets/secrets.txt file..")
os.remove("secrets/secrets.txt")
def log_in():
error = 0
try:
global hostname
global bot_username
hostname = input("Enter Mastodon hostname: ")
user_name = input("User name, ex. user@" + hostname +"? ")
user_password = getpass.getpass("User password? ")
bot_username = input("Bot's username, ex. wikibot: ")
app_name = input("This app name? ")
Mastodon.create_app(app_name, scopes=["read","write"],
to_file="app_clientcred.txt", api_base_url=hostname)
mastodon = Mastodon(client_id = "app_clientcred.txt", api_base_url = hostname)
mastodon.log_in(
user_name,
user_password,
scopes = ["read", "write"],
to_file = "app_usercred.txt"
)
except MastodonIllegalArgumentError as i_error:
error = 1
sys.stdout.write(f'\n{str(i_error)}\n')
except MastodonNetworkError as n_error:
error = 1
sys.stdout.write(f'\n{str(n_error)}\n')
except MastodonReadTimeout as r_error:
error = 1
sys.stdout.write(f'\n{str(r_error)}\n')
except MastodonAPIError as a_error:
error = 1
sys.stdout.write(f'\n{str(a_error)}\n')
finally:
if error == 0:
create_dir()
create_file()
write_params()
client_path = 'app_clientcred.txt'
read_client_lines(client_path)
token_path = 'app_usercred.txt'
read_token_line(token_path)
if os.path.exists("app_clientcred.txt"):
print("Removing app_clientcred.txt temp file..")
os.remove("app_clientcred.txt")
if os.path.exists("app_usercred.txt"):
print("Removing app_usercred.txt temp file..")
os.remove("app_usercred.txt")
print("Secrets setup done!\n")
else:
remove_secrets()
if os.path.exists("app_clientcred.txt"):
print("Removing app_clientcred.txt file..")
os.remove("app_clientcred.txt")
sys.exit()
def modify_file(file_name,pattern,value=""):
fh=fileinput.input(file_name,inplace=True)
for line in fh:
replacement=pattern + value
line=re.sub(pattern,replacement,line)
sys.stdout.write(line)
fh.close()
def get_parameter( parameter, file_path ):
# Check if secrets file exists
if not os.path.isfile(file_path):
print("File %s not found, creating it."%file_path)
log_in()
# Find parameter in file
with open( file_path ) as f:
for line in f:
if line.startswith( parameter ):
return line.replace(parameter + ":", "").strip()
# Cannot find parameter, exit
print(file_path + " Missing parameter %s "%parameter)
sys.exit(0)
def get_bot_host( parameter, config_filepath ):
# Check if secrets file exists
if not os.path.isfile(config_filepath):
print("File %s not found, creating it."%config_filepath)
create_config()
# Find parameter in file
with open( config_filepath ) as f:
for line in f:
if line.startswith( parameter ):
return line.replace(parameter + ":", "").strip()
# Cannot find parameter, exit
print(config_filepath + " Missing parameter %s "%parameter)
write_config()
read_config_line()
print("hostname & bot_username setup done!")
sys.exit(0)
###############################################################################
# main
if __name__ == '__main__':
# Load secrets from secrets file
secrets_filepath = "secrets/secrets.txt"
uc_client_id = get_parameter("uc_client_id", secrets_filepath)
uc_client_secret = get_parameter("uc_client_secret", secrets_filepath)
uc_access_token = get_parameter("uc_access_token", secrets_filepath)
# Load configuration from config file
config_filepath = "config/config.txt"
mastodon_hostname = get_bot_host("mastodon_hostname", config_filepath)
bot_username = get_bot_host("bot_username", config_filepath)