-#!/usr/bin/python3
+#!venv/bin/python3
# -*- coding: utf-8 -*-
# Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
templates = Jinja2Templates(directory="templates")
@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', '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)")
+def api_info() -> None:
+ database.cursor.execute("SELECT (SELECT COUNT(domain) FROM instances) AS total_websites, (SELECT COUNT(domain) FROM instances WHERE software IN ('pleroma', 'mastodon', 'lemmy', 'friendica', 'misskey', 'peertube', 'takahe', 'gotosocial', 'brighteon', 'wildebeest', 'bookwyrm', 'mitra', 'areionskey', 'mammuthus', 'neodb', 'smithereen', 'vebinet', 'toki', 'snac', 'biblioreads', 'wordpress', 'oolong', 'diaspora')) AS supported_instances, (SELECT COUNT(blocker) FROM blocks) AS total_blocks, (SELECT COUNT(domain) FROM instances WHERE last_error_details IS NOT NULL) AS erroneous_instances")
+
row = database.cursor.fetchone()
return JSONResponse(status_code=200, content={
- "known_instances" : row[0],
- "supported_instances": row[1],
- "blocks_recorded" : row[2],
- "erroneous_instances": row[3],
+ "total_websites" : row["total_websites"],
+ "supported_instances": row["supported_instances"],
+ "total_blocks" : row["total_blocks"],
+ "erroneous_instances": row["erroneous_instances"],
})
+
@router.get(config.get("base_url") + "/api/scoreboard.json", response_class=JSONResponse)
-def api_scoreboard(mode: str, amount: int):
+def api_scoreboard(mode: str, amount: int) -> None:
if amount > config.get("api_limit"):
raise HTTPException(status_code=400, detail="Too many results")
database.cursor.execute("SELECT blocker, COUNT(blocker) AS score FROM blocks GROUP BY blocker ORDER BY score DESC LIMIT ?", [amount])
elif mode == "reference":
database.cursor.execute("SELECT origin, COUNT(domain) AS score FROM instances WHERE origin IS NOT NULL GROUP BY origin ORDER BY score DESC LIMIT ?", [amount])
+ elif mode == "original_software":
+ database.cursor.execute("SELECT original_software, COUNT(domain) AS score FROM instances WHERE original_software IS NOT NULL GROUP BY original_software ORDER BY score DESC, original_software ASC LIMIT ?", [amount])
elif mode == "software":
database.cursor.execute("SELECT software, COUNT(domain) AS score FROM instances WHERE software IS NOT NULL GROUP BY software ORDER BY score DESC, software ASC LIMIT ?", [amount])
elif mode == "command":
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":
- database.cursor.execute("SELECT has_obfuscation, COUNT(domain) AS cnt FROM instances WHERE software IN ('pleroma', 'mastodon', 'friendica') GROUP BY has_obfuscation ORDER BY cnt DESC LIMIT ?", [amount])
+ database.cursor.execute("SELECT has_obfuscation, COUNT(domain) AS cnt FROM instances WHERE software IN ('pleroma', 'lemmy', 'mastodon', 'misskey', 'friendica') GROUP BY has_obfuscation ORDER BY cnt DESC LIMIT ?", [amount])
elif mode == "block_level":
database.cursor.execute("SELECT block_level, COUNT(rowid) AS cnt FROM blocks GROUP BY block_level ORDER BY cnt DESC LIMIT ?", [amount])
else:
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):
+def api_list(request: Request, mode: str, value: str, amount: int) -> None:
if mode is None or value is None or amount is None:
raise HTTPException(status_code=500, detail="No filter specified")
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", "origin"):
+ if mode in ("detection_mode", "original_software", "software", "command", "origin"):
database.cursor.execute(
f"SELECT * \
FROM instances \
ORDER BY domain \
LIMIT ?", [value, amount]
)
+ elif mode == "added":
+ database.cursor.execute(
+ f"SELECT * \
+FROM instances \
+ORDER BY first_seen DESC \
+LIMIT ?", [amount]
+ )
+ elif mode == "updated":
+ database.cursor.execute(
+ f"SELECT * \
+FROM instances \
+ORDER BY last_updated DESC \
+LIMIT ?", [amount]
+ )
+ else:
+ raise HTTPException(status_code=500, detail=f"mode='{mode}' is unsupported")
domainlist = database.cursor.fetchall()
return domainlist
@router.get(config.get("base_url") + "/api/top.json", response_class=JSONResponse)
-def api_index(request: Request, mode: str, value: str, amount: int):
+def api_index(request: Request, mode: str, value: str, amount: int) -> None:
if mode is None or value is None or amount is None:
raise HTTPException(status_code=500, detail="No filter specified")
elif amount > config.get("api_limit"):
return result
@router.get(config.get("base_url") + "/api/domain.json", response_class=JSONResponse)
-def api_domain(domain: str):
+def api_domain(domain: str) -> None:
if domain is None:
raise HTTPException(status_code=400, detail="Invalid request, parameter 'domain' missing")
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()):
+def api_mutual(domains: list[str] = Query()) -> None:
"""Return 200 if federation is open between the two, 4xx otherwise"""
database.cursor.execute(
"SELECT block_level FROM blocks " \
return JSONResponse(status_code=200, content={})
@router.get(config.get("base_url") + "/.well-known/nodeinfo", response_class=JSONResponse)
-def wellknown_nodeinfo(request: Request):
+def wellknown_nodeinfo(request: Request) -> None:
return JSONResponse(status_code=200, content={
"links": ({
"rel" : "http://nodeinfo.diaspora.software/ns/schema/1.0",
})
@router.get(config.get("base_url") + "/nodeinfo/1.0", response_class=JSONResponse)
-def nodeinfo_1_0(request: Request):
+def nodeinfo_1_0(request: Request) -> None:
return JSONResponse(status_code=200, content={
"version": "1.0",
"software": {
})
@router.get(config.get("base_url") + "/api/v1/instance/domain_blocks", response_class=JSONResponse)
-def api_domain_blocks(request: Request):
+def api_domain_blocks(request: Request) -> None:
blocked = blacklist.get_all()
blocking = list()
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):
+def api_peers(request: Request) -> None:
database.cursor.execute("SELECT domain FROM instances WHERE nodeinfo_url IS NOT NULL")
peers = list()
return JSONResponse(status_code=200, content=peers)
@router.get(config.get("base_url") + "/scoreboard")
-def scoreboard(request: Request, mode: str, amount: int):
+def scoreboard(request: Request, mode: str, amount: int) -> None:
if mode == "":
raise HTTPException(status_code=400, detail="No mode specified")
elif amount <= 0:
})
@router.get(config.get("base_url") + "/list")
-def list_domains(request: Request, mode: str, value: str, amount: int = config.get("api_limit")):
+def list_domains(request: Request, mode: str, value: str, amount: int = config.get("api_limit")) -> None:
if mode == "detection_mode" and not instances.valid(value, "detection_mode"):
raise HTTPException(status_code=500, detail="Invalid detection mode provided")
})
@router.get(config.get("base_url") + "/top")
-def top(request: Request, mode: str, value: str, amount: int = config.get("api_limit")):
+def top(request: Request, mode: str, value: str, amount: int = config.get("api_limit")) -> None:
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 domain_helper.is_wanted(value):
})
@router.get(config.get("base_url") + "/infos")
-def infos(request: Request, domain: str):
+def infos(request: Request, domain: str) -> None:
if domain is None:
raise HTTPException(status_code=400, detail="Invalid request, parameter 'domain' missing")
})
@router.get(config.get("base_url") + "/rss")
-def rss(request: Request, domain: str = None):
+def rss(request: Request, domain: str = None) -> None:
if domain is not None:
domain = tidyup.domain(domain).encode("idna").decode("utf-8")
wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
punycode = domain.encode("idna").decode("utf-8")
- database.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 ?", [
+ database.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 ?", [
domain,
"*." + domain, wildchar,
utils.get_hash(domain),
config.get("rss_limit")
])
else:
- database.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks ORDER BY first_seen DESC LIMIT ?", [config.get("rss_limit")])
+ database.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen \
+FROM blocks \
+ORDER BY first_seen DESC \
+LIMIT ?", [config.get("rss_limit")])
result = database.cursor.fetchall()
blocklist = []
})
@router.get(config.get("base_url") + "/robots.txt", response_class=PlainTextResponse)
-def robots(request: Request):
+def robots(request: Request) -> None:
return templates.TemplateResponse("views/robots.txt", {
"request" : request,
"base_url": config.get("base_url")
})
@router.get(config.get("base_url") + "/")
-def index(request: Request):
+def index(request: Request) -> None:
# Get info
response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/info.json")
})
if __name__ == "__main__":
- uvicorn.run("daemon:router", host=config.get("host"), port=config.get("port"), log_level=config.get("log_level"), proxy_headers=True)
+ 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