api.py

   1 # Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
   2 # Copyright (C) 2023 Free Software Foundation
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published
   6 # by the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
  16
  17 from fastapi import Request, HTTPException, responses, Query
  18 from fastapi.templating import Jinja2Templates
  19 from datetime import datetime
  20 from email import utils
  21
  22 import fastapi
  23 import uvicorn
  24 import requests
  25 import re
  26 import fba
  27
  28 app = fastapi.FastAPI(docs_url=fba.config["base_url"] + "/docs", redoc_url=fba.config["base_url"] + "/redoc")
  29 templates = Jinja2Templates(directory=".")
  30
  31 @app.get(fba.config["base_url"] + "/api/info")
  32 def info():
  33     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)")
  34     known, indexed, blocks, errorous = fba.cursor.fetchone()
  35
  36     return {
  37         "known_instances"   : known,
  38         "indexed_instances" : indexed,
  39         "blocks_recorded"   : blocks,
  40         "errorous_instances": errorous,
  41         "slogan"            : fba.config["slogan"]
  42     }
  43
  44 @app.get(fba.config["base_url"] + "/api/top")
  45 def top(blocked: int = None, blockers: int = None, reference: int = None, software: int = None):
  46     if blocked != None:
  47         if blocked > 500:
  48             raise HTTPException(status_code=400, detail="Too many results")
  49         fba.cursor.execute("SELECT blocked, COUNT(blocked) FROM blocks WHERE block_level = 'reject' GROUP BY blocked ORDER BY COUNT(blocked) DESC LIMIT ?", [blocked])
  50     elif blockers != None:
  51         if blockers > 500:
  52             raise HTTPException(status_code=400, detail="Too many results")
  53         fba.cursor.execute("SELECT blocker, COUNT(blocker) FROM blocks WHERE block_level = 'reject' GROUP BY blocker ORDER BY COUNT(blocker) DESC LIMIT ?", [blockers])
  54     elif reference != None:
  55         if reference > 500:
  56             raise HTTPException(status_code=400, detail="Too many results")
  57         fba.cursor.execute("SELECT origin, COUNT(domain) FROM instances GROUP BY origin ORDER BY COUNT(domain) DESC LIMIT ?", [reference])
  58     elif software != None:
  59         if software > 500:
  60             raise HTTPException(status_code=400, detail="Too many results")
  61         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])
  62     else:
  63         raise HTTPException(status_code=400, detail="No filter specified")
  64
  65     scores = fba.cursor.fetchall()
  66
  67     scoreboard = []
  68
  69     for domain, highscore in scores:
  70         scoreboard.append({
  71             "domain"   : domain,
  72             "highscore": highscore
  73         })
  74
  75     return scoreboard
  76
  77 @app.get(fba.config["base_url"] + "/api")
  78 def blocked(domain: str = None, reason: str = None, reverse: str = None):
  79     if domain == None and reason == None and reverse == None:
  80         raise HTTPException(status_code=400, detail="No filter specified")
  81
  82     if reason != None:
  83         reason = re.sub("(%|_)", "", reason)
  84         if len(reason) < 3:
  85             raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
  86
  87     if domain != None:
  88         wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
  89         punycode = domain.encode('idna').decode('utf-8')
  90         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",
  91                   (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
  92     elif reverse != None:
  93         fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE blocker = ? ORDER BY first_seen ASC", [reverse])
  94     else:
  95         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 + "%"])
  96
  97     blocks = fba.cursor.fetchall()
  98
  99     result = {}
 100     for blocker, blocked, block_level, reason, first_seen, last_seen in blocks:
 101         entry = {
 102             "blocker"   : blocker,
 103             "blocked"   : blocked,
 104             "reason"    : reason,
 105             "first_seen": first_seen,
 106             "last_seen" : last_seen
 107         }
 108         if block_level in result:
 109             result[block_level].append(entry)
 110         else:
 111             result[block_level] = [entry]
 112
 113     return result
 114
 115 @app.get(fba.config["base_url"] + "/scoreboard")
 116 def index(request: Request, blockers: int = None, blocked: int = None, reference: int = None, software: int = None):
 117     scores = None
 118
 119     if blockers != None and blockers > 0:
 120         res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top?blockers={blockers}")
 121     elif blocked != None and blocked > 0:
 122         res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top?blocked={blocked}")
 123     elif reference != None and reference > 0:
 124         res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top?reference={reference}")
 125     elif software != None and software > 0:
 126         res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top?software={software}")
 127     else:
 128         raise HTTPException(status_code=400, detail="No filter specified")
 129
 130     if res == None:
 131         raise HTTPException(status_code=500, detail="Could not determine scores")
 132     elif not res.ok:
 133         raise HTTPException(status_code=res.status_code, detail=res.text)
 134
 135     return templates.TemplateResponse("index.html", {
 136         "base_url"  : fba.config["base_url"],
 137         "request"   : request,
 138         "scoreboard": True,
 139         "blockers"  : blockers,
 140         "blocked"   : blocked,
 141         "reference" : reference,
 142         "software"  : software,
 143         "scores"    : res.json()
 144     })
 145
 146 @app.get(fba.config["base_url"] + "/")
 147 def index(request: Request, domain: str = None, reason: str = None, reverse: str = None):
 148     if domain == "" or reason == "" or reverse == "":
 149         return responses.RedirectResponse("/")
 150
 151     info = None
 152     blocks = None
 153
 154     if domain == None and reason == None and reverse == None:
 155         info = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/info")
 156
 157         if not info.ok:
 158             raise HTTPException(status_code=info.status_code, detail=info.text)
 159
 160         info = info.json()
 161     elif domain != None:
 162         blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api?domain={domain}")
 163     elif reason != None:
 164         blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api?reason={reason}")
 165     elif reverse != None:
 166         blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api?reverse={reverse}")
 167
 168     if blocks != None:
 169         if not blocks.ok:
 170             raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
 171         blocks = blocks.json()
 172         for block_level in blocks:
 173             for block in blocks[block_level]:
 174                 block["first_seen"] = datetime.utcfromtimestamp(block["first_seen"]).strftime('%Y-%m-%d %H:%M')
 175                 block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime('%Y-%m-%d %H:%M')
 176
 177     return templates.TemplateResponse("index.html", {
 178         "request": request,
 179         "domain" : domain,
 180         "blocks" : blocks,
 181         "reason" : reason,
 182         "reverse": reverse,
 183         "info"   : info
 184     })
 185
 186 @app.get(fba.config["base_url"] + "/api/mutual")
 187 def mutual(domains: list[str] = Query()):
 188     """Return 200 if federation is open between the two, 4xx otherwise"""
 189     fba.cursor.execute(
 190         "SELECT block_level FROM blocks " \
 191         "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
 192         "AND block_level = 'reject' " \
 193         "LIMIT 1",
 194         {
 195             "a" : domains[0],
 196             "b" : domains[1],
 197             "aw": "*." + domains[0],
 198             "bw": "*." + domains[1],
 199         },
 200     )
 201     res = fba.cursor.fetchone()
 202
 203     if res is not None:
 204         # Blocks found
 205         return responses.JSONResponse(status_code=418, content={})
 206
 207     # No known blocks
 208     return responses.JSONResponse(status_code=200, content={})
 209
 210 @app.get(fba.config["base_url"] + "/rss")
 211 def rss(request: Request, domain: str = None):
 212     if domain != None:
 213         wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
 214         punycode = domain.encode('idna').decode('utf-8')
 215         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",
 216                   (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
 217     else:
 218         fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks ORDER BY first_seen DESC LIMIT 50")
 219
 220     blocks = fba.cursor.fetchall()
 221
 222     result = []
 223     for blocker, blocked, block_level, reason, first_seen, last_seen in blocks:
 224         first_seen = utils.format_datetime(datetime.fromtimestamp(first_seen))
 225         if reason == None or reason == '':
 226             reason = "No reason provided."
 227         else:
 228             reason = "Provided reason: '" + reason + "'"
 229
 230         result.append({
 231             "blocker"    : blocker,
 232             "blocked"    : blocked,
 233             "block_level": block_level,
 234             "reason"     : reason,
 235             "first_seen" : first_seen
 236         })
 237
 238     timestamp = utils.format_datetime(datetime.now())
 239
 240     return templates.TemplateResponse("rss.xml", {
 241         "request"  : request,
 242         "timestamp": timestamp,
 243         "domain"   : domain,
 244         "blocks"   : result
 245     }, headers={
 246         "Content-Type": "application/rss+xml"
 247     })
 248
 249 if __name__ == "__main__":
 250     uvicorn.run("api:app", host=fba.config["host"], port=fba.config["port"], log_level=fba.config["log_level"])