# 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/>.
+from datetime import datetime
+from email import utils
+
+import re
+
from fastapi import Request, HTTPException, Query
from fastapi.responses import JSONResponse
from fastapi.responses import PlainTextResponse
from fastapi.templating import Jinja2Templates
-from datetime import datetime
-from email import utils
import fastapi
import uvicorn
import requests
-import re
-import fba
+import validators
+
+from fba import config
+from fba import fba
+from fba import network
-router = fastapi.FastAPI(docs_url=fba.config["base_url"] + "/docs", redoc_url=fba.config["base_url"] + "/redoc")
+from fba.helpers import tidyup
+
+router = fastapi.FastAPI(docs_url=config.get("base_url") + "/docs", redoc_url=config.get("base_url") + "/redoc")
templates = Jinja2Templates(directory="templates")
-@router.get(fba.config["base_url"] + "/api/info.json", response_class=JSONResponse)
-def info():
- fba.cursor.execute("SELECT (SELECT COUNT(domain) FROM instances), (SELECT COUNT(domain) FROM instances WHERE software IN ('pleroma', 'mastodon', 'misskey', 'gotosocial', 'friendica', 'bookwyrm', 'takahe', 'peertube')), (SELECT COUNT(blocker) FROM blocks), (SELECT COUNT(domain) FROM instances WHERE last_status_code IS NOT NULL)")
- known, indexed, blocks, errorous = fba.cursor.fetchone()
+@router.get(config.get("base_url") + "/api/info.json", response_class=JSONResponse)
+def api_info():
+ fba.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 = fba.cursor.fetchone()
return {
- "known_instances" : known,
- "indexed_instances" : indexed,
- "blocks_recorded" : blocks,
- "errorous_instances": errorous,
- "slogan" : fba.config["slogan"]
+ "known_instances" : row[0],
+ "indexed_instances" : row[1],
+ "blocks_recorded" : row[2],
+ "errorous_instances": row[3],
+ "slogan" : config.get("slogan")
}
-@router.get(fba.config["base_url"] + "/api/top.json", response_class=JSONResponse)
-def top(blocked: int = None, blockers: int = None, reference: int = None, software: int = None):
- if blocked != None:
- if blocked > 500:
- raise HTTPException(status_code=400, detail="Too many results")
- fba.cursor.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")
- fba.cursor.execute("SELECT blocker, COUNT(blocker) FROM blocks WHERE block_level = 'reject' GROUP BY blocker ORDER BY COUNT(blocker) DESC LIMIT ?", [blockers])
- elif reference != None:
- if reference > 500:
- raise HTTPException(status_code=400, detail="Too many results")
- fba.cursor.execute("SELECT origin, COUNT(domain) FROM instances WHERE software IS NOT NULL GROUP BY origin ORDER BY COUNT(domain) DESC LIMIT ?", [reference])
- elif software != None:
- if software > 500:
- raise HTTPException(status_code=400, detail="Too many results")
- fba.cursor.execute("SELECT software, COUNT(domain) FROM instances WHERE software IS NOT NULL GROUP BY software ORDER BY COUNT(domain) DESC, software ASC LIMIT ?", [software])
+@router.get(config.get("base_url") + "/api/top.json", response_class=JSONResponse)
+def api_top(mode: str, amount: int):
+ if amount > 500:
+ raise HTTPException(status_code=400, detail="Too many results")
+
+ if mode == "blocked":
+ fba.cursor.execute("SELECT blocked, COUNT(blocked) AS score FROM blocks WHERE block_level = 'reject' GROUP BY blocked ORDER BY score DESC LIMIT ?", [amount])
+ elif mode == "blocker":
+ fba.cursor.execute("SELECT blocker, COUNT(blocker) AS score FROM blocks WHERE block_level = 'reject' GROUP BY blocker ORDER BY score DESC LIMIT ?", [amount])
+ elif mode == "reference":
+ fba.cursor.execute("SELECT origin, COUNT(domain) AS score FROM instances WHERE software IS NOT NULL GROUP BY origin ORDER BY score DESC LIMIT ?", [amount])
+ elif mode == "software":
+ fba.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":
+ fba.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":
+ fba.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])
else:
raise HTTPException(status_code=400, detail="No filter specified")
- scores = fba.cursor.fetchall()
-
- scoreboard = []
+ scores = list()
- for domain, highscore in scores:
- scoreboard.append({
- "domain" : domain,
- "highscore": highscore
+ for domain, score in fba.cursor.fetchall():
+ scores.append({
+ "domain": domain,
+ "score" : score
})
- return scoreboard
+ return scores
-@router.get(fba.config["base_url"] + "/api/index.json", response_class=JSONResponse)
-def blocked(domain: str = None, reason: str = None, reverse: str = None):
- if domain == None and reason == None and reverse == None:
+@router.get(config.get("base_url") + "/api/index.json", response_class=JSONResponse)
+def api_blocked(domain: str = None, reason: str = None, reverse: str = None):
+ if domain is None and reason is None and reverse is None:
raise HTTPException(status_code=400, detail="No filter specified")
- if reason != None:
- reason = re.sub("(%|_)", "", reason)
+ if reason is not None:
+ reason = re.sub("(%|_)", "", tidyup.reason(reason))
if len(reason) < 3:
raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
- if domain != None:
+ if domain is not None:
+ domain = tidyup.domain(domain)
+ if not validators.domain(domain):
+ raise HTTPException(status_code=500, detail="Invalid domain")
+
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 ASC",
(domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
- elif reverse != None:
+ elif reverse is not None:
+ reverse = tidyup.domain(reverse)
+ if not validators.domain(reverse):
+ raise HTTPException(status_code=500, detail="Invalid domain")
+
fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE blocker = ? ORDER BY first_seen ASC", [reverse])
else:
fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE reason like ? AND reason != '' ORDER BY first_seen ASC", ["%" + reason + "%"])
- blocks = fba.cursor.fetchall()
+ blocklist = fba.cursor.fetchall()
result = {}
- for blocker, blocked, block_level, reason, first_seen, last_seen in blocks:
+ 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,
"first_seen": first_seen,
"last_seen" : last_seen
}
+
if block_level in result:
result[block_level].append(entry)
else:
return result
-@router.get(fba.config["base_url"] + "/api/mutual.json", response_class=JSONResponse)
-def mutual(domains: list[str] = Query()):
+@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"""
fba.cursor.execute(
"SELECT block_level FROM blocks " \
"bw": "*." + domains[1],
},
)
- res = fba.cursor.fetchone()
+ response = fba.cursor.fetchone()
- if res is not None:
+ if response is not None:
# Blocks found
return JSONResponse(status_code=418, content={})
# No known blocks
return JSONResponse(status_code=200, content={})
-@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.json?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.json?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.json?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.json?software={software}")
+@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/top.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/top.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/top.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/top.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/top.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/top.json?mode=error_code&amount={amount}")
else:
raise HTTPException(status_code=400, detail="No filter specified")
- if res == None:
+ if response is 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)
+ elif not response.ok:
+ raise HTTPException(status_code=response.status_code, detail=response.text)
- return templates.TemplateResponse("scoreboard.html", {
- "base_url" : fba.config["base_url"],
- "slogan" : fba.config["slogan"],
+ return templates.TemplateResponse("views/scoreboard.html", {
+ "base_url" : config.get("base_url"),
+ "slogan" : config.get("slogan"),
"request" : request,
"scoreboard": True,
- "blockers" : blockers,
- "blocked" : blocked,
- "reference" : reference,
- "software" : software,
- "scores" : res.json()
+ "mode" : mode,
+ "amount" : amount,
+ "scores" : network.json_from_response(response)
})
-@router.get(fba.config["base_url"] + "/")
-def index(request: Request, domain: str = None, reason: str = None, reverse: str = None):
+@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,
+ "info" : response.json()
+ })
+
+@router.get(config.get("base_url") + "/top")
+def top(request: Request, domain: str = None, reason: str = None, reverse: str = None):
if domain == "" or reason == "" or reverse == "":
- return fastapi.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.json")
-
- 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/index.json?domain={domain}")
- elif reason != None:
- blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/index.json?reason={reason}")
- elif reverse != None:
- blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/index.json?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", {
+ raise HTTPException(status_code=500, detail="Insufficient parameter provided")
+
+ 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)
+
+ info = response.json()
+ response = None
+
+ if domain is not None:
+ domain = tidyup.domain(domain)
+ if not validators.domain(domain):
+ raise HTTPException(status_code=500, detail="Invalid domain")
+
+ response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/index.json?domain={domain}")
+ elif reason is not None:
+ response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/index.json?reason={reason}")
+ elif reverse is not None:
+ reverse = tidyup.domain(reverse)
+ if not validators.domain(reverse):
+ raise HTTPException(status_code=500, detail="Invalid domain")
+
+ response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/index.json?reverse={reverse}")
+
+ if response is not None:
+ if not response.ok:
+ raise HTTPException(status_code=response.status_code, detail=response.text)
+
+ blocklist = response.json()
+
+ 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"))
+
+ return templates.TemplateResponse("views/top.html", {
"request": request,
"domain" : domain,
- "blocks" : blocks,
+ "blocks" : blocklist,
"reason" : reason,
"reverse": reverse,
"info" : info
})
-@router.get(fba.config["base_url"] + "/rss")
+@router.get(config.get("base_url") + "/rss")
def rss(request: Request, domain: str = None):
- if domain != None:
+ if domain is not None:
+ domain = tidyup.domain(domain)
+
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))
+ 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 ?", [
+ domain,
+ "*." + domain, wildchar,
+ fba.get_hash(domain),
+ punycode,
+ "*." + punycode,
+ config.get("rss_limit")
+ ])
else:
- fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks ORDER BY first_seen DESC LIMIT 50")
+ fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks ORDER BY first_seen DESC LIMIT ?", [config.get("rss_limit")])
- blocks = fba.cursor.fetchall()
+ result = fba.cursor.fetchall()
- result = []
- for blocker, blocked, block_level, reason, first_seen, last_seen in blocks:
+ blocklist = []
+ for blocker, blocked, block_level, reason, first_seen, last_seen in result:
first_seen = utils.format_datetime(datetime.fromtimestamp(first_seen))
- if reason == None or reason == '':
+ if reason is None or reason == "":
reason = "No reason provided."
else:
reason = "Provided reason: '" + reason + "'"
- result.append({
+ blocklist.append({
"blocker" : blocker,
"blocked" : blocked,
"block_level": block_level,
"reason" : reason,
- "first_seen" : first_seen
+ "first_seen" : first_seen,
+ "last_seen" : last_seen,
})
- timestamp = utils.format_datetime(datetime.now())
-
return templates.TemplateResponse("rss.xml", {
"request" : request,
- "timestamp": timestamp,
+ "timestamp": utils.format_datetime(datetime.now()),
"domain" : domain,
- "blocks" : result
+ "hostname" : config.get("hostname"),
+ "blocks" : blocklist
}, headers={
"Content-Type": "routerlication/rss+xml"
})
-@router.get(fba.config["base_url"] + "/robots.txt", response_class=PlainTextResponse)
+@router.get(config.get("base_url") + "/robots.txt", response_class=PlainTextResponse)
def robots(request: Request):
return templates.TemplateResponse("robots.txt", {
"request" : request,
- "base_url": fba.config["base_url"]
+ "base_url": config.get("base_url")
})
if __name__ == "__main__":
- uvicorn.run("api:router", host=fba.config["host"], port=fba.config["port"], log_level=fba.config["log_level"])
+ uvicorn.run("api:router", host=config.get("host"), port=config.get("port"), log_level=config.get("log_level"))