api.py

   1 import uvicorn
   2 from fastapi import FastAPI, Request, HTTPException, responses
   3 import sqlite3
   4 from hashlib import sha256
   5 from fastapi.templating import Jinja2Templates
   6 from requests import get
   7 from json import loads
   8
   9 with open("config.json") as f:
  10     config = loads(f.read())
  11     base_url = config["base_url"]
  12     port = config["port"]
  13 app = FastAPI(docs_url=base_url+"/docs", redoc_url=base_url+"/redoc")
  14 templates = Jinja2Templates(directory=".")
  15
  16 def get_hash(domain: str) -> str:
  17     return sha256(domain.encode("utf-8")).hexdigest()
  18
  19 @app.get(base_url+"/info")
  20 def info():
  21     conn = sqlite3.connect("blocks.db")
  22     c = conn.cursor()
  23     c.execute("select (select count(domain) from instances), (select count(domain) from instances where software in ('pleroma', 'mastodon')), (select count(blocker) from blocks)")
  24     known, indexed, blocks = c.fetchone()
  25     c.close()
  26     return {
  27         "known_instances": known,
  28         "indexed_instances": indexed,
  29         "blocks_recorded": blocks,
  30         "source_code": "https://gitlab.com/EnjuAihara/fedi-block-api",
  31     }
  32
  33 @app.get(base_url+"/api")
  34 def blocked(domain: str = None):
  35     if domain == None:
  36         raise HTTPException(status_code=400, detail="No domain specified")
  37     conn = sqlite3.connect("blocks.db")
  38     c = conn.cursor()
  39     wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
  40     c.execute("select blocker, block_level, reason from blocks where blocked = ? or blocked = ? or blocked = ?", (domain, wildchar, get_hash(domain)))
  41     blocks = c.fetchall()
  42     conn.close()
  43
  44     result = {}
  45     reasons = {}
  46
  47     for domain, block_level, reason in blocks:
  48         if block_level in result:
  49             result[block_level].append(domain)
  50         else:
  51             result[block_level] = [domain]
  52
  53         if reason != "":
  54             if block_level in reasons:
  55                 reasons[block_level][domain] = reason
  56             else:
  57                 reasons[block_level] = {domain: reason}
  58
  59     return {"blocks": result, "reasons": reasons}
  60
  61 @app.get(base_url+"/")
  62 def index(request: Request, domain: str = None):
  63     if domain == "":
  64         return responses.RedirectResponse("/")
  65     blocks = get(f"http://127.0.0.1:{port}{base_url}/api?domain={domain}")
  66     info = None
  67     if domain == None:
  68         info = get(f"http://127.0.0.1:{port}{base_url}/info")
  69         if not info.ok:
  70             raise HTTPException(status_code=info.status_code, detail=info.text)
  71         info = info.json()
  72     if not blocks.ok:
  73         raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
  74     return templates.TemplateResponse("index.html", {"request": request, "domain": domain, "blocks": blocks.json(), "info": info})
  75
  76 if __name__ == "__main__":
  77     uvicorn.run("api:app", host="127.0.0.1", port=port, log_level="info")