hermes-agent/tests/hermes_cli/test_auth_manual_paste.py

"""Tests for the OAuth manual-paste fallback for browser-only remotes.

Regression coverage for [#26923](https://github.com/NousResearch/hermes-agent/issues/26923):
GCP Cloud Shell, GitHub Codespaces, AWS EC2 Instance Connect and
other browser-only remote consoles can't reach the
``http://127.0.0.1:56121/callback`` loopback listener bound on the
remote VM.  The previous SSH-tunnel hint was useless without a real
SSH client, leaving the user with no path forward.  This test file
locks in four things:

* ``_is_remote_session`` recognises the cloud-shell / Codespaces
  envvars (so the existing hint at least fires).
* ``_parse_pasted_callback`` accepts every form a user might paste
  (full URL, ``?code=...&state=...`` fragment, bare ``code=...``,
  bare opaque value) and returns the same shape the loopback HTTP
  handler does.
* ``_prompt_manual_callback_paste`` reads stdin and produces that
  same shape.
* ``_xai_oauth_loopback_login(manual_paste=True)`` skips the HTTP
  server entirely, validates ``state``, and goes straight to the
  token exchange — proving the paste path actually wires up.
"""

from __future__ import annotations

import builtins
import io
import contextlib

import pytest

from hermes_cli import auth as auth_mod


# ---------------------------------------------------------------------------
# _is_remote_session — broadened detection (#26923)
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "envvar",
    [
        "SSH_CLIENT",
        "SSH_TTY",
        "CLOUD_SHELL",
        "CODESPACES",
        "CODESPACE_NAME",
        "GITPOD_WORKSPACE_ID",
        "REPL_ID",
        "STACKBLITZ",
    ],
)
def test_is_remote_session_detects_known_remote_envvar(monkeypatch, envvar):
    """Each documented remote-console env var must trip the check.

    The SSH ones preserve historical behaviour; the cloud-shell ones
    are what closes #26923.  Without these, the SSH hint never fires
    and the user has no signal that ``--manual-paste`` exists.
    """
    for name in (
        "SSH_CLIENT",
        "SSH_TTY",
        "CLOUD_SHELL",
        "CODESPACES",
        "CODESPACE_NAME",
        "GITPOD_WORKSPACE_ID",
        "REPL_ID",
        "STACKBLITZ",
    ):
        monkeypatch.delenv(name, raising=False)
    monkeypatch.setenv(envvar, "1")
    assert auth_mod._is_remote_session() is True


def test_is_remote_session_false_when_no_remote_envvars(monkeypatch):
    for name in (
        "SSH_CLIENT",
        "SSH_TTY",
        "CLOUD_SHELL",
        "CODESPACES",
        "CODESPACE_NAME",
        "GITPOD_WORKSPACE_ID",
        "REPL_ID",
        "STACKBLITZ",
    ):
        monkeypatch.delenv(name, raising=False)
    assert auth_mod._is_remote_session() is False


# ---------------------------------------------------------------------------
# _parse_pasted_callback — accept every plausible paste form
# ---------------------------------------------------------------------------


def test_parse_full_callback_url():
    out = auth_mod._parse_pasted_callback(
        "http://127.0.0.1:56121/callback?code=abc123&state=deadbeef"
    )
    assert out == {
        "code": "abc123",
        "state": "deadbeef",
        "error": None,
        "error_description": None,
    }


def test_parse_callback_url_https_and_extra_params():
    out = auth_mod._parse_pasted_callback(
        "https://127.0.0.1:56121/callback?code=abc&state=xyz&scope=openid"
    )
    assert out["code"] == "abc"
    assert out["state"] == "xyz"


def test_parse_bare_query_string_with_leading_question_mark():
    out = auth_mod._parse_pasted_callback("?code=p1&state=s1")
    assert out["code"] == "p1"
    assert out["state"] == "s1"


def test_parse_bare_query_fragment_no_question_mark():
    out = auth_mod._parse_pasted_callback("code=p2&state=s2")
    assert out["code"] == "p2"
    assert out["state"] == "s2"


def test_parse_bare_opaque_code_value():
    """Some users only copy the ``code`` value itself."""
    out = auth_mod._parse_pasted_callback("ABCDEF-the-code-value")
    assert out["code"] == "ABCDEF-the-code-value"
    assert out["state"] is None


