- reformatted config.defaults.json
- also store script name
- "base_url": "",
- "log_level": "info",
- "port": 8069,
- "useragent": "Mozilla/5.0 (Windows NT 10.0; rv:102.0) Gecko/20100101 Firefox/102.0",
- "bot_enabled": false,
- "bot_instance": "https://example.com",
- "bot_token": "",
- "bot_visibility": "unlisted",
- "slogan": "### Your footer slogan ###",
+ "base_url" : "",
+ "log_level" : "info",
+ "port" : 8069,
+ "useragent" : "Mozilla/5.0 (Windows NT 10.0; rv:102.0) Gecko/20100101 Firefox/102.0",
+ "timeout" : 5,
+ "bot_enabled" : false,
+ "bot_instance" : "https://example.com",
+ "bot_token" : "",
+ "bot_visibility" : "unlisted",
+ "slogan" : "### Your footer slogan ###",
"recheck_instance": 3600,
"recheck_instance": 3600,
- res = reqto.get(f"https://{domain}/api/v1/instance/peers", headers=headers, timeout=5)
+ res = reqto.get(f"https://{domain}/api/v1/instance/peers", headers=headers, timeout=config["timeout"])
if not res.ok or res.status_code >= 400:
print("WARNING: Cannot fetch peers:", domain)
if not res.ok or res.status_code >= 400:
print("WARNING: Cannot fetch peers:", domain)
def post_json_api(domain: str, path: str, data: str) -> list:
try:
# NOISY-DEBUG: print("DEBUG: Sending POST to domain,path,data:", domain, path, data)
def post_json_api(domain: str, path: str, data: str) -> list:
try:
# NOISY-DEBUG: print("DEBUG: Sending POST to domain,path,data:", domain, path, data)
- res = reqto.post(f"https://{domain}{path}", data=data, headers=headers, timeout=5)
+ res = reqto.post(f"https://{domain}{path}", data=data, headers=headers, timeout=config["timeout"])
if not res.ok or res.status_code >= 400:
print("WARNING: Cannot query JSON API:", domain, path, data, res.status_code)
if not res.ok or res.status_code >= 400:
print("WARNING: Cannot query JSON API:", domain, path, data, res.status_code)
json = None
for request in requests:
# NOISY-DEBUG: print("DEBUG: Fetching request:", request)
json = None
for request in requests:
# NOISY-DEBUG: print("DEBUG: Fetching request:", request)
- res = reqto.get(request, headers=headers, timeout=5)
+ res = reqto.get(request, headers=headers, timeout=config["timeout"])
# NOISY-DEBUG: print("DEBUG: res.ok,res.json[]:", res.ok, type(res.json()))
if res.ok and res.json() is not None:
# NOISY-DEBUG: print("DEBUG: res.ok,res.json[]:", res.ok, type(res.json()))
if res.ok and res.json() is not None:
print("--- New block:", blocker, blocked, reason, block_level, first_added, last_seen)
try:
c.execute(
print("--- New block:", blocker, blocked, reason, block_level, first_added, last_seen)
try:
c.execute(
- "INSERT INTO blocks SELECT ?, ?, ?, ?, ?, ?",
+ "INSERT INTO blocks (blocker, blocked, reason, block_level, first_added, last_seen) VALUES(?, ?, ?, ?, ?, ?)",
print("ERROR: failed SQL query:", blocker, blocked, reason, block_level, first_added, last_seen)
sys.exit(255)
print("ERROR: failed SQL query:", blocker, blocked, reason, block_level, first_added, last_seen)
sys.exit(255)
-def add_instance(domain: str, originator: str):
- # NOISY-DEBUG: print("DEBUG: domain,originator:", domain, originator)
+def add_instance(domain: str, origin: str, originator: str):
+ # NOISY-DEBUG: print("DEBUG: domain,origin:", domain, origin, originator)
if domain.find("@") > 0:
print("WARNING: Bad domain name:", domain)
raise
if domain.find("@") > 0:
print("WARNING: Bad domain name:", domain)
raise
- elif originator is not None and originator.find("@") > 0:
- print("WARNING: Bad originator name:", originator)
+ elif origin is not None and origin.find("@") > 0:
+ print("WARNING: Bad origin name:", origin)
- print("--- Adding new instance:", domain, originator)
+ print(f"--- Adding new instance {domain} (origin: {origin})")
- "INSERT INTO instances (domain,originator,hash,software) VALUES (?, ?, ?, ?)",
+ "INSERT INTO instances (domain,origin,originator,hash,software) VALUES (?, ?, ?, ?, ?)",
originator,
get_hash(domain),
determine_software(domain)
originator,
get_hash(domain),
determine_software(domain)
try:
doc = BeautifulSoup(
try:
doc = BeautifulSoup(
- reqto.get(f"https://{domain}/about/more", headers=headers, timeout=5).text,
+ reqto.get(f"https://{domain}/about/more", headers=headers, timeout=config["timeout"]).text,
try:
doc = BeautifulSoup(
try:
doc = BeautifulSoup(
- reqto.get(f"https://{domain}/friendica", headers=headers, timeout=5).text,
+ reqto.get(f"https://{domain}/friendica", headers=headers, timeout=config["timeout"]).text,
if fba.c.fetchone() == None:
# NOISY-DEBUG: print("DEBUG: Domain wasn't found, adding:", blocked, blocker)
if fba.c.fetchone() == None:
# NOISY-DEBUG: print("DEBUG: Domain wasn't found, adding:", blocked, blocker)
- fba.add_instance(blocked, blocker)
+ fba.add_instance(blocked, blocker, argv[0])
timestamp = int(time.time())
fba.c.execute(
timestamp = int(time.time())
fba.c.execute(
# handling CSRF, I've saw at least one server requiring it to access the endpoint
# NOISY-DEBUG: print("DEBUG: Fetching meta:", blocker)
meta = bs4.BeautifulSoup(
# handling CSRF, I've saw at least one server requiring it to access the endpoint
# NOISY-DEBUG: print("DEBUG: Fetching meta:", blocker)
meta = bs4.BeautifulSoup(
- reqto.get(f"https://{blocker}/about", headers=fba.headers, timeout=5).text,
+ reqto.get(f"https://{blocker}/about", headers=fba.headers, timeout=fba.config["timeout"]).text,
reqheaders = fba.headers
# NOISY-DEBUG: print("DEBUG: Quering API domain_blocks:", blocker)
reqheaders = fba.headers
# NOISY-DEBUG: print("DEBUG: Quering API domain_blocks:", blocker)
- blocks = reqto.get(f"https://{blocker}/api/v1/instance/domain_blocks", headers=reqheaders, timeout=5).json()
+ blocks = reqto.get(f"https://{blocker}/api/v1/instance/domain_blocks", headers=reqheaders, timeout=fba.config["timeout"]).json()
# NOISY-DEBUG: print("DEBUG: blocks():", len(blocks))
for block in blocks:
# NOISY-DEBUG: print("DEBUG: blocks():", len(blocks))
for block in blocks:
if fba.c.fetchone() == None:
# NOISY-DEBUG: print("DEBUG: Hash wasn't found, adding:", blocked, blocker)
if fba.c.fetchone() == None:
# NOISY-DEBUG: print("DEBUG: Hash wasn't found, adding:", blocked, blocker)
- fba.add_instance(blocked, blocker)
+ fba.add_instance(blocked, blocker, argv[0])
else:
# Doing the hash search for instance names as well to tidy up DB
fba.c.execute(
else:
# Doing the hash search for instance names as well to tidy up DB
fba.c.execute(
print("INFO: blocker:", blocker)
try:
# Blocks
print("INFO: blocker:", blocker)
try:
# Blocks
- federation = reqto.get(f"https://{blocker}/api/v1/instance/peers?filter=suspended", headers=fba.headers, timeout=5).json()
+ federation = reqto.get(f"https://{blocker}/api/v1/instance/peers?filter=suspended", headers=fba.headers, timeout=fba.config["timeout"]).json()
if (federation == None):
print("WARNING: No valid response:", blocker);
if (federation == None):
print("WARNING: No valid response:", blocker);
-def fetch_instances(domain: str, originator: str):
- # NOISY-DEBUG: print("DEBUG: Fetching instances for domain:", domain, originator)
+def fetch_instances(domain: str, origin: str):
+ # NOISY-DEBUG: print("DEBUG: Fetching instances for domain:", domain, origin)
peerlist = fba.get_peers(domain)
if (peerlist is None):
peerlist = fba.get_peers(domain)
if (peerlist is None):
)
if fba.c.fetchone() == None:
)
if fba.c.fetchone() == None:
- # NOISY-DEBUG: print("DEBUG: Adding new domain:", domain, originator)
- fba.add_instance(domain, originator)
+ # NOISY-DEBUG: print("DEBUG: Adding new domain:", domain, origin)
+ fba.add_instance(domain, origin, sys.argv[0])
if fba.c.fetchone() == None:
# NOISY-DEBUG: print("DEBUG: Adding new instance:", instance, domain)
if fba.c.fetchone() == None:
# NOISY-DEBUG: print("DEBUG: Adding new instance:", instance, domain)
- fba.add_instance(instance, domain)
+ fba.add_instance(instance, domain, sys.argv[0])