"""Tests for HanzoSession's authentication state. `test_fake_token_is_not_authenticated` is THE regression. Against the shipped implementation — def is_authenticated(self) -> bool: return self.load_token() is not None — it FAILS: a stored token of "fake.not.a.real.jwt" made it return True, and every tool that gated on it believed the caller was signed in. """ from __future__ import annotations import json import time import jwt import pytest from cryptography.hazmat.primitives.asymmetric import rsa from hanzo_iam import store from hanzo_iam import tokens from hanzo_tools.auth.session import HanzoSession ISSUER = "https://hanzo.id" FAKE = "fake.not.a.real.jwt" @pytest.fixture(scope="module") def keypair(): return rsa.generate_private_key(public_exponent=65537, key_size=2048) def _jwks(key, kid="test-key"): data = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(key.public_key())) data.update({"kid": kid, "use": "sig", "alg": "RS256"}) return {"keys": [data]} def _token(key, **overrides): claims = { "iss": ISSUER, "sub": "hanzo/z", "aud": "hanzo-app", "exp": int(time.time()) + 3600, "iat": int(time.time()), } claims.update(overrides) return jwt.encode(claims, key, algorithm="RS256", headers={"kid": "test-key"}) @pytest.fixture(autouse=True) def clean_session(tmp_path, monkeypatch): """Isolate the store, clear env credentials, reset the singleton.""" monkeypatch.setattr(store, "TOKEN_DIR", tmp_path / "auth") monkeypatch.setattr(store, "TOKEN_FILE", tmp_path / "auth" / "token.json") monkeypatch.setattr(store, "_keyring", lambda: None) monkeypatch.delenv("HANZO_AUTH_TOKEN", raising=False) monkeypatch.delenv("HANZO_API_KEY", raising=False) tokens.reset_jwks_cache() HanzoSession.reset() yield HanzoSession.reset() tokens.reset_jwks_cache() @pytest.fixture def serve_jwks(monkeypatch): def _serve(document): monkeypatch.setattr(jwt.PyJWKClient, "fetch_data", lambda self: document) return _serve @pytest.fixture def no_network(monkeypatch): """Any HTTP call is a test bug unless the test opted into one.""" import httpx def forbidden(*a, **kw): raise AssertionError("unexpected network call") monkeypatch.setattr(httpx, "get", forbidden) def _store(token: str, **extra): store.save({"access_token": token, "server_url": ISSUER, **extra}) # --- The defect ----------------------------------------------------------- def test_fake_token_is_not_authenticated(monkeypatch): """A stored token of 'fake.not.a.real.jwt' must not authenticate anybody. It is not a JWT, so it is treated as an opaque credential and put to the issuer, which rejects it. Under the old presence check this returned True. """ import httpx class Rejected: status_code = 401 headers = {"content-type": "application/json"} monkeypatch.setattr(httpx, "get", lambda *a, **kw: Rejected()) _store(FAKE) session = HanzoSession.get() assert session.has_credential() is True # the string IS there assert session.is_authenticated() is False # but it proves nothing assert session.get_token_info()["authenticated"] is False def test_fake_token_in_env_is_not_authenticated(monkeypatch): """Same defect via HANZO_AUTH_TOKEN, which bypasses the store entirely.""" import httpx class Rejected: status_code = 401 headers = {"content-type": "application/json"} monkeypatch.setattr(httpx, "get", lambda *a, **kw: Rejected()) monkeypatch.setenv("HANZO_AUTH_TOKEN", FAKE) assert HanzoSession.get().is_authenticated() is False def test_presence_and_validity_are_different_questions(serve_jwks, keypair, no_network): """The two used to be one method. Keeping them apart is the fix.""" serve_jwks(_jwks(keypair)) _store(_token(keypair, exp=int(time.time()) - 3600)) session = HanzoSession.get() assert session.has_credential() is True assert session.is_authenticated() is False assert session.verify().reason == tokens.EXPIRED # --- Genuine credentials -------------------------------------------------- def test_real_token_authenticates(serve_jwks, keypair, no_network): serve_jwks(_jwks(keypair)) _store(_token(keypair)) session = HanzoSession.get() assert session.is_authenticated() is True info = session.get_token_info() assert info["authenticated"] is True assert info["expired"] is False assert session.verify().claims["sub"] == "hanzo/z" def test_no_credential_at_all(no_network): session = HanzoSession.get() assert session.has_credential() is False assert session.is_authenticated() is False assert session.get_token_info() == { "authenticated": False, "reason": tokens.NO_CREDENTIAL, } # --- Forgeries ------------------------------------------------------------ def test_token_from_another_issuer_is_rejected(serve_jwks, keypair, no_network): """Correctly signed, wrong issuer: still not a Hanzo session.""" serve_jwks(_jwks(keypair)) _store(_token(keypair, iss="https://evil.example")) assert HanzoSession.get().verify().reason == tokens.WRONG_ISSUER def test_self_signed_token_is_rejected(serve_jwks, keypair, no_network): other = rsa.generate_private_key(public_exponent=65537, key_size=2048) serve_jwks(_jwks(keypair)) _store(_token(other)) assert HanzoSession.get().is_authenticated() is False def test_expiry_comes_from_the_token_not_from_local_bookkeeping( serve_jwks, keypair, no_network ): """A caller could otherwise claim a fresh session by editing login_time in the stored JSON. `exp` is signed; login_time is not.""" serve_jwks(_jwks(keypair)) exp = int(time.time()) + 60 store.save( { "access_token": _token(keypair, exp=exp), "server_url": ISSUER, "login_time": int(time.time()) + 10**6, "expires_in": 10**6, } ) assert HanzoSession.get().get_token_info()["expires_at"] == exp # --- Fail closed ---------------------------------------------------------- def test_offline_does_not_grant_access(monkeypatch, keypair): """No JWKS, no verification, no session — an unreachable IdP must not be a free pass.""" def down(self): raise jwt.exceptions.PyJWKClientConnectionError("network down") monkeypatch.setattr(jwt.PyJWKClient, "fetch_data", down) _store(_token(keypair)) session = HanzoSession.get() assert session.is_authenticated() is False assert session.verify().reason == tokens.JWKS_UNREACHABLE def test_jwks_path_carries_the_v1_iam_prefix(serve_jwks, keypair, monkeypatch): """The unprefixed /.well-known/jwks answers 200 text/html on hanzo.id, so verification would silently have nothing to check against.""" seen: list[str] = [] monkeypatch.setattr( jwt.PyJWKClient, "fetch_data", lambda self: (seen.append(self.uri), _jwks(keypair))[1], ) _store(_token(keypair)) HanzoSession.get().verify() assert seen == [f"{ISSUER}/v1/iam/.well-known/jwks"] # --- Storage -------------------------------------------------------------- def test_session_persists_tokens_privately(serve_jwks, keypair, no_network): import stat serve_jwks(_jwks(keypair)) HanzoSession.get()._save_token({"access_token": _token(keypair), "server_url": ISSUER}) assert stat.S_IMODE(store.TOKEN_FILE.stat().st_mode) == 0o600 def test_logout_clears_the_credential(serve_jwks, keypair, no_network): serve_jwks(_jwks(keypair)) _store(_token(keypair)) session = HanzoSession.get() assert session.is_authenticated() is True session.logout() HanzoSession.reset() assert HanzoSession.get().is_authenticated() is False def test_login_refuses_to_store_a_token_it_cannot_verify(monkeypatch, keypair): """If IAM hands back something unverifiable, that is a bug to surface, not a credential to keep and present as proof of identity later.""" from hanzo_iam import oauth monkeypatch.setattr( oauth, "login", lambda **kw: {"access_token": FAKE, "server_url": ISSUER, "client_id": "hanzo-app"}, ) monkeypatch.setattr( jwt.PyJWKClient, "fetch_data", lambda self: _jwks(keypair) ) with pytest.raises(RuntimeError, match="cannot verify"): HanzoSession.get().login() assert store.load() is None