def test_parse_callback_with_error_field():
    out = auth_mod._parse_pasted_callback(
        "http://127.0.0.1:56121/callback?error=access_denied"
        "&error_description=user+rejected"
    )
    assert out["code"] is None
    assert out["error"] == "access_denied"
    assert out["error_description"] == "user rejected"


def test_parse_empty_input_returns_all_none():
    out = auth_mod._parse_pasted_callback("")
    assert out == {
        "code": None,
        "state": None,
        "error": None,
        "error_description": None,
    }


def test_parse_whitespace_only_returns_all_none():
    out = auth_mod._parse_pasted_callback("   \n\t  ")
    assert out["code"] is None


def test_parse_malformed_url_does_not_crash():
    out = auth_mod._parse_pasted_callback("http://[not a url")
    # Malformed URLs return all-None rather than raising — the caller
    # (state check) will reject the empty payload with a clear error.
    assert out["code"] is None


# ---------------------------------------------------------------------------
# _prompt_manual_callback_paste — stdin handling
# ---------------------------------------------------------------------------


def test_prompt_reads_stdin_and_parses(monkeypatch):
    monkeypatch.setattr(
        builtins, "input",
        lambda *_a, **_k: "http://127.0.0.1:56121/callback?code=abc&state=xyz",
    )
    buf = io.StringIO()
    with contextlib.redirect_stdout(buf):
        out = auth_mod._prompt_manual_callback_paste(
            "http://127.0.0.1:56121/callback"
        )
    rendered = buf.getvalue()
    assert "Manual callback paste" in rendered
    assert "127.0.0.1:56121" in rendered
    assert out["code"] == "abc"
    assert out["state"] == "xyz"


def test_prompt_eof_returns_all_none(monkeypatch):
    def _raise_eof(*_a, **_k):
        raise EOFError()

    monkeypatch.setattr(builtins, "input", _raise_eof)
    with contextlib.redirect_stdout(io.StringIO()):
        out = auth_mod._prompt_manual_callback_paste(
            "http://127.0.0.1:56121/callback"
        )
    assert out["code"] is None


def test_prompt_keyboard_interrupt_returns_all_none(monkeypatch):
    def _raise_kbi(*_a, **_k):
        raise KeyboardInterrupt()

    monkeypatch.setattr(builtins, "input", _raise_kbi)
    with contextlib.redirect_stdout(io.StringIO()):
        out = auth_mod._prompt_manual_callback_paste(
            "http://127.0.0.1:56121/callback"
        )
    assert out["code"] is None


# ---------------------------------------------------------------------------
# _xai_oauth_loopback_login(manual_paste=True) — full integration
# ---------------------------------------------------------------------------


class _StubTokenResponse:
    status_code = 200

    def __init__(self, payload):
        self._payload = payload
        self.text = ""

    def json(self):
        return self._payload


def test_xai_loopback_login_manual_paste_skips_http_server(monkeypatch):
    """``manual_paste=True`` must NOT bind a loopback HTTP server.

    Direct end-to-end regression for #26923: the whole point is that
    the listener is unreachable on browser-only remotes, so the paste
    path must avoid it entirely.  We assert this by replacing
    ``_xai_start_callback_server`` with a function that fails if
    invoked, then driving the full happy path with a stubbed prompt
    + stubbed token endpoint.
    """
    monkeypatch.setattr(
        auth_mod, "_xai_oauth_discovery",
        lambda *_a, **_k: {
            "authorization_endpoint": "https://auth.x.ai/oauth2/authorize",
            "token_endpoint": "https://auth.x.ai/oauth2/token",
        },
    )

    def _server_must_not_be_called(*_a, **_k):
        raise AssertionError(
            "manual_paste=True must skip the loopback HTTP server "
            "(regression for #26923)"
        )

    monkeypatch.setattr(
        auth_mod, "_xai_start_callback_server", _server_must_not_be_called
    )

    captured_state: dict = {}

    def _fake_prompt(_redirect_uri):
        # Hermes generates state internally; we won't know it ahead of
        # time, so capture the state Hermes baked into the authorize
        # URL via a sneak peek on ``_xai_oauth_build_authorize_url``.
        return {
            "code": "fake-auth-code",
            "state": captured_state["value"],
            "error": None,
            "error_description": None,
        }

    monkeypatch.setattr(
        auth_mod, "_prompt_manual_callback_paste", _fake_prompt
    )

    original_build = auth_mod._xai_oauth_build_authorize_url

    def _capture_state(**kwargs):
        captured_state["value"] = kwargs["state"]
        return original_build(**kwargs)

    monkeypatch.setattr(
        auth_mod, "_xai_oauth_build_authorize_url", _capture_state
    )

    def _fake_token_post(*_a, **_k):
        return _StubTokenResponse(
            {
                "access_token": "at",
                "refresh_token": "rt",
                "id_token": "",
                "expires_in": 3600,
                "token_type": "Bearer",
            }
        )

    monkeypatch.setattr(auth_mod.httpx, "post", _fake_token_post)

    with contextlib.redirect_stdout(io.StringIO()):
        creds = auth_mod._xai_oauth_loopback_login(manual_paste=True)

    assert creds["tokens"]["access_token"] == "at"
    assert creds["tokens"]["refresh_token"] == "rt"
    assert "127.0.0.1:56121" in creds["redirect_uri"]


