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, Query
  18 from fastapi.responses import JSONResponse
  19 from fastapi.responses import PlainTextResponse
  20 from fastapi.templating import Jinja2Templates
  21 from datetime import datetime
  22 from email import utils
  23
  24 import fastapi
  25 import uvicorn
  26 import requests
  27 import re
  28 from fba import *
  29
  30 router = fastapi.FastAPI(docs_url=config.get("base_url") + "/docs", redoc_url=config.get("base_url") + "/redoc")
  31 templates = Jinja2Templates(directory="templates")
  32
  33 @router.get(config.get("base_url") + "/api/info.json", response_class=JSONResponse)
  34 def info():
  35     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)")
  36     known, indexed, blocks, errorous = fba.cursor.fetchone()
  37
  38     return {
  39         "known_instances"   : known,
  40         "indexed_instances" : indexed,
  41         "blocks_recorded"   : blocks,
  42         "errorous_instances": errorous,
  43         "slogan"            : config.get("slogan")
  44     }
  45
  46 @router.get(config.get("base_url") + "/api/top.json", response_class=JSONResponse)
  47 def top(blocked: int = None, blockers: int = None, reference: int = None, software: int = None, originator: int = None):
  48     if blocked != None:
  49         if blocked > 500:
  50             raise HTTPException(status_code=400, detail="Too many results")
  51         fba.cursor.execute("SELECT blocked, COUNT(blocked) FROM blocks WHERE block_level = 'reject' GROUP BY blocked ORDER BY COUNT(blocked) DESC LIMIT ?", [blocked])
  52     elif blockers != None:
  53         if blockers > 500:
  54             raise HTTPException(status_code=400, detail="Too many results")
  55         fba.cursor.execute("SELECT blocker, COUNT(blocker) FROM blocks WHERE block_level = 'reject' GROUP BY blocker ORDER BY COUNT(blocker) DESC LIMIT ?", [blockers])
  56     elif reference != None:
  57         if reference > 500:
  58             raise HTTPException(status_code=400, detail="Too many results")
  59         fba.cursor.execute("SELECT origin, COUNT(domain) FROM instances WHERE software IS NOT NULL GROUP BY origin ORDER BY COUNT(domain) DESC LIMIT ?", [reference])
  60     elif software != None:
  61         if software > 500:
  62             raise HTTPException(status_code=400, detail="Too many results")
  63         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])
  64     elif originator != None:
  65         if originator > 500:
  66             raise HTTPException(status_code=400, detail="Too many results")
  67         fba.cursor.execute("SELECT originator, COUNT(domain) FROM instances WHERE originator IS NOT NULL GROUP BY originator ORDER BY COUNT(domain) DESC, originator ASC LIMIT ?", [originator])
  68     else:
  69         raise HTTPException(status_code=400, detail="No filter specified")
  70
  71     scores = fba.cursor.fetchall()
  72
  73     scoreboard = []
  74
  75     for domain, highscore in scores:
  76         scoreboard.append({
  77             "domain"   : domain,
  78             "highscore": highscore
  79         })
  80
  81     return scoreboard
  82
  83 @router.get(config.get("base_url") + "/api/index.json", response_class=JSONResponse)
  84 def blocked(domain: str = None, reason: str = None, reverse: str = None):
  85     if domain == None and reason == None and reverse == None:
  86         raise HTTPException(status_code=400, detail="No filter specified")
  87
  88     if reason != None:
  89         reason = re.sub("(%|_)", "", reason)
  90         if len(reason) < 3:
  91             raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
  92
  93     if domain != None:
  94         wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
  95         punycode = domain.encode('idna').decode('utf-8')
  96         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",
  97                   (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
  98     elif reverse != None:
  99         fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE blocker = ? ORDER BY first_seen ASC", [reverse])
 100     else:
 101         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 + "%"])
 102
 103     blocks = fba.cursor.fetchall()
 104
 105     result = {}
 106     for blocker, blocked, block_level, reason, first_seen, last_seen in blocks:
 107         entry = {
 108             "blocker"   : blocker,
 109             "blocked"   : blocked,
 110             "reason"    : reason,
 111             "first_seen": first_seen,
 112             "last_seen" : last_seen
 113         }
 114         if block_level in result:
 115             result[block_level].append(entry)
 116         else:
 117             result[block_level] = [entry]
 118
 119     return result
 120
 121 @router.get(config.get("base_url") + "/api/mutual.json", response_class=JSONResponse)
 122 def mutual(domains: list[str] = Query()):
 123     """Return 200 if federation is open between the two, 4xx otherwise"""
 124     fba.cursor.execute(
 125         "SELECT block_level FROM blocks " \
 126         "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
 127         "AND block_level = 'reject' " \
 128         "LIMIT 1",
 129         {
 130             "a" : domains[0],
 131             "b" : domains[1],
 132             "aw": "*." + domains[0],
 133             "bw": "*." + domains[1],
 134         },
 135     )
 136     response = fba.cursor.fetchone()
 137
 138     if response is not None:
 139         # Blocks found
 140         return JSONResponse(status_code=418, content={})
 141
 142     # No known blocks
 143     return JSONResponse(status_code=200, content={})
 144
 145 @router.get(config.get("base_url") + "/scoreboard")
 146 def index(request: Request, blockers: int = None, blocked: int = None, reference: int = None, software: int = None, originator: int = None):
 147     scores = None
 148
 149     if blockers != None and blockers > 0:
 150         response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?blockers={blockers}")
 151     elif blocked != None and blocked > 0:
 152         response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?blocked={blocked}")
 153     elif reference != None and reference > 0:
 154         response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?reference={reference}")
 155     elif software != None and software > 0:
 156         response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?software={software}")
 157     elif originator != None and originator > 0:
 158         response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?originator={originator}")
 159     else:
 160         raise HTTPException(status_code=400, detail="No filter specified")
 161
 162     if response == None:
 163         raise HTTPException(status_code=500, detail="Could not determine scores")
 164     elif not response.ok:
 165         raise HTTPException(status_code=response.status_code, detail=response.text)
 166
 167     return templates.TemplateResponse("scoreboard.html", {
 168         "base_url"  : config.get("base_url"),
 169         "slogan"    : config.get("slogan"),
 170         "request"   : request,
 171         "scoreboard": True,
 172         "blockers"  : blockers,
 173         "blocked"   : blocked,
 174         "reference" : reference,
 175         "software"  : software,
 176         "originator": originator,
 177         "scores"    : fba.json_from_response(response)
 178     })
 179
 180 @router.get(config.get("base_url") + "/")
 181 def index(request: Request, domain: str = None, reason: str = None, reverse: str = None):
 182     if domain == "" or reason == "" or reverse == "":
 183         return fastapi.responses.RedirectResponse("/")
 184
 185     response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/info.json")
 186
 187     if not response.ok:
 188         raise HTTPException(status_code=response.status_code, detail=response.text)
 189
 190     info = response.json()
 191     blocks = None
 192
 193     if domain != None:
 194         blocks = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/index.json?domain={domain}")
 195     elif reason != None:
 196         blocks = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/index.json?reason={reason}")
 197     elif reverse != None:
 198         blocks = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/index.json?reverse={reverse}")
 199
 200     if blocks != None:
 201         if not blocks.ok:
 202             raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
 203         blocks = blocks.json()
 204         for block_level in blocks:
 205             for block in blocks[block_level]:
 206                 block["first_seen"] = datetime.utcfromtimestamp(block["first_seen"]).strftime(config.get("timestamp_format"))
 207                 block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime(config.get("timestamp_format"))
 208
 209     return templates.TemplateResponse("index.html", {
 210         "request": request,
 211         "domain" : domain,
 212         "blocks" : blocks,
 213         "reason" : reason,
 214         "reverse": reverse,
 215         "info"   : info
 216     })
 217
 218 @router.get(config.get("base_url") + "/rss")
 219 def rss(request: Request, domain: str = None):
 220     if domain != None:
 221         wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
 222         punycode = domain.encode('idna').decode('utf-8')
 223         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",
 224                   (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
 225     else:
 226         fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks ORDER BY first_seen DESC LIMIT 50")
 227
 228     result = fba.cursor.fetchall()
 229
 230     blocks = []
 231     for blocker, blocked, block_level, reason, first_seen, last_seen in result:
 232         first_seen = utils.format_datetime(datetime.fromtimestamp(first_seen))
 233         if reason == None or reason == '':
 234             reason = "No reason provided."
 235         else:
 236             reason = "Provided reason: '" + reason + "'"
 237
 238         blocks.append({
 239             "blocker"    : blocker,
 240             "blocked"    : blocked,
 241             "block_level": block_level,
 242             "reason"     : reason,
 243             "first_seen" : first_seen
 244         })
 245
 246     return templates.TemplateResponse("rss.xml", {
 247         "request"  : request,
 248         "timestamp": utils.format_datetime(datetime.now()),
 249         "domain"   : domain,
 250         "hostname" : config.get("hostname"),
 251         "blocks"   : blocks
 252     }, headers={
 253         "Content-Type": "routerlication/rss+xml"
 254     })
 255
 256 @router.get(config.get("base_url") + "/robots.txt", response_class=PlainTextResponse)
 257 def robots(request: Request):
 258     return templates.TemplateResponse("robots.txt", {
 259         "request" : request,
 260         "base_url": config.get("base_url")
 261     })
 262
 263 if __name__ == "__main__":
 264     uvicorn.run("api:router", host=config.get("host"), port=config.get("port"), log_level=config.get("log_level"))