from fba import database
from fba import utils
+from fba.helpers import blacklist
from fba.helpers import config
+from fba.helpers import domain as domain_helper
from fba.helpers import json as json_helper
from fba.helpers import tidyup
@router.get(config.get("base_url") + "/api/info.json", response_class=JSONResponse)
def api_info():
- database.cursor.execute("SELECT (SELECT COUNT(domain) FROM instances), (SELECT COUNT(domain) FROM instances WHERE software IN ('pleroma', 'mastodon', 'lemmy', 'friendica', 'misskey', 'peertube', 'takahe')), (SELECT COUNT(blocker) FROM blocks), (SELECT COUNT(domain) FROM instances WHERE last_error_details IS NOT NULL)")
+ database.cursor.execute("SELECT (SELECT COUNT(domain) FROM instances), (SELECT COUNT(domain) FROM instances WHERE software IN ('pleroma', 'mastodon', 'lemmy', 'friendica', 'misskey', 'peertube', 'takahe', 'gotosocial', 'brighteon', 'wildebeest', 'bookwyrm', 'mitra', 'areionskey', 'mammuthus', 'neodb')), (SELECT COUNT(blocker) FROM blocks), (SELECT COUNT(domain) FROM instances WHERE last_error_details IS NOT NULL)")
row = database.cursor.fetchone()
- return {
+ return JSONResponse(status_code=200, content={
"known_instances" : row[0],
"supported_instances": row[1],
"blocks_recorded" : row[2],
"erroneous_instances": row[3],
- }
+ })
@router.get(config.get("base_url") + "/api/scoreboard.json", response_class=JSONResponse)
def api_scoreboard(mode: str, amount: int):
elif mode == "detection_mode":
database.cursor.execute("SELECT detection_mode, COUNT(domain) AS cnt FROM instances GROUP BY detection_mode ORDER BY cnt DESC LIMIT ?", [amount])
elif mode == "avg_peers":
- database.cursor.execute("SELECT software, AVG(total_peers) AS average FROM instances WHERE software IS NOT NULL GROUP BY software HAVING average > 0 ORDER BY average DESC LIMIT ?", [amount])
+ database.cursor.execute("SELECT software, AVG(total_peers) AS average FROM instances WHERE software IS NOT NULL AND total_peers IS NOT NULL GROUP BY software HAVING average > 0 ORDER BY average DESC LIMIT ?", [amount])
+ elif mode == "avg_blocks":
+ database.cursor.execute("SELECT software, AVG(total_blocks) AS average FROM instances WHERE software IS NOT NULL AND total_blocks IS NOT NULL GROUP BY software HAVING average > 0 ORDER BY average DESC LIMIT ?", [amount])
elif mode == "obfuscator":
database.cursor.execute("SELECT software, COUNT(domain) AS cnt FROM instances WHERE has_obfuscation = 1 GROUP BY software ORDER BY cnt DESC LIMIT ?", [amount])
elif mode == "obfuscation":
scores = list()
- for domain, score in database.cursor.fetchall():
+ for row in database.cursor.fetchall():
scores.append({
- "domain": domain,
- "score" : round(score)
+ "domain": row[0],
+ "score" : round(row[1]),
})
- return scores
+ return JSONResponse(status_code=200, content=scores)
@router.get(config.get("base_url") + "/api/list.json", response_class=JSONResponse)
def api_list(request: Request, mode: str, value: str, amount: int):
elif amount > config.get("api_limit"):
raise HTTPException(status_code=500, detail=f"amount={amount} is to big")
- if mode in ("detection_mode", "software", "command"):
+ if mode in ("detection_mode", "software", "command", "origin"):
database.cursor.execute(
- f"SELECT domain, origin, software, detection_mode, command, total_peers, total_blocks, first_seen, last_updated \
+ f"SELECT * \
FROM instances \
WHERE {mode} = ? \
ORDER BY domain \
)
domainlist = database.cursor.fetchall()
-
return domainlist
@router.get(config.get("base_url") + "/api/top.json", response_class=JSONResponse)
)
elif mode in ["domain", "reverse"]:
domain = tidyup.domain(value)
- if not utils.is_domain_wanted(domain):
+ if not domain_helper.is_wanted(domain):
raise HTTPException(status_code=500, detail=f"domain='{domain}' is not wanted")
wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
# Tidy up domain name
domain = tidyup.domain(domain).encode("idna").decode("utf-8")
- if not utils.is_domain_wanted(domain):
+ if not domain_helper.is_wanted(domain):
raise HTTPException(status_code=500, detail=f"domain='{domain}' is not wanted")
# Fetch domain data
if domain_data is None:
raise HTTPException(status_code=404, detail=f"domain='{domain}' not found")
- return domain_data
+ return JSONResponse(status_code=200, content=dict(domain_data))
@router.get(config.get("base_url") + "/api/mutual.json", response_class=JSONResponse)
def api_mutual(domains: list[str] = Query()):
# No known blocks
return JSONResponse(status_code=200, content={})
+@router.get(config.get("base_url") + "/.well-known/nodeinfo", response_class=JSONResponse)
+def wellknown_nodeinfo(request: Request):
+ return JSONResponse(status_code=200, content={
+ "links": ({
+ "rel" : "http://nodeinfo.diaspora.software/ns/schema/1.0",
+ "href": f"{config.get('scheme')}://{config.get('hostname')}{config.get('base_url')}/nodeinfo/1.0"
+ })
+ })
+
+@router.get(config.get("base_url") + "/nodeinfo/1.0", response_class=JSONResponse)
+def nodeinfo_1_0(request: Request):
+ return JSONResponse(status_code=200, content={
+ "version": "1.0",
+ "software": {
+ "name": "FBA",
+ "version": "0.1",
+ },
+ "protocols": {
+ "inbound": (),
+ "outbound": (
+ "rss",
+ ),
+ },
+ "services": {
+ "inbound": (),
+ "outbound": (
+ "rss",
+ ),
+ },
+ "usage": {
+ "users": {},
+ },
+ "openRegistrations": False,
+ "metadata": {
+ "nodeName": "Fedi Block API",
+ "protocols": {
+ "inbound": (),
+ "outbound": (
+ "rss",
+ ),
+ },
+ "services": {
+ "inbound": (),
+ "outbound": (
+ "rss",
+ ),
+ },
+ "explicitContent": False,
+ },
+ })
+
+@router.get(config.get("base_url") + "/api/v1/instance/domain_blocks", response_class=JSONResponse)
+def api_domain_blocks(request: Request):
+ blocked = blacklist.get_all()
+ blocking = list()
+
+ for block in blocked:
+ blocking.append({
+ "domain" : block,
+ "digest" : utils.get_hash(block),
+ "severity": "suspend",
+ "comment" : blocked[block],
+ })
+
+ return JSONResponse(status_code=200, content=blocking)
+
+@router.get(config.get("base_url") + "/api/v1/instance/peers", response_class=JSONResponse)
+def api_peers(request: Request):
+ database.cursor.execute("SELECT domain FROM instances WHERE nodeinfo_url IS NOT NULL")
+
+ peers = list()
+ for row in database.cursor.fetchall():
+ peers.append(row["domain"])
+
+ return JSONResponse(status_code=200, content=peers)
+
@router.get(config.get("base_url") + "/scoreboard")
def scoreboard(request: Request, mode: str, amount: int):
- response = None
-
- if mode == "blocker" and amount > 0:
- response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/scoreboard.json?mode=blocker&amount={amount}")
- elif mode == "blocked" and amount > 0:
- response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/scoreboard.json?mode=blocked&amount={amount}")
- elif mode == "reference" and amount > 0:
- response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/scoreboard.json?mode=reference&amount={amount}")
- elif mode == "software" and amount > 0:
- response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/scoreboard.json?mode=software&amount={amount}")
- elif mode == "command" and amount > 0:
- response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/scoreboard.json?mode=command&amount={amount}")
- elif mode == "error_code" and amount > 0:
- response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/scoreboard.json?mode=error_code&amount={amount}")
- elif mode == "detection_mode" and amount > 0:
- response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/scoreboard.json?mode=detection_mode&amount={amount}")
- elif mode == "avg_peers" and amount > 0:
- response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/scoreboard.json?mode=avg_peers&amount={amount}")
- elif mode == "obfuscator" and amount > 0:
- response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/scoreboard.json?mode=obfuscator&amount={amount}")
- elif mode == "obfuscation" and amount > 0:
- response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/scoreboard.json?mode=obfuscation&amount={amount}")
- elif mode == "block_level" and amount > 0:
- response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/scoreboard.json?mode=block_level&amount={amount}")
- else:
- raise HTTPException(status_code=400, detail="No filter specified")
+ if mode == "":
+ raise HTTPException(status_code=400, detail="No mode specified")
+ elif amount <= 0:
+ raise HTTPException(status_code=500, detail="Invalid amount specified")
+
+ response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/scoreboard.json?mode={mode}&amount={amount}")
if response is None:
raise HTTPException(status_code=500, detail="Could not determine scores")
raise HTTPException(status_code=response.status_code, detail=response.text)
return templates.TemplateResponse("views/scoreboard.html", {
- "base_url" : config.get("base_url"),
+ "base_url" : utils.base_url(),
"slogan" : config.get("slogan"),
"theme" : config.get("theme"),
"request" : request,
"scores" : json_helper.from_response(response)
})
-@router.get(config.get("base_url") + "/")
-def index(request: Request):
- # Get info
- response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/info.json")
-
- if not response.ok:
- raise HTTPException(status_code=response.status_code, detail=response.text)
-
- return templates.TemplateResponse("views/index.html", {
- "request": request,
- "theme" : config.get("theme"),
- "info" : response.json(),
- "slogan" : config.get("slogan"),
- })
-
@router.get(config.get("base_url") + "/list")
def list_domains(request: Request, mode: str, value: str, amount: int = config.get("api_limit")):
if mode == "detection_mode" and not instances.valid(value, "detection_mode"):
def top(request: Request, mode: str, value: str, amount: int = config.get("api_limit")):
if mode == "block_level" and not blocks.valid(value, "block_level"):
raise HTTPException(status_code=500, detail="Invalid block level provided")
- elif mode in ["domain", "reverse"] and not utils.is_domain_wanted(value):
+ elif mode in ["domain", "reverse"] and not domain_helper.is_wanted(value):
raise HTTPException(status_code=500, detail="Invalid or blocked domain specified")
response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?mode={mode}&value={value}&amount={amount}")
# Tidy up domain name
domain = tidyup.domain(domain).encode("idna").decode("utf-8")
- if not utils.is_domain_wanted(domain):
+ if not domain_helper.is_wanted(domain):
raise HTTPException(status_code=500, detail=f"domain='{domain}' is not wanted")
response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/domain.json?domain={domain}")
- if not response.ok or response.status_code >= 300 or response.text.strip() == "":
+ if not response.ok or response.status_code > 200 or response.text.strip() == "":
raise HTTPException(status_code=response.status_code, detail=response.reason)
domain_data = response.json()
tformat = config.get("timestamp_format")
instance = dict()
for key in domain_data.keys():
- if key in ["last_nodeinfo", "last_blocked", "first_seen", "last_updated", "last_instance_fetch"] and isinstance(domain_data[key], float):
+ if key in ["last_nodeinfo", "last_blocked", "first_seen", "last_updated", "last_instance_fetch", "last_response_time"] and isinstance(domain_data[key], float):
# Timestamps
instance[key] = datetime.utcfromtimestamp(domain_data[key]).strftime(tformat)
else:
"request" : request,
"timestamp": format_datetime(datetime.now()),
"domain" : domain,
+ "scheme" : config.get("scheme"),
"hostname" : config.get("hostname"),
"blocks" : blocklist
}, headers={
"base_url": config.get("base_url")
})
+@router.get(config.get("base_url") + "/")
+def index(request: Request):
+ # Get info
+ response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/info.json")
+
+ if not response.ok:
+ raise HTTPException(status_code=response.status_code, detail=response.text)
+
+ return templates.TemplateResponse("views/index.html", {
+ "request": request,
+ "theme" : config.get("theme"),
+ "info" : response.json(),
+ "slogan" : config.get("slogan"),
+ })
+
if __name__ == "__main__":
- uvicorn.run("daemon:router", host=config.get("host"), port=config.get("port"), log_level=config.get("log_level"))
+ uvicorn.run("daemon:router", host=config.get("host"), port=config.get("port"), log_level=config.get("log_level"), proxy_headers=True)
projects / fba.git / blobdiff