123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211 |
- import uvicorn
- from fastapi import FastAPI, Request, HTTPException, responses, Query
- import sqlite3
- from hashlib import sha256
- from fastapi.templating import Jinja2Templates
- from requests import get
- from json import loads
- from re import sub
- from datetime import datetime
- from email import utils
- with open("config.json") as f:
- config = loads(f.read())
- base_url = config["base_url"]
- port = config["port"]
- app = FastAPI(docs_url=base_url+"/docs", redoc_url=base_url+"/redoc")
- templates = Jinja2Templates(directory=".")
- def get_hash(domain: str) -> str:
- return sha256(domain.encode("utf-8")).hexdigest()
- @app.get(base_url+"/info")
- def info():
- conn = sqlite3.connect("blocks.db")
- c = conn.cursor()
- c.execute("select (select count(domain) from instances), (select count(domain) from instances where software in ('pleroma', 'mastodon', 'misskey', 'gotosocial', 'friendica', 'lemmy')), (select count(blocker) from blocks)")
- known, indexed, blocks = c.fetchone()
- c.close()
- return {
- "known_instances": known,
- "indexed_instances": indexed,
- "blocks_recorded": blocks
- }
- @app.get(base_url+"/top")
- def top(blocked: int = None, blockers: int = None):
- conn = sqlite3.connect("blocks.db")
- c = conn.cursor()
- if blocked == None and blockers == None:
- raise HTTPException(status_code=400, detail="No filter specified")
- elif blocked != None:
- if blocked > 500:
- raise HTTPException(status_code=400, detail="Too many results")
- c.execute("select blocked, count(blocked) from blocks where block_level = 'reject' group by blocked order by count(blocked) desc limit ?", (blocked,))
- elif blockers != None:
- if blockers > 500:
- raise HTTPException(status_code=400, detail="Too many results")
- c.execute("select blocker, count(blocker) from blocks where block_level = 'reject' group by blocker order by count(blocker) desc limit ?", (blockers,))
- scores = c.fetchall()
- c.close()
- scoreboard = []
- print(scores)
- for domain, highscore in scores:
- scoreboard.append({"domain": domain, "highscore": highscore})
- return scoreboard
- @app.get(base_url+"/top/weighted")
- def weighted():
- conn = sqlite3.connect("blocks.db")
- c = conn.cursor()
- # trying to make the limit adjustable makes the app crash with `sqlite3.IntegrityError: datatype mismatch`, so rolling with hardcoded one
- c.execute("with weighted as (SELECT blocker, count(*) as instance_weight FROM blocks where block_level = 'reject' GROUP BY blocker) SELECT rank() OVER (order by sum(1 / log(instance_weight)) desc) as rank, b.blocked, count(*) as total_blocks, sum(1 / log(instance_weight + 1)) as weighted_score FROM blocks b JOIN weighted USING (blocker) GROUP BY b.blocked order by weighted_score DESC LIMIT 100")
- scores = c.fetchall()
- c.close()
- scoreboard = []
- print(scores)
- for rank, domain, blocks, weight in scores:
- scoreboard.append({"domain": domain, "blocks": blocks, "weight": round(weight, 4)})
- return scoreboard
- @app.get(base_url+"/api")
- def blocked(domain: str = None, reason: str = None, reverse: str = None):
- if domain == None and reason == None and reverse == None:
- raise HTTPException(status_code=400, detail="No filter specified")
- if reason != None:
- reason = sub("(%|_)", "", reason)
- if len(reason) < 3:
- raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
- conn = sqlite3.connect("blocks.db")
- c = conn.cursor()
- if domain != None:
- wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
- punycode = domain.encode('idna').decode('utf-8')
- c.execute("select blocker, blocked, block_level, reason, first_added, last_seen from blocks where blocked = ? collate nocase or blocked = ? collate nocase or blocked = ? collate nocase or blocked = ? collate nocase or blocked = ? collate nocase or blocked = ? collate nocase order by first_added asc",
- (domain, "*." + domain, wildchar, get_hash(domain), punycode, "*." + punycode))
- elif reverse != None:
- c.execute("select blocker, blocked, block_level, reason, first_added, last_seen from blocks where blocker = ? collate nocase order by first_added asc", (reverse,))
- else:
- c.execute("select blocker, blocked, block_level, reason, first_added, last_seen from blocks where reason like ? and reason != '' collate nocase order by first_added asc", ("%"+reason+"%",))
- blocks = c.fetchall()
- c.close()
- result = {}
- for blocker, blocked, block_level, reason, first_added, last_seen in blocks:
- entry = {"blocker": blocker, "blocked": blocked, "reason": reason, "first_added": first_added, "last_seen": last_seen}
- if block_level in result:
- result[block_level].append(entry)
- else:
- result[block_level] = [entry]
- return result
- @app.get(base_url+"/scoreboard")
- def index(request: Request, blockers: int = None, blocked: int = None):
- if blockers == None and blocked == None:
- raise HTTPException(status_code=400, detail="No filter specified")
- elif blockers != None:
- scores = get(f"http://127.0.0.1:{port}{base_url}/top?blockers={blockers}")
- elif blocked != None:
- scores = get(f"http://127.0.0.1:{port}{base_url}/top?blocked={blocked}")
- if scores != None:
- if not scores.ok:
- raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
- scores = scores.json()
- return templates.TemplateResponse("index.html", {"request": request, "scoreboard": True, "blockers": blockers, "blocked": blocked, "scores": scores})
- @app.get(base_url+"/weighted")
- def index(request: Request):
- scores = get(f"http://127.0.0.1:{port}{base_url}/top/weighted")
- if scores != None:
- if not scores.ok:
- raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
- scores = scores.json()
- return templates.TemplateResponse("index.html", {"request": request, "weighted": True, "scores": scores})
- @app.get(base_url+"/")
- def index(request: Request, domain: str = None, reason: str = None, reverse: str = None):
- if domain == "" or reason == "" or reverse == "":
- return responses.RedirectResponse("/")
- info = None
- blocks = None
- if domain == None and reason == None and reverse == None:
- info = get(f"http://127.0.0.1:{port}{base_url}/info")
- if not info.ok:
- raise HTTPException(status_code=info.status_code, detail=info.text)
- info = info.json()
- elif domain != None:
- blocks = get(f"http://127.0.0.1:{port}{base_url}/api?domain={domain}")
- elif reason != None:
- blocks = get(f"http://127.0.0.1:{port}{base_url}/api?reason={reason}")
- elif reverse != None:
- blocks = get(f"http://127.0.0.1:{port}{base_url}/api?reverse={reverse}")
- if blocks != None:
- if not blocks.ok:
- raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
- blocks = blocks.json()
- for block_level in blocks:
- for block in blocks[block_level]:
- block["first_added"] = datetime.utcfromtimestamp(block["first_added"]).strftime('%Y-%m-%d %H:%M')
- block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime('%Y-%m-%d %H:%M')
- return templates.TemplateResponse("index.html", {"request": request, "domain": domain, "blocks": blocks, "reason": reason, "reverse": reverse, "info": info})
- @app.get(base_url+"/api/mutual")
- def mutual(domains: list[str] = Query()):
- """Return 200 if federation is open between the two, 4xx otherwise"""
- conn = sqlite3.connect('blocks.db')
- c = conn.cursor()
- c.execute(
- "SELECT block_level FROM blocks " \
- "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
- "AND block_level = 'reject' " \
- "LIMIT 1",
- {
- "a": domains[0],
- "b": domains[1],
- "aw": "*." + domains[0],
- "bw": "*." + domains[1],
- },
- )
- res = c.fetchone()
- c.close()
- if res is not None:
- # Blocks found
- return responses.JSONResponse(status_code=418, content={})
- # No known blocks
- return responses.JSONResponse(status_code=200, content={})
- @app.get(base_url+"/rss")
- def rss(request: Request, domain: str = None):
- conn = sqlite3.connect("blocks.db")
- c = conn.cursor()
- if domain != None:
- wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
- punycode = domain.encode('idna').decode('utf-8')
- c.execute("select blocker, blocked, block_level, reason, first_added, last_seen from blocks where blocked = ? or blocked = ? or blocked = ? or blocked = ? or blocked = ? or blocked = ? order by first_added desc limit 50",
- (domain, "*." + domain, wildchar, get_hash(domain), punycode, "*." + punycode))
- else:
- c.execute("select blocker, blocked, block_level, reason, first_added, last_seen from blocks order by first_added desc limit 50")
- blocks = c.fetchall()
- c.close()
- result = []
- for blocker, blocked, block_level, reason, first_added, last_seen in blocks:
- first_added = utils.format_datetime(datetime.fromtimestamp(first_added))
- if reason == None or reason == '':
- reason = "No reason provided."
- else:
- reason = "Provided reason: '" + reason + "'"
- result.append({"blocker": blocker, "blocked": blocked, "block_level": block_level, "reason": reason, "first_added": first_added})
- timestamp = utils.format_datetime(datetime.now())
- return templates.TemplateResponse("rss.xml", {"request": request, "timestamp": timestamp, "domain": domain, "blocks": result}, headers={"Content-Type": "application/rss+xml"})
- if __name__ == "__main__":
- uvicorn.run("api:app", host="127.0.0.1", port=port, log_level="info")
|