""" Federation routes for L2 server. Handles WebFinger, nodeinfo, and ActivityPub discovery. """ import logging from fastapi import APIRouter, Request, HTTPException from fastapi.responses import JSONResponse from ..config import settings router = APIRouter() logger = logging.getLogger(__name__) @router.get("/.well-known/webfinger") async def webfinger(resource: str): """WebFinger endpoint for actor discovery.""" import db # Parse resource (acct:username@domain) if not resource.startswith("acct:"): raise HTTPException(400, "Invalid resource format") parts = resource[5:].split("@") if len(parts) != 2: raise HTTPException(400, "Invalid resource format") username, domain = parts if domain != settings.domain: raise HTTPException(404, "User not on this server") user = await db.get_user(username) if not user: raise HTTPException(404, "User not found") return JSONResponse( content={ "subject": resource, "aliases": [f"https://{settings.domain}/users/{username}"], "links": [ { "rel": "self", "type": "application/activity+json", "href": f"https://{settings.domain}/users/{username}", }, { "rel": "http://webfinger.net/rel/profile-page", "type": "text/html", "href": f"https://{settings.domain}/users/{username}", }, ], }, media_type="application/jrd+json", ) @router.get("/.well-known/nodeinfo") async def nodeinfo_index(): """NodeInfo index.""" return JSONResponse( content={ "links": [ { "rel": "http://nodeinfo.diaspora.software/ns/schema/2.0", "href": f"https://{settings.domain}/nodeinfo/2.0", } ] }, media_type="application/json", ) @router.get("/nodeinfo/2.0") async def nodeinfo(): """NodeInfo 2.0 endpoint.""" import db user_count = await db.count_users() activity_count = await db.count_activities() return JSONResponse( content={ "version": "2.0", "software": { "name": "artdag", "version": "1.0.0", }, "protocols": ["activitypub"], "usage": { "users": {"total": user_count, "activeMonth": user_count}, "localPosts": activity_count, }, "openRegistrations": True, "metadata": { "nodeName": "Art-DAG", "nodeDescription": "Content-addressable media processing with ActivityPub federation", }, }, media_type="application/json", ) @router.get("/.well-known/host-meta") async def host_meta(): """Host-meta endpoint.""" xml = f''' ''' from fastapi.responses import Response return Response(content=xml, media_type="application/xrd+xml")