def test_xai_loopback_login_manual_paste_state_mismatch_raises(monkeypatch):
    """A pasted callback with the wrong state must still be rejected.

    The HTTP-server path uses the same state check; manual-paste
    must not be a CSRF bypass.
    """
    monkeypatch.setattr(
        auth_mod, "_xai_oauth_discovery",
        lambda *_a, **_k: {
            "authorization_endpoint": "https://auth.x.ai/oauth2/authorize",
            "token_endpoint": "https://auth.x.ai/oauth2/token",
        },
    )
    monkeypatch.setattr(
        auth_mod, "_prompt_manual_callback_paste",
        lambda _ru: {
            "code": "fake",
            "state": "WRONG-STATE",
            "error": None,
            "error_description": None,
        },
    )

    with contextlib.redirect_stdout(io.StringIO()):
        with pytest.raises(auth_mod.AuthError) as exc:
            auth_mod._xai_oauth_loopback_login(manual_paste=True)
    assert exc.value.code == "xai_state_mismatch"


def test_xai_loopback_login_manual_paste_missing_code_raises(monkeypatch):
    """Empty paste must surface as ``xai_code_missing``, not crash."""
    monkeypatch.setattr(
        auth_mod, "_xai_oauth_discovery",
        lambda *_a, **_k: {
            "authorization_endpoint": "https://auth.x.ai/oauth2/authorize",
            "token_endpoint": "https://auth.x.ai/oauth2/token",
        },
    )
    captured: dict = {"state": None}
    original_build = auth_mod._xai_oauth_build_authorize_url

    def _capture(**kw):
        captured["state"] = kw["state"]
        return original_build(**kw)

    monkeypatch.setattr(auth_mod, "_xai_oauth_build_authorize_url", _capture)
    monkeypatch.setattr(
        auth_mod, "_prompt_manual_callback_paste",
        lambda _ru: {
            "code": None,
            "state": captured["state"],
            "error": None,
            "error_description": None,
        },
    )

    with contextlib.redirect_stdout(io.StringIO()):
        with pytest.raises(auth_mod.AuthError) as exc:
            auth_mod._xai_oauth_loopback_login(manual_paste=True)
    assert exc.value.code == "xai_code_missing"


# ---------------------------------------------------------------------------
# _print_loopback_ssh_hint — now also mentions --manual-paste
# ---------------------------------------------------------------------------


def test_ssh_hint_mentions_manual_paste_for_non_ssh_remotes(monkeypatch):
    """Users on Cloud Shell / Codespaces have no real SSH client; the
    hint must point them at the new ``--manual-paste`` flag instead
    of leaving them stuck on the ``ssh -L`` recipe."""
    monkeypatch.setattr(auth_mod, "_is_remote_session", lambda: True)
    buf = io.StringIO()
    with contextlib.redirect_stdout(buf):
        auth_mod._print_loopback_ssh_hint(
            "http://127.0.0.1:56121/callback",
            docs_url=auth_mod.XAI_OAUTH_DOCS_URL,
        )
    rendered = buf.getvalue()
    assert "--manual-paste" in rendered
    assert "Cloud Shell" in rendered or "Codespaces" in rendered