+++ /dev/null
-# Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
-# Copyright (C) 2023 Free Software Foundation
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as published
-# by the Free Software Foundation, either version 3 of the License, or
-# (at your option) any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU Affero General Public License for more details.
-#
-# You should have received a copy of the GNU Affero General Public License
-# along with this program. If not, see <https://www.gnu.org/licenses/>.
-
-import re
-
-from datetime import datetime
-from email.utils import format_datetime
-from pathlib import Path
-
-import fastapi
-from fastapi import Request, HTTPException, Query
-from fastapi.responses import JSONResponse
-from fastapi.responses import PlainTextResponse
-from fastapi.staticfiles import StaticFiles
-from fastapi.templating import Jinja2Templates
-
-import uvicorn
-import requests
-
-from fba import database
-from fba import utils
-
-from fba.helpers import config
-from fba.helpers import tidyup
-
-from fba.http import network
-
-from fba.models import blocks
-
-router = fastapi.FastAPI(docs_url=config.get("base_url") + "/docs", redoc_url=config.get("base_url") + "/redoc")
-router.mount(
- "/static",
- StaticFiles(directory=Path(__file__).parent.absolute() / "static"),
- name="static",
-)
-
-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')), (SELECT COUNT(blocker) FROM blocks), (SELECT COUNT(domain) FROM instances WHERE last_error_details IS NOT NULL)")
- row = database.cursor.fetchone()
-
- return {
- "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):
- if amount > config.get("api_limit"):
- raise HTTPException(status_code=400, detail="Too many results")
-
- if mode == "blocked":
- database.cursor.execute("SELECT blocked, COUNT(blocked) AS score FROM blocks GROUP BY blocked ORDER BY score DESC LIMIT ?", [amount])
- elif mode == "blocker":
- 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 == "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":
- database.cursor.execute("SELECT command, COUNT(domain) AS score FROM instances WHERE command IS NOT NULL GROUP BY command ORDER BY score DESC, command ASC LIMIT ?", [amount])
- elif mode == "error_code":
- database.cursor.execute("SELECT last_status_code, COUNT(domain) AS score FROM instances WHERE last_status_code IS NOT NULL AND last_status_code != '200' GROUP BY last_status_code ORDER BY score DESC LIMIT ?", [amount])
- 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])
- 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])
- 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:
- raise HTTPException(status_code=400, detail="No filter specified")
-
- scores = list()
-
- for domain, score in database.cursor.fetchall():
- scores.append({
- "domain": domain,
- "score" : round(score)
- })
-
- return scores
-
-@router.get(config.get("base_url") + "/api/index.json", response_class=JSONResponse)
-def api_index(request: Request, mode: str, value: str, amount: int):
- 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")
-
- domain = wildchar = punycode = reason = None
-
- if mode == "block_level":
- database.cursor.execute(
- "SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE block_level = ? LIMIT ?", [value, amount]
- )
- elif mode in ["domain", "reverse"]:
- domain = tidyup.domain(value)
- if not utils.is_domain_wanted(domain):
- raise HTTPException(status_code=500, detail=f"domain='{domain}' is not wanted")
-
- wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
- punycode = domain.encode('idna').decode('utf-8')
- elif mode == "reason":
- reason = re.sub("(%|_)", "", tidyup.reason(value))
- if len(reason) < 3:
- raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
-
- if mode == "domain":
- 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 ASC LIMIT ?",
- [
- domain,
- "*." + domain,
- wildchar,
- utils.get_hash(domain),
- punycode,
- "*." + punycode,
- amount
- ]
- )
- elif mode == "reverse":
- database.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen \
-FROM blocks \
-WHERE blocker = ? OR blocker = ? OR blocker = ? OR blocker = ? OR blocker = ? OR blocker = ? \
-ORDER BY first_seen ASC \
-LIMIT ?", [
- domain,
- "*." + domain,
- wildchar,
- utils.get_hash(domain),
- punycode,
- "*." + punycode,
- amount
- ])
- elif mode == "reason":
- database.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen \
-FROM blocks \
-WHERE reason LIKE ? AND reason != '' \
-ORDER BY first_seen ASC \
-LIMIT ?", [
- "%" + reason + "%",
- amount
- ])
-
- blocklist = database.cursor.fetchall()
-
- result = {}
- for blocker, blocked, block_level, reason, first_seen, last_seen in blocklist:
- if reason is not None and reason != "":
- reason = reason.replace(",", " ").replace(" ", " ")
-
- entry = {
- "blocker" : blocker,
- "blocked" : blocked,
- "reason" : reason,
- "first_seen": first_seen,
- "last_seen" : last_seen
- }
-
- if block_level in result:
- result[block_level].append(entry)
- else:
- result[block_level] = [entry]
-
- return result
-
-@router.get(config.get("base_url") + "/api/mutual.json", response_class=JSONResponse)
-def api_mutual(domains: list[str] = Query()):
- """Return 200 if federation is open between the two, 4xx otherwise"""
- database.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],
- },
- )
- response = database.cursor.fetchone()
-
- if response is not None:
- # Blocks found
- return JSONResponse(status_code=418, content={})
-
- # No known blocks
- return JSONResponse(status_code=200, content={})
-
-@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 response is None:
- raise HTTPException(status_code=500, detail="Could not determine scores")
- elif not response.ok:
- raise HTTPException(status_code=response.status_code, detail=response.text)
-
- return templates.TemplateResponse("views/scoreboard.html", {
- "base_url" : config.get("base_url"),
- "slogan" : config.get("slogan"),
- "theme" : config.get("theme"),
- "request" : request,
- "scoreboard": True,
- "mode" : mode,
- "amount" : amount,
- "scores" : network.json_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") + "/top")
-def top(request: Request, mode: str, value: str, amount: int = config.get("api_limit")):
- 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)
- elif mode == "" or value == "" or amount == 0:
- raise HTTPException(status_code=500, detail="Parameter mode, value and amount must always be set")
- elif amount > config.get("api_limit"):
- raise HTTPException(status_code=500, detail=f"amount='{amount}' is to big")
-
- info = response.json()
- response = None
- blocklist = list()
-
- if mode == "block_level" and not blocks.is_valid_level(value):
- raise HTTPException(status_code=500, detail="Invalid block level provided")
- elif mode in ["domain", "reverse"] and not utils.is_domain_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/index.json?mode={mode}&value={value}&amount={amount}")
-
- if response is not None:
- blocklist = response.json()
-
- found = 0
- for block_level in blocklist:
- for block in blocklist[block_level]:
- block["first_seen"] = datetime.utcfromtimestamp(block["first_seen"]).strftime(config.get("timestamp_format"))
- block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime(config.get("timestamp_format"))
- found = found + 1
-
- return templates.TemplateResponse("views/top.html", {
- "request" : request,
- "mode" : mode if response is not None else None,
- "value" : value if response is not None else None,
- "amount" : amount if response is not None else None,
- "found" : found,
- "blocklist": blocklist,
- "info" : info,
- "theme" : config.get("theme"),
- })
-
-@router.get(config.get("base_url") + "/rss")
-def rss(request: Request, domain: str = None):
- if domain is not None:
- domain = tidyup.domain(domain)
-
- 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 ?", [
- domain,
- "*." + domain, wildchar,
- utils.get_hash(domain),
- punycode,
- "*." + punycode,
- 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")])
-
- result = database.cursor.fetchall()
- blocklist = []
-
- for row in result:
- blocklist.append({
- "blocker" : row[0],
- "blocked" : row[1],
- "block_level": row[2],
- "reason" : "Provided reason: '" + row[3] + "'" if row[3] is not None and row[3] != "" else "No reason provided.",
- "first_seen" : format_datetime(datetime.fromtimestamp(row[4])),
- "last_seen" : format_datetime(datetime.fromtimestamp(row[5])),
- })
-
- return templates.TemplateResponse("views/rss.xml", {
- "request" : request,
- "timestamp": format_datetime(datetime.now()),
- "domain" : domain,
- "hostname" : config.get("hostname"),
- "blocks" : blocklist
- }, headers={
- "Content-Type": "routerlication/rss+xml"
- })
-
-@router.get(config.get("base_url") + "/robots.txt", response_class=PlainTextResponse)
-def robots(request: Request):
- return templates.TemplateResponse("views/robots.txt", {
- "request" : request,
- "base_url": config.get("base_url")
- })
-
-if __name__ == "__main__":
- uvicorn.run("api:router", host=config.get("host"), port=config.get("port"), log_level=config.get("log_level"))
if [ -z "${CHECK}" ]
then
echo "$0: Daemon isn't running, starting in background ..."
- screen -dmS daemon python3 api.py
+ screen -dmS daemon python3 daemon.py
fi
RestartSec=10
User=fba
WorkingDirectory=/opt/fedi-block-api
-ExecStart=python3 api.py
+ExecStart=python3 daemon.py
[Install]
WantedBy=multi-user.target
--- /dev/null
+# Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
+# Copyright (C) 2023 Free Software Foundation
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+import re
+
+from datetime import datetime
+from email.utils import format_datetime
+from pathlib import Path
+
+import fastapi
+from fastapi import Request, HTTPException, Query
+from fastapi.responses import JSONResponse
+from fastapi.responses import PlainTextResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.templating import Jinja2Templates
+
+import uvicorn
+import requests
+
+from fba import database
+from fba import utils
+
+from fba.helpers import config
+from fba.helpers import tidyup
+
+from fba.http import network
+
+from fba.models import blocks
+
+router = fastapi.FastAPI(docs_url=config.get("base_url") + "/docs", redoc_url=config.get("base_url") + "/redoc")
+router.mount(
+ "/static",
+ StaticFiles(directory=Path(__file__).parent.absolute() / "static"),
+ name="static",
+)
+
+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')), (SELECT COUNT(blocker) FROM blocks), (SELECT COUNT(domain) FROM instances WHERE last_error_details IS NOT NULL)")
+ row = database.cursor.fetchone()
+
+ return {
+ "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):
+ if amount > config.get("api_limit"):
+ raise HTTPException(status_code=400, detail="Too many results")
+
+ if mode == "blocked":
+ database.cursor.execute("SELECT blocked, COUNT(blocked) AS score FROM blocks GROUP BY blocked ORDER BY score DESC LIMIT ?", [amount])
+ elif mode == "blocker":
+ 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 == "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":
+ database.cursor.execute("SELECT command, COUNT(domain) AS score FROM instances WHERE command IS NOT NULL GROUP BY command ORDER BY score DESC, command ASC LIMIT ?", [amount])
+ elif mode == "error_code":
+ database.cursor.execute("SELECT last_status_code, COUNT(domain) AS score FROM instances WHERE last_status_code IS NOT NULL AND last_status_code != '200' GROUP BY last_status_code ORDER BY score DESC LIMIT ?", [amount])
+ 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])
+ 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])
+ 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:
+ raise HTTPException(status_code=400, detail="No filter specified")
+
+ scores = list()
+
+ for domain, score in database.cursor.fetchall():
+ scores.append({
+ "domain": domain,
+ "score" : round(score)
+ })
+
+ return scores
+
+@router.get(config.get("base_url") + "/api/index.json", response_class=JSONResponse)
+def api_index(request: Request, mode: str, value: str, amount: int):
+ 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")
+
+ domain = wildchar = punycode = reason = None
+
+ if mode == "block_level":
+ database.cursor.execute(
+ "SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE block_level = ? LIMIT ?", [value, amount]
+ )
+ elif mode in ["domain", "reverse"]:
+ domain = tidyup.domain(value)
+ if not utils.is_domain_wanted(domain):
+ raise HTTPException(status_code=500, detail=f"domain='{domain}' is not wanted")
+
+ wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
+ punycode = domain.encode('idna').decode('utf-8')
+ elif mode == "reason":
+ reason = re.sub("(%|_)", "", tidyup.reason(value))
+ if len(reason) < 3:
+ raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
+
+ if mode == "domain":
+ 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 ASC LIMIT ?",
+ [
+ domain,
+ "*." + domain,
+ wildchar,
+ utils.get_hash(domain),
+ punycode,
+ "*." + punycode,
+ amount
+ ]
+ )
+ elif mode == "reverse":
+ database.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen \
+FROM blocks \
+WHERE blocker = ? OR blocker = ? OR blocker = ? OR blocker = ? OR blocker = ? OR blocker = ? \
+ORDER BY first_seen ASC \
+LIMIT ?", [
+ domain,
+ "*." + domain,
+ wildchar,
+ utils.get_hash(domain),
+ punycode,
+ "*." + punycode,
+ amount
+ ])
+ elif mode == "reason":
+ database.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen \
+FROM blocks \
+WHERE reason LIKE ? AND reason != '' \
+ORDER BY first_seen ASC \
+LIMIT ?", [
+ "%" + reason + "%",
+ amount
+ ])
+
+ blocklist = database.cursor.fetchall()
+
+ result = {}
+ for blocker, blocked, block_level, reason, first_seen, last_seen in blocklist:
+ if reason is not None and reason != "":
+ reason = reason.replace(",", " ").replace(" ", " ")
+
+ entry = {
+ "blocker" : blocker,
+ "blocked" : blocked,
+ "reason" : reason,
+ "first_seen": first_seen,
+ "last_seen" : last_seen
+ }
+
+ if block_level in result:
+ result[block_level].append(entry)
+ else:
+ result[block_level] = [entry]
+
+ return result
+
+@router.get(config.get("base_url") + "/api/mutual.json", response_class=JSONResponse)
+def api_mutual(domains: list[str] = Query()):
+ """Return 200 if federation is open between the two, 4xx otherwise"""
+ database.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],
+ },
+ )
+ response = database.cursor.fetchone()
+
+ if response is not None:
+ # Blocks found
+ return JSONResponse(status_code=418, content={})
+
+ # No known blocks
+ return JSONResponse(status_code=200, content={})
+
+@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 response is None:
+ raise HTTPException(status_code=500, detail="Could not determine scores")
+ elif not response.ok:
+ raise HTTPException(status_code=response.status_code, detail=response.text)
+
+ return templates.TemplateResponse("views/scoreboard.html", {
+ "base_url" : config.get("base_url"),
+ "slogan" : config.get("slogan"),
+ "theme" : config.get("theme"),
+ "request" : request,
+ "scoreboard": True,
+ "mode" : mode,
+ "amount" : amount,
+ "scores" : network.json_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") + "/top")
+def top(request: Request, mode: str, value: str, amount: int = config.get("api_limit")):
+ 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)
+ elif mode == "" or value == "" or amount == 0:
+ raise HTTPException(status_code=500, detail="Parameter mode, value and amount must always be set")
+ elif amount > config.get("api_limit"):
+ raise HTTPException(status_code=500, detail=f"amount='{amount}' is to big")
+
+ info = response.json()
+ response = None
+ blocklist = list()
+
+ if mode == "block_level" and not blocks.is_valid_level(value):
+ raise HTTPException(status_code=500, detail="Invalid block level provided")
+ elif mode in ["domain", "reverse"] and not utils.is_domain_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/index.json?mode={mode}&value={value}&amount={amount}")
+
+ if response is not None:
+ blocklist = response.json()
+
+ found = 0
+ for block_level in blocklist:
+ for block in blocklist[block_level]:
+ block["first_seen"] = datetime.utcfromtimestamp(block["first_seen"]).strftime(config.get("timestamp_format"))
+ block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime(config.get("timestamp_format"))
+ found = found + 1
+
+ return templates.TemplateResponse("views/top.html", {
+ "request" : request,
+ "mode" : mode if response is not None else None,
+ "value" : value if response is not None else None,
+ "amount" : amount if response is not None else None,
+ "found" : found,
+ "blocklist": blocklist,
+ "info" : info,
+ "theme" : config.get("theme"),
+ })
+
+@router.get(config.get("base_url") + "/rss")
+def rss(request: Request, domain: str = None):
+ if domain is not None:
+ domain = tidyup.domain(domain)
+
+ 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 ?", [
+ domain,
+ "*." + domain, wildchar,
+ utils.get_hash(domain),
+ punycode,
+ "*." + punycode,
+ 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")])
+
+ result = database.cursor.fetchall()
+ blocklist = []
+
+ for row in result:
+ blocklist.append({
+ "blocker" : row[0],
+ "blocked" : row[1],
+ "block_level": row[2],
+ "reason" : "Provided reason: '" + row[3] + "'" if row[3] is not None and row[3] != "" else "No reason provided.",
+ "first_seen" : format_datetime(datetime.fromtimestamp(row[4])),
+ "last_seen" : format_datetime(datetime.fromtimestamp(row[5])),
+ })
+
+ return templates.TemplateResponse("views/rss.xml", {
+ "request" : request,
+ "timestamp": format_datetime(datetime.now()),
+ "domain" : domain,
+ "hostname" : config.get("hostname"),
+ "blocks" : blocklist
+ }, headers={
+ "Content-Type": "routerlication/rss+xml"
+ })
+
+@router.get(config.get("base_url") + "/robots.txt", response_class=PlainTextResponse)
+def robots(request: Request):
+ return templates.TemplateResponse("views/robots.txt", {
+ "request" : request,
+ "base_url": config.get("base_url")
+ })
+
+if __name__ == "__main__":
+ uvicorn.run("api:router", host=config.get("host"), port=config.get("port"), log_level=config.get("log_level"))
projects / fba.git / commitdiff