aly.codes
Retro Bulletin Board Systems on atproto. Web app and TUI.
atbbs.xyz
python
tui
atproto
bbs
1"""Temporary local HTTP server to catch OAuth callbacks."""
2
3import asyncio
4
5from aiohttp import web
6
7
8async def wait_for_callback(port: int = 23847, timeout: float = 90.0) -> dict:
9 """Start a local server and wait for the OAuth callback.
10
11 Returns dict with 'code', 'state', and 'iss' from the callback query params.
12 """
13 result: dict = {}
14 event = asyncio.Event()
15
16 async def handle_callback(request: web.Request) -> web.Response:
17 result["code"] = request.query.get("code", "")
18 result["state"] = request.query.get("state", "")
19 result["iss"] = request.query.get("iss", "")
20 event.set()
21 return web.Response(
22 text="<html><body><p>Login complete. You can close this tab.</p></body></html>",
23 content_type="text/html",
24 )
25
26 app = web.Application()
27 app.router.add_get("/oauth/callback", handle_callback)
28
29 runner = web.AppRunner(app)
30 await runner.setup()
31 site = web.TCPSite(runner, "127.0.0.1", port)
32 try:
33 await site.start()
34 except OSError as e:
35 await runner.cleanup()
36 raise RuntimeError(
37 f"Could not bind OAuth callback server on 127.0.0.1:{port} "
38 f"({e}). Is another atbbs login in progress?"
39 ) from e
40
41 try:
42 try:
43 await asyncio.wait_for(event.wait(), timeout=timeout)
44 except asyncio.TimeoutError:
45 raise RuntimeError(
46 f"Timed out waiting for OAuth callback after {int(timeout)}s."
47 ) from None
48 finally:
49 await runner.cleanup()
50
51 return result