foxhole/app/main.py

#!/usr/bin/env python3
import logging
import sys
import fastapi
import httpx
import json

from fastapi import FastAPI
from fastapi import Depends
from fastapi import Request
from fastapi import Response
from fastapi.exceptions import HTTPException
from sqlalchemy.util import monkeypatch_proxied_specials
from starlette.responses import JSONResponse
from typing import Any

from loguru import logger

from app import httpsig
from app.database import get_db_session
from app.config import DEBUG
from app.activitypub import ME
from app.config import BASE_URL
from app.config import DOMAIN
from app.config import ID
from app.config import USERNAME
from app.config import VERSION
from app.database import AsyncSession
from app.database import get_db_session
from app.boxes import save_incoming

def _check_0rtt_early_data(request: Request) -> None:
    """Disable TLS1.3 0-RTT requests for non-GET."""
    if request.headers.get("Early-Data", None) == "1" and request.method != "GET":
        raise fastapi.HTTPException(status_code=425, detail="Too early")

app = FastAPI(
    docs_url=None, redoc_url=None, dependencies=[Depends(_check_0rtt_early_data)]
)

logger.remove()
logger.add(sys.stdout, level="DEBUG" if DEBUG else "INFO")

@app.get("/")
async def index():
    return ME

@app.post("/inbox")
async def inbox(
    request: Request,
    db_session: AsyncSession = Depends(get_db_session),
    httpsig_checker = Depends(httpsig.httpsig_checker),
) -> Response:
    payload = await request.json()

    if httpsig_checker:
        await save_incoming(db_session, payload)
        return Response(status_code=202)
    else:
        return Response(status_code=406)

@app.get("/.well-known/webfinger")
async def wellknown_webfinger(resource: str) -> JSONResponse:
    """Exposes/servers WebFinger data."""
    if resource not in [f"acct:{USERNAME}@{DOMAIN}", ID]:
        logger.info(f"Got invalid req for {resource}")
        raise HTTPException(status_code=404)

    out = {
        "subject": f"acct:{USERNAME}@{DOMAIN}",
        "aliases": [ID],
        "links": [
            {
                "rel": "http://webfinger.net/rel/profile-page",
                "type": "text/html",
                "href": ID + "/",
            },
            {"rel": "self", "type": "application/activity+json", "href": ID},
            {
                "rel": "http://ostatus.org/schema/1.0/subscribe",
                "template": BASE_URL + "/admin/lookup?query={uri}",
            },
        ],
    }

    return JSONResponse(
        out,
        media_type="application/jrd+json; charset=utf-8",
        headers={"Access-Control-Allow-Origin": "*"},
    )

@app.get("/.well-known/nodeinfo")
async def well_known_nodeinfo() -> dict[str, Any]:
    return {
        "links": [
            {
                "rel": "http://nodeinfo.diaspora.software/ns/schema/2.0",
                "href": f"{BASE_URL}/nodeinfo/2.0",
            }
        ]
    }


@app.get("/nodeinfo/2.0")
async def nodeinfo(
    db_session: AsyncSession = Depends(get_db_session),
):
    return JSONResponse(
        {
            "version": "2.0",
            "software": {
                "name": "foxhole",
                "version": VERSION,
            },
            "protocols": ["activitypub"],
            "services": {"inbound": [], "outbound": []},
            "usage": {"users": {"total": 1}, "localPosts": 0}, #TODO
            "openRegistrations": False,
            "metadata": {},
        },
    )