Function login

agent/src/opentrace_agent/cli/auth.py:387–456

Run the full OAuth 2.0 + PKCE browser login flow. 1. Fetch authorization server metadata (RFC 8414). 2. Ensure a client is registered (RFC 7591 dynamic registration or cached). 3. Start a local HTTP server for the callback. 4. Open the browser to the authorization endpoint. 5. W

()

def login() -> dict[str, Any]:
    """Run the full OAuth 2.0 + PKCE browser login flow.

    1. Fetch authorization server metadata (RFC 8414).
    2. Ensure a client is registered (RFC 7591 dynamic registration or cached).
    3. Start a local HTTP server for the callback.
    4. Open the browser to the authorization endpoint.
    5. Wait for the redirect with the authorization code.
    6. Exchange the code for tokens and persist to disk.

    Returns the saved auth payload.
    """
    disco = discover()

    port = _find_open_port()
    redirect_uris = [f"http://{CALLBACK_HOST}:{p}/callback" for p in CALLBACK_PORT_RANGE]
    redirect_uri = f"http://{CALLBACK_HOST}:{port}/callback"

    client = _ensure_client(disco, redirect_uris)
    client_id = client["client_id"]

    verifier, challenge = _generate_pkce()
    state = secrets.token_urlsafe(32)

    params = urlencode(
        {
            "response_type": "code",
            "client_id": client_id,
            "redirect_uri": redirect_uri,
            "scope": SCOPES,
            "state": state,
            "code_challenge": challenge,
            "code_challenge_method": "S256",
            "prompt": "none",  # Skip org selector — CLI controls org via config
        }
    )
    authorize_url = f"{disco['authorization_endpoint']}?{params}"

    result = _OAuthResult()
    handler_cls = _make_handler(result, state)

    server = _ThreadedHTTPServer((CALLBACK_HOST, port), handler_cls)
    server_thread = threading.Thread(target=server.serve_forever, daemon=True)
    server_thread.start()

    try:
        webbrowser.open(authorize_url)

        # Wait for the callback (up to 5 minutes).
        if not result.ready.wait(timeout=300):
            raise TimeoutError("Timed out waiting for browser login (5 min)")

        if result.error:
            raise RuntimeError(result.error)

        code = result.code
        assert code is not None  # noqa: S101