Add api endpoint to test if there are blocks between two domains. Added an index to cover the query.

This commit is contained in:
pwm 2023-02-26 15:23:35 -06:00 committed by Mint
parent 73a8774f5a
commit e4caf63d48
2 changed files with 25 additions and 1 deletions

26
api.py
View file

@ -1,5 +1,5 @@
import uvicorn import uvicorn
from fastapi import FastAPI, Request, HTTPException, responses from fastapi import FastAPI, Request, HTTPException, responses, Query
import sqlite3 import sqlite3
from hashlib import sha256 from hashlib import sha256
from fastapi.templating import Jinja2Templates from fastapi.templating import Jinja2Templates
@ -129,5 +129,29 @@ def index(request: Request, domain: str = None, reason: str = None, reverse: str
return templates.TemplateResponse("index.html", {"request": request, "domain": domain, "blocks": blocks, "reason": reason, "reverse": reverse, "info": info}) return templates.TemplateResponse("index.html", {"request": request, "domain": domain, "blocks": blocks, "reason": reason, "reverse": reverse, "info": info})
@app.get(base_url+"/api/mutual")
def mutual(domains: list[str] = Query()):
"""Return 200 if federation is open between the two, 4xx otherwise"""
conn = sqlite3.connect('blocks.db')
c = conn.cursor()
c.execute(
"SELECT block_level FROM blocks " \
"WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a)) " \
"AND block_level = 'reject' " \
"LIMIT 1",
{
"a": domains[0],
"b": domains[1],
},
)
res = c.fetchone()
c.close()
if res is not None:
# Blocks found
return responses.JSONResponse(status_code=418, content={})
# No known blocks
return responses.JSONResponse(status_code=200, content={})
if __name__ == "__main__": if __name__ == "__main__":
uvicorn.run("api:app", host="127.0.0.1", port=port, log_level="info") uvicorn.run("api:app", host="127.0.0.1", port=port, log_level="info")

Binary file not shown.