# 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 fastapi import Request, HTTPException, responses, Query
+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
router = fastapi.FastAPI(docs_url=fba.config["base_url"] + "/docs", redoc_url=fba.config["base_url"] + "/redoc")
templates = Jinja2Templates(directory="templates")
-@router.get(fba.config["base_url"] + "/api/info")
+@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()
"slogan" : fba.config["slogan"]
}
-@router.get(fba.config["base_url"] + "/api/top")
+@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:
return scoreboard
-@router.get(fba.config["base_url"] + "/api")
+@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:
raise HTTPException(status_code=400, detail="No filter specified")
return result
+@router.get(fba.config["base_url"] + "/api/mutual.json", response_class=JSONResponse)
+def mutual(domains: list[str] = Query()):
+ """Return 200 if federation is open between the two, 4xx otherwise"""
+ fba.cursor.execute(
+ "SELECT block_level FROM blocks " \
+ "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
+ "AND block_level = 'reject' " \
+ "LIMIT 1",
+ {
+ "a" : domains[0],
+ "b" : domains[1],
+ "aw": "*." + domains[0],
+ "bw": "*." + domains[1],
+ },
+ )
+ res = fba.cursor.fetchone()
+
+ if res 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?blockers={blockers}")
+ 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?blocked={blocked}")
+ 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?reference={reference}")
+ 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?software={software}")
+ res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top.json?software={software}")
else:
raise HTTPException(status_code=400, detail="No filter specified")
@router.get(fba.config["base_url"] + "/")
def index(request: Request, domain: str = None, reason: str = None, reverse: str = None):
if domain == "" or reason == "" or reverse == "":
- return responses.RedirectResponse("/")
+ 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")
+ 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?domain={domain}")
+ 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?reason={reason}")
+ 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?reverse={reverse}")
+ 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:
"info" : info
})
-@router.get(fba.config["base_url"] + "/api/mutual")
-def mutual(domains: list[str] = Query()):
- """Return 200 if federation is open between the two, 4xx otherwise"""
- fba.cursor.execute(
- "SELECT block_level FROM blocks " \
- "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
- "AND block_level = 'reject' " \
- "LIMIT 1",
- {
- "a" : domains[0],
- "b" : domains[1],
- "aw": "*." + domains[0],
- "bw": "*." + domains[1],
- },
- )
- res = fba.cursor.fetchone()
-
- if res is not None:
- # Blocks found
- return responses.JSONResponse(status_code=418, content={})
-
- # No known blocks
- return responses.JSONResponse(status_code=200, content={})
-
@router.get(fba.config["base_url"] + "/rss")
def rss(request: Request, domain: str = None):
if domain != None:
return templates.TemplateResponse("robots.txt", {
"request" : request,
"base_url": fba.config["base_url"]
- }, headers={
- "Content-Type": "text/plain"
})
if __name__ == "__main__":