diff --git a/examples/key_server.py b/examples/key_server.py new file mode 100644 index 0000000000000000000000000000000000000000..630529c3d45040d86ee070f9d72bdb8d8e028674 --- /dev/null +++ b/examples/key_server.py @@ -0,0 +1,168 @@ +from fastapi import FastAPI, HTTPException, Query +from fastapi.responses import JSONResponse +import uvicorn +from pathlib import Path +from config import CONFIG +from datetime import datetime +from urllib.parse import unquote + +app = FastAPI() + +@app.get("/keys/{key_id}") +async def get_public_key(key_id: str): + """Serve public key for Mastodon to verify signatures""" + print(f"\n=== Received request for key: {key_id} ===") + + try: + # Construct the key path + key_files = list(Path(CONFIG['keys_path']).glob("*public.pem")) + print(f"Available files in {CONFIG['keys_path']}:") + print(key_files) + + # Find the key file that matches the key_id + key_path = None + for file in key_files: + if key_id in str(file): + key_path = file + break + + if not key_path: + print(f"ERROR: No key file found containing ID {key_id}") + raise HTTPException(status_code=404, detail="Key not found") + + print(f"Found key file at: {key_path}") + + # Read the public key + with open(key_path, 'r') as f: + public_key = f.read() + + response = { + "@context": [ + "https://www.w3.org/ns/activitystreams", + "https://w3id.org/security/v1" + ], + "id": f"https://{CONFIG['domain']}/keys/{key_id}", + "owner": f"https://{CONFIG['domain']}/users/{CONFIG['user']}", + "publicKeyPem": public_key + } + + print(f"Sending response: {response}") + return JSONResponse( + content=response, + headers={ + "Content-Type": "application/activity+json" + } + ) + except Exception as e: + print(f"ERROR serving key: {str(e)}") + raise + +@app.get("/users/{username}") +async def get_user(username: str): + """Serve actor profile""" + print(f"\n=== Received request for user: {username} ===") + + if username != CONFIG['user']: + raise HTTPException(status_code=404, detail="User not found") + + # Get the active key ID from the key file + key_files = list(Path(CONFIG['keys_path']).glob("*public.pem")) + if not key_files: + raise HTTPException(status_code=500, detail="No public key found") + + # Extract key ID from filename + key_id = key_files[0].stem.split('_')[-2] + + actor = { + "@context": [ + "https://www.w3.org/ns/activitystreams", + "https://w3id.org/security/v1" + ], + "id": f"https://{CONFIG['domain']}/users/{username}", + "type": "Person", + "preferredUsername": username, + "inbox": f"https://{CONFIG['domain']}/users/{username}/inbox", + "outbox": f"https://{CONFIG['domain']}/users/{username}/outbox", + "followers": f"https://{CONFIG['domain']}/users/{username}/followers", + "following": f"https://{CONFIG['domain']}/users/{username}/following", + "publicKey": { + "id": f"https://{CONFIG['domain']}/keys/{key_id}", + "owner": f"https://{CONFIG['domain']}/users/{username}", + "publicKeyPem": open(key_files[0], 'r').read() + }, + "endpoints": { + "sharedInbox": f"https://{CONFIG['domain']}/inbox" + } + } + + return JSONResponse( + content=actor, + headers={ + "Content-Type": "application/activity+json" + } + ) + +@app.get("/.well-known/webfinger") +async def webfinger(resource: str = Query(...)): + """Handle WebFinger requests""" + print(f"\n=== Received WebFinger request for: {resource} ===") + + # Parse the resource + resource = unquote(resource) + if not resource.startswith("acct:"): + raise HTTPException(status_code=400, detail="Invalid resource format") + + # Extract username and domain + try: + _, identifier = resource.split("acct:") + username, domain = identifier.split("@") + except ValueError: + raise HTTPException(status_code=400, detail="Invalid resource format") + + # Verify domain and username + if domain != CONFIG['domain'] or username != CONFIG['user']: + raise HTTPException(status_code=404, detail="User not found") + + response = { + "subject": f"acct:{username}@{domain}", + "aliases": [ + f"https://{domain}/users/{username}", + ], + "links": [ + { + "rel": "self", + "type": "application/activity+json", + "href": f"https://{domain}/users/{username}" + }, + { + "rel": "http://webfinger.net/rel/profile-page", + "type": "text/html", + "href": f"https://{domain}/users/{username}" + } + ] + } + + return JSONResponse( + content=response, + headers={ + "Content-Type": "application/jrd+json" + } + ) + +@app.get("/") +async def root(): + """Test endpoint""" + return { + "status": "running", + "domain": CONFIG['domain'], + "available_keys": list(str(p) for p in Path(CONFIG['keys_path']).glob("*public.pem")) + } + +if __name__ == "__main__": + print("\n=== Starting Key Server ===") + print(f"Domain: {CONFIG['domain']}") + print(f"Keys path: {CONFIG['keys_path']}") + print("Available keys:") + print(list(Path(CONFIG['keys_path']).glob("*public.pem"))) + + uvicorn.run(app, host="0.0.0.0", port=8880) \ No newline at end of file