"""EVE SSO auth: PKCE, JWT claims, token store, exchange/refresh, get_auth.""" import asyncio import base64 import hashlib import httpx import pytest from conftest import make_jwt from familiar import auth @pytest.fixture def tmp_store(tmp_path, monkeypatch): """Point the token/key files at a temp dir.""" monkeypatch.setenv("FAMILIAR_DATA_DIR", str(tmp_path)) return tmp_path def test_generate_pkce_challenge_matches_verifier(): verifier, challenge = auth.generate_pkce() expected = ( base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode() ) assert challenge == expected assert "=" not in challenge def test_build_authorize_url_has_required_params(): url = auth.build_authorize_url("the-state", "the-challenge") assert "response_type=code" in url assert "code_challenge=the-challenge" in url assert "code_challenge_method=S256" in url assert "state=the-state" in url assert "client_id=" in url def test_decode_jwt_and_character_id(): token = make_jwt({"sub": "CHARACTER:EVE:2112000001", "name": "Latte", "scp": ["a"]}) claims = auth.decode_jwt_claims(token) assert claims["name"] == "Latte" assert auth.character_id_from_claims(claims) == 2112000001 def test_character_id_from_bad_sub_raises(): with pytest.raises(auth.AuthError): auth.character_id_from_claims({"sub": "nonsense"}) def test_decode_malformed_token_raises(): with pytest.raises(auth.AuthError): auth.decode_jwt_claims("not-a-jwt") def test_token_store_round_trip(tmp_store): bundle = {"refresh_token": "r", "character_id": 7, "character_name": "X"} assert auth.load_tokens() is None auth.save_tokens(bundle) assert auth.load_tokens() == bundle assert (tmp_store / "token.enc").exists() assert auth.clear_tokens() is True assert auth.load_tokens() is None assert auth.clear_tokens() is False def test_stored_token_is_encrypted_not_plaintext(tmp_store): auth.save_tokens({"refresh_token": "super-secret-value", "character_id": 1}) raw = (tmp_store / "token.enc").read_bytes() assert b"super-secret-value" not in raw def _token_transport(payload): return httpx.MockTransport(lambda req: httpx.Response(200, json=payload)) def test_exchange_code_posts_and_parses(): payload = {"access_token": "a", "refresh_token": "r", "expires_in": 1200} out = asyncio.run(auth.exchange_code("code", "verifier", transport=_token_transport(payload))) assert out["refresh_token"] == "r" def test_token_request_raises_on_error(): transport = httpx.MockTransport(lambda req: httpx.Response(400, text="bad_request")) with pytest.raises(auth.AuthError): asyncio.run(auth.refresh_tokens("r", transport=transport)) def test_get_auth_without_login_raises(tmp_store): with pytest.raises(auth.NotAuthenticated): asyncio.run(auth.get_auth()) def test_get_auth_uses_unexpired_token(tmp_store): auth.save_tokens( { "access_token": "still-good", "refresh_token": "r", "character_id": 9, "expires_at": 10_000, } ) token, cid = asyncio.run(auth.get_auth(now=5_000)) assert token == "still-good" assert cid == 9 def test_get_auth_refreshes_when_expired(tmp_store): auth.save_tokens( {"access_token": "old", "refresh_token": "r0", "character_id": 9, "expires_at": 1_000} ) new_access = make_jwt({"sub": "CHARACTER:EVE:9", "name": "Latte"}) payload = {"access_token": new_access, "refresh_token": "r1", "expires_in": 1200} token, cid = asyncio.run(auth.get_auth(now=2_000, transport=_token_transport(payload))) assert token == new_access assert cid == 9 # The rotated refresh token is persisted. assert auth.load_tokens()["refresh_token"] == "r1" def test_status_reflects_store(tmp_store): assert auth.status() == {"authenticated": False} auth.save_tokens({"refresh_token": "r", "character_id": 3, "character_name": "Y", "scopes": []}) info = auth.status() assert info["authenticated"] is True assert info["character_id"] == 3