+ result[block_level] = [entry]
+
+ return result
+
+@router.get(fba.config["base_url"] + "/scoreboard")
+def index(request: Request, blockers: int = None, blocked: int = None, reference: int = None, software: int = None):
+ scores = None
+
+ if blockers != None and blockers > 0:
+ res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top?blockers={blockers}")
+ elif blocked != None and blocked > 0:
+ res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top?blocked={blocked}")
+ elif reference != None and reference > 0:
+ res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top?reference={reference}")
+ elif software != None and software > 0:
+ res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top?software={software}")
+ else:
+ raise HTTPException(status_code=400, detail="No filter specified")
+
+ if res == None:
+ raise HTTPException(status_code=500, detail="Could not determine scores")
+ elif not res.ok:
+ raise HTTPException(status_code=res.status_code, detail=res.text)
+
+ return templates.TemplateResponse("scoreboard.html", {
+ "base_url" : fba.config["base_url"],
+ "slogan" : fba.config["slogan"],
+ "request" : request,
+ "scoreboard": True,
+ "blockers" : blockers,
+ "blocked" : blocked,
+ "reference" : reference,
+ "software" : software,
+ "scores" : res.json()
+ })
+
+@router.get(fba.config["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 = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/info")
+
+ if not info.ok:
+ raise HTTPException(status_code=info.status_code, detail=info.text)
+
+ info = info.json()
+ elif domain != None:
+ blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api?domain={domain}")
+ elif reason != None:
+ blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api?reason={reason}")
+ elif reverse != None:
+ blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['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_seen"] = datetime.utcfromtimestamp(block["first_seen"]).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
+ })
+
+@router.get(fba.config["base_url"] + "/api/mutual")
+def mutual(domains: list[str] = Query()):
+ """Return 200 if federation is open between the two, 4xx otherwise"""
+ fba.cursor.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 = fba.cursor.fetchone()
+
+ if res is not None:
+ # Blocks found
+ return responses.JSONResponse(status_code=418, content={})
+
+ # No known blocks
+ return responses.JSONResponse(status_code=200, content={})
+
+@router.get(fba.config["base_url"] + "/rss")
+def rss(request: Request, domain: str = None):
+ if domain != None:
+ wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
+ punycode = domain.encode('idna').decode('utf-8')
+ fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? ORDER BY first_seen DESC LIMIT 50",
+ (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
+ else:
+ fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks ORDER BY first_seen DESC LIMIT 50")
+
+ blocks = fba.cursor.fetchall()
+
+ result = []
+ for blocker, blocked, block_level, reason, first_seen, last_seen in blocks:
+ first_seen = utils.format_datetime(datetime.fromtimestamp(first_seen))
+ 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_seen" : first_seen
+ })
+
+ timestamp = utils.format_datetime(datetime.now())
+
+ return templates.TemplateResponse("rss.xml", {
+ "request" : request,
+ "timestamp": timestamp,
+ "domain" : domain,
+ "blocks" : result
+ }, headers={
+ "Content-Type": "routerlication/rss+xml"
+ })