Source code for satorbis_kit.auth.callback_server
"""
auth.callback_server
-----------------------------
Starts a single-use local HTTP server that listens for the OAuth2
authorization-code redirect and returns the full callback URL.
Designed to run in a background daemon thread so it does not block
the Jupyter kernel.
"""
import queue
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
# Shared queue between the HTTP handler and the main thread.
# maxsize=1 so a second accidental request is dropped.
_callback_queue: queue.Queue = queue.Queue(maxsize=1)
_SUCCESS_HTML = """\
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>Login successful</title>
<style>
body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif;
display: flex; align-items: center; justify-content: center;
min-height: 100vh; margin: 0; background: #f0fdf4; }}
.card {{ background: white; border-radius: 12px; padding: 48px 56px;
box-shadow: 0 4px 24px rgba(0,0,0,.08); text-align: center; }}
h2 {{ color: #16a34a; margin: 0 0 12px; font-size: 1.6rem; }}
p {{ color: #64748b; margin: 0; }}
</style>
</head>
<body>
<div class="card">
<h2>✅ Login successful!</h2>
<p>You can close this tab and return to the notebook.</p>
</div>
</body>
</html>
"""
def _make_handler(redirect_port: int) -> type:
"""
Factory that bakes ``redirect_port`` into the handler class so the
handler does not need to reference any outer scope.
"""
class _CallbackHandler(BaseHTTPRequestHandler):
"""Captures the first GET request (the OAuth redirect) and queues it."""
_port = redirect_port
def do_GET(self) -> None:
full_url = f"http://localhost:{self._port}{self.path}"
try:
_callback_queue.put_nowait(full_url)
except queue.Full:
pass # Second request — ignore
body = _SUCCESS_HTML.encode()
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *_args) -> None:
pass # Suppress access-log noise in the notebook
return _CallbackHandler
[docs]
def start_callback_server(
port: int,
timeout_seconds: int = 300,
) -> None:
"""
Start the callback HTTP server. Intended to be called inside a
``threading.Thread`` — it blocks until one request has been handled
or ``timeout_seconds`` has elapsed.
Parameters
----------
port:
Port to listen on (must match the redirect URI registered with the provider).
timeout_seconds:
How long (in seconds) to wait for the callback before giving up.
"""
handler_class = _make_handler(port)
server = HTTPServer(("localhost", port), handler_class)
server.timeout = timeout_seconds
server.handle_request() # Block until exactly one request arrives
server.server_close()
[docs]
def launch_callback_listener(
port: int,
timeout_seconds: int = 300,
) -> None:
"""
Spawn a daemon thread that runs :func:`start_callback_server`.
The thread is a daemon so it is automatically killed when the
Jupyter kernel shuts down, with no cleanup needed.
"""
# Clear any stale entry from a previous run
while not _callback_queue.empty():
try:
_callback_queue.get_nowait()
except queue.Empty:
break
thread = threading.Thread(
target=start_callback_server,
kwargs={"port": port, "timeout_seconds": timeout_seconds},
daemon=True,
name="oidc-callback-server",
)
thread.start()
[docs]
def get_callback_url(timeout_seconds: int = 300) -> str:
"""
Block until the local server receives the OAuth redirect, then
return the full callback URL (including ``?code=...&state=...``).
Parameters
----------
timeout_seconds:
How long to wait. Raises ``TimeoutError`` if no callback arrives.
Returns
-------
str
Full callback URL, e.g.
``http://localhost:4200/callback?code=abc&state=xyz``
"""
try:
return _callback_queue.get(timeout=timeout_seconds)
except queue.Empty as exc:
raise TimeoutError(
f"No OAuth callback received within {timeout_seconds} seconds.\n"
"Did you click the login URL and complete authentication in the browser?"
) from exc