Multi-character support across accounts
CI / lint (pull_request) Successful in 49s
CI / test (pull_request) Successful in 56s
CI / lint (push) Successful in 50s
CI / publish (pull_request) Has been skipped
CI / test (push) Successful in 55s
CI / publish (push) Successful in 50s

Rework the token store to hold many characters keyed by character_id,
migrating the previous single-character file transparently. `familiar login`
now adds a character; `familiar status` lists all; `familiar logout [name|id]`
removes one or all.

Every account tool gains an optional `character` (name or id): omitted means
all logged-in characters, given means just that one. Results are uniform
{"characters": [...]}. get_auth refreshes per character.

58 tests, coverage 89%. Closes #9

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
LatteandClaude Opus 4.8 committed 2026-06-27 18:47:34 +02:00
1 parent eb6440db40
commit 61837d4783
5 files changed
+379 -210

No files matched your search

+21 -9
View File
@@ -4,9 +4,10 @@ With no arguments it runs the MCP server over stdio (how Claude Desktop /
Claude Code / Cowork launch it). The auth subcommands are meant to be run by Claude Code / Cowork launch it). The auth subcommands are meant to be run by
hand in a terminal: hand in a terminal:
familiar login # browser-based EVE SSO login (run once) familiar login # add a character via EVE SSO (run once per character)
familiar status # show whether you're logged in familiar status # list logged-in characters
familiar logout # forget stored credentials familiar logout # forget ALL characters
familiar logout NAME # forget one character (by name or id)
""" """
from __future__ import annotations from __future__ import annotations
@@ -18,21 +19,32 @@ from . import auth
def _login() -> int: def _login() -> int:
bundle = auth.interactive_login() bundle = auth.interactive_login()
print(f"Logged in as {bundle.get('character_name')} (id {bundle.get('character_id')}).") print(f"Added {bundle.get('character_name')} (id {bundle.get('character_id')}).")
chars = auth.list_characters()
print(f"Logged-in characters ({len(chars)}): {', '.join(c['character_name'] for c in chars)}")
return 0 return 0
def _status() -> int: def _status() -> int:
info = auth.status() chars = auth.list_characters()
if info["authenticated"]: if not chars:
print(f"Logged in as {info.get('character_name')} (id {info.get('character_id')}).")
else:
print("Not logged in. Run `familiar login`.") print("Not logged in. Run `familiar login`.")
return 0
print(f"Logged-in characters ({len(chars)}):")
for c in chars:
print(f" - {c['character_name']} (id {c['character_id']})")
return 0 return 0
def _logout() -> int: def _logout() -> int:
print("Logged out." if auth.clear_tokens() else "Was not logged in.") target = sys.argv[2] if len(sys.argv) > 2 else None
removed = auth.remove_character(target)
if not removed:
print("Nothing to remove." if target is None else f"No character matching {target!r}.")
elif target is None:
print("Logged out all characters.")
else:
print(f"Removed {target}.")
return 0 return 0
+116 -82
View File
@@ -1,12 +1,18 @@
"""Authenticated, read-only tools for the logged-in character. """Authenticated, read-only tools for the logged-in character(s).
Each tool resolves a valid access token via :mod:`familiar.auth` (refreshing Every tool takes an optional ``character`` (name or id):
transparently) and then calls the character ESI endpoints. If the user has not
logged in, the tools surface a clear "run `familiar login`" error. * omitted -> data for *all* logged-in characters
* given -> just that one character
Results are always returned as a uniform ``{"characters": [ ... ]}`` list, one
entry per character, so the AI can handle one or many the same way. Tokens are
resolved (and refreshed) per character via :mod:`familiar.auth`.
""" """
from __future__ import annotations from __future__ import annotations
from collections.abc import Awaitable, Callable
from typing import Any from typing import Any
from fastmcp import FastMCP from fastmcp import FastMCP
@@ -14,101 +20,129 @@ from fastmcp import FastMCP
from . import auth from . import auth
from .esi import client from .esi import client
# A per-character fetcher: given (access_token, character_id) -> payload dict.
Fetcher = Callable[[str, int], Awaitable[dict[str, Any]]]
async def _auth() -> tuple[str, int]:
return await auth.get_auth() async def _run(character: str | int | None, fetch: Fetcher) -> dict[str, Any]:
chars = auth.list_characters()
if not chars:
raise auth.NotAuthenticated("Not logged in. Run `familiar login` in a terminal first.")
names = {c["character_id"]: c["character_name"] for c in chars}
if character is None:
targets = [c["character_id"] for c in chars]
else:
targets = [auth.resolve_character(character)]
results = []
for cid in targets:
token, _ = await auth.get_auth(cid)
payload = await fetch(token, cid)
results.append({"character_id": cid, "character_name": names.get(cid), **payload})
return {"characters": results}
async def account_status() -> dict[str, Any]: async def account_status() -> dict[str, Any]:
"""Whether familiar is logged in, and as which character (no network call).""" """List which characters familiar is logged in as (no network call)."""
return auth.status() return auth.status()
async def get_wallet() -> dict[str, Any]: async def get_wallet(character: str | int | None = None) -> dict[str, Any]:
"""The character's wallet balance in ISK.""" """Wallet balance (ISK) for one character or all of them."""
token, cid = await _auth()
balance = await client().get(f"/characters/{cid}/wallet/", token=token) async def fetch(token: str, cid: int) -> dict[str, Any]:
return {"character_id": cid, "balance": balance} return {"balance": await client().get(f"/characters/{cid}/wallet/", token=token)}
return await _run(character, fetch)
async def get_skills() -> dict[str, Any]: async def get_skills(character: str | int | None = None) -> dict[str, Any]:
"""Total skill points, unallocated SP, and how many skills are trained.""" """Total SP, unallocated SP, and skill count for one character or all."""
token, cid = await _auth()
data = await client().get(f"/characters/{cid}/skills/", token=token)
return {
"character_id": cid,
"total_sp": data.get("total_sp"),
"unallocated_sp": data.get("unallocated_sp"),
"skills_trained": len(data.get("skills") or []),
}
async def fetch(token: str, cid: int) -> dict[str, Any]:
async def get_skill_queue(limit: int = 10) -> dict[str, Any]: data = await client().get(f"/characters/{cid}/skills/", token=token)
"""The next entries in the skill training queue.""" return {
token, cid = await _auth() "total_sp": data.get("total_sp"),
queue = await client().get(f"/characters/{cid}/skillqueue/", token=token) or [] "unallocated_sp": data.get("unallocated_sp"),
upcoming = [ "skills_trained": len(data.get("skills") or []),
{
"skill_id": q.get("skill_id"),
"finished_level": q.get("finished_level"),
"position": q.get("queue_position"),
"finish_date": q.get("finish_date"),
} }
for q in queue[: max(1, limit)]
] return await _run(character, fetch)
return {"character_id": cid, "in_queue": len(queue), "upcoming": upcoming}
async def get_location() -> dict[str, Any]: async def get_skill_queue(character: str | int | None = None, limit: int = 10) -> dict[str, Any]:
"""Current system/station, active ship, and online status.""" """Upcoming skill-queue entries for one character or all."""
token, cid = await _auth()
loc = await client().get(f"/characters/{cid}/location/", token=token) or {} async def fetch(token: str, cid: int) -> dict[str, Any]:
ship = await client().get(f"/characters/{cid}/ship/", token=token) or {} queue = await client().get(f"/characters/{cid}/skillqueue/", token=token) or []
online = await client().get(f"/characters/{cid}/online/", token=token) or {} upcoming = [
return { {
"character_id": cid, "skill_id": q.get("skill_id"),
"solar_system_id": loc.get("solar_system_id"), "finished_level": q.get("finished_level"),
"station_id": loc.get("station_id"), "position": q.get("queue_position"),
"structure_id": loc.get("structure_id"), "finish_date": q.get("finish_date"),
"ship_type_id": ship.get("ship_type_id"), }
"ship_name": ship.get("ship_name"), for q in queue[: max(1, limit)]
"online": online.get("online"), ]
"last_login": online.get("last_login"), return {"in_queue": len(queue), "upcoming": upcoming}
}
return await _run(character, fetch)
async def get_clones() -> dict[str, Any]: async def get_location(character: str | int | None = None) -> dict[str, Any]:
"""Home station, jump clones, and currently plugged-in implants.""" """Current system/station, active ship, and online status for one or all."""
token, cid = await _auth()
clones = await client().get(f"/characters/{cid}/clones/", token=token) or {}
implants = await client().get(f"/characters/{cid}/implants/", token=token) or []
return {
"character_id": cid,
"home_location": clones.get("home_location"),
"jump_clones": len(clones.get("jump_clones") or []),
"active_implants": implants,
}
async def fetch(token: str, cid: int) -> dict[str, Any]:
async def get_assets(limit: int = 20) -> dict[str, Any]: loc = await client().get(f"/characters/{cid}/location/", token=token) or {}
"""Asset summary: total item stacks and a sample of them. ship = await client().get(f"/characters/{cid}/ship/", token=token) or {}
online = await client().get(f"/characters/{cid}/online/", token=token) or {}
The full asset list can be thousands of stacks, so only the first ``limit`` return {
are returned alongside the total count. "solar_system_id": loc.get("solar_system_id"),
""" "station_id": loc.get("station_id"),
token, cid = await _auth() "structure_id": loc.get("structure_id"),
assets = await client().get_all_pages(f"/characters/{cid}/assets/", token=token) "ship_type_id": ship.get("ship_type_id"),
sample = [ "ship_name": ship.get("ship_name"),
{ "online": online.get("online"),
"type_id": a.get("type_id"), "last_login": online.get("last_login"),
"quantity": a.get("quantity"),
"location_id": a.get("location_id"),
"location_flag": a.get("location_flag"),
"is_singleton": a.get("is_singleton"),
} }
for a in assets[: max(1, limit)]
] return await _run(character, fetch)
return {"character_id": cid, "total_stacks": len(assets), "sample": sample}
async def get_clones(character: str | int | None = None) -> dict[str, Any]:
"""Home station, jump clones, and active implants for one character or all."""
async def fetch(token: str, cid: int) -> dict[str, Any]:
clones = await client().get(f"/characters/{cid}/clones/", token=token) or {}
implants = await client().get(f"/characters/{cid}/implants/", token=token) or []
return {
"home_location": clones.get("home_location"),
"jump_clones": len(clones.get("jump_clones") or []),
"active_implants": implants,
}
return await _run(character, fetch)
async def get_assets(character: str | int | None = None, limit: int = 20) -> dict[str, Any]:
"""Asset summary (total stacks + a sample) for one character or all."""
async def fetch(token: str, cid: int) -> dict[str, Any]:
assets = await client().get_all_pages(f"/characters/{cid}/assets/", token=token)
sample = [
{
"type_id": a.get("type_id"),
"quantity": a.get("quantity"),
"location_id": a.get("location_id"),
"location_flag": a.get("location_flag"),
"is_singleton": a.get("is_singleton"),
}
for a in assets[: max(1, limit)]
]
return {"total_stacks": len(assets), "sample": sample}
return await _run(character, fetch)
TOOLS = [ TOOLS = [
+95 -38
View File
@@ -146,8 +146,10 @@ def _load_key() -> bytes:
return key return key
def save_tokens(store: dict[str, Any]) -> None: # --------------------------------------------------------------------------- #
"""Encrypt and persist the token bundle.""" # Multi-character store: { "characters": { "<id>": <bundle>, ... } }
# --------------------------------------------------------------------------- #
def _write_store(store: dict[str, Any]) -> None:
blob = Fernet(_load_key()).encrypt(json.dumps(store).encode()) blob = Fernet(_load_key()).encrypt(json.dumps(store).encode())
tf = config.token_file() tf = config.token_file()
tf.write_bytes(blob) tf.write_bytes(blob)
@@ -157,21 +159,70 @@ def save_tokens(store: dict[str, Any]) -> None:
pass pass
def load_tokens() -> dict[str, Any] | None: def load_store() -> dict[str, Any]:
"""Load and decrypt the token bundle, or None if not logged in.""" """Load the decrypted store, migrating a legacy single-character file.
Always returns a dict with a ``characters`` mapping (possibly empty).
"""
tf = config.token_file() tf = config.token_file()
if not tf.exists(): if not tf.exists():
return None return {"characters": {}}
return json.loads(Fernet(_load_key()).decrypt(tf.read_bytes())) data = json.loads(Fernet(_load_key()).decrypt(tf.read_bytes()))
if "characters" in data:
return data
# Legacy format: a single bundle at the top level. Migrate it.
if "refresh_token" in data and "character_id" in data:
return {"characters": {str(data["character_id"]): data}}
return {"characters": {}}
def clear_tokens() -> bool: def _save_character(bundle: dict[str, Any]) -> None:
"""Remove stored credentials. Returns True if something was removed.""" store = load_store()
tf = config.token_file() store["characters"][str(bundle["character_id"])] = bundle
if tf.exists(): _write_store(store)
tf.unlink()
def list_characters() -> list[dict[str, Any]]:
"""Logged-in characters (id, name, scopes), sorted by name."""
chars = load_store()["characters"].values()
out = [
{
"character_id": b.get("character_id"),
"character_name": b.get("character_name"),
"scopes": b.get("scopes"),
}
for b in chars
]
return sorted(out, key=lambda c: (c.get("character_name") or "").lower())
def remove_character(identifier: str | int | None = None) -> bool:
"""Remove one character (by id or name) or, if None, all of them."""
store = load_store()
chars = store["characters"]
if not chars:
return False
if identifier is None:
store["characters"] = {}
_write_store(store)
return True return True
return False target = _match_character(identifier, chars)
if target is None:
return False
del chars[target]
_write_store(store)
return True
def _match_character(identifier: str | int, chars: dict[str, Any]) -> str | None:
"""Return the store key for a character matched by id or (case-insensitive) name."""
ident = str(identifier).strip()
if ident in chars:
return ident
for key, bundle in chars.items():
if (bundle.get("character_name") or "").lower() == ident.lower():
return key
return None
def _bundle_from_response(resp: dict[str, Any], *, now: float) -> dict[str, Any]: def _bundle_from_response(resp: dict[str, Any], *, now: float) -> dict[str, Any]:
@@ -186,27 +237,40 @@ def _bundle_from_response(resp: dict[str, Any], *, now: float) -> dict[str, Any]
} }
def resolve_character(identifier: str | int) -> int:
"""Resolve a name-or-id to a logged-in character id, or raise."""
chars = load_store()["characters"]
key = _match_character(identifier, chars)
if key is None:
known = ", ".join(c["character_name"] or str(c["character_id"]) for c in list_characters())
raise NotAuthenticated(f"No logged-in character matching {identifier!r}. Known: {known}.")
return int(key)
async def get_auth( async def get_auth(
*, now: float | None = None, transport: httpx.AsyncBaseTransport | None = None character_id: int,
*,
now: float | None = None,
transport: httpx.AsyncBaseTransport | None = None,
) -> tuple[str, int]: ) -> tuple[str, int]:
"""Return ``(access_token, character_id)``, refreshing if near expiry. """Return ``(access_token, character_id)`` for one character, refreshing if needed.
Raises :class:`NotAuthenticated` if the user has not logged in. Raises :class:`NotAuthenticated` if that character is not logged in.
""" """
store = load_tokens() store = load_store()
if not store: bundle = store["characters"].get(str(character_id))
raise NotAuthenticated("Not logged in. Run `familiar login` in a terminal first.") if not bundle:
raise NotAuthenticated(f"Character {character_id} is not logged in. Run `familiar login`.")
now = now if now is not None else time.time() now = now if now is not None else time.time()
if store.get("access_token") and store.get("expires_at", 0) - now > 60: if bundle.get("access_token") and bundle.get("expires_at", 0) - now > 60:
return store["access_token"], store["character_id"] return bundle["access_token"], character_id
resp = await refresh_tokens(store["refresh_token"], transport=transport) resp = await refresh_tokens(bundle["refresh_token"], transport=transport)
bundle = _bundle_from_response(resp, now=now) fresh = _bundle_from_response(resp, now=now)
# Preserve the known character even if the new token omits a field. fresh["character_id"] = character_id
bundle["character_id"] = store.get("character_id", bundle["character_id"]) fresh["character_name"] = fresh["character_name"] or bundle.get("character_name")
bundle["character_name"] = bundle["character_name"] or store.get("character_name") _save_character(fresh)
save_tokens(bundle) return fresh["access_token"], character_id
return bundle["access_token"], bundle["character_id"]
_SUCCESS_PAGE = ( _SUCCESS_PAGE = (
@@ -260,18 +324,11 @@ def interactive_login(*, open_browser: bool = True) -> dict[str, Any]:
raise AuthError("State mismatch — possible CSRF; aborting login.") raise AuthError("State mismatch — possible CSRF; aborting login.")
resp = asyncio.run(exchange_code(captured["code"], verifier)) resp = asyncio.run(exchange_code(captured["code"], verifier))
bundle = _bundle_from_response(resp, now=time.time()) bundle = _bundle_from_response(resp, now=time.time())
save_tokens(bundle) _save_character(bundle) # adds or updates this character, keeping the others
return bundle return bundle
def status() -> dict[str, Any]: def status() -> dict[str, Any]:
"""Report whether we're logged in and as whom (no network).""" """Report which characters are logged in (no network call)."""
store = load_tokens() chars = list_characters()
if not store: return {"authenticated": bool(chars), "characters": chars}
return {"authenticated": False}
return {
"authenticated": True,
"character_id": store.get("character_id"),
"character_name": store.get("character_name"),
"scopes": store.get("scopes"),
}
+78 -44
View File
@@ -1,100 +1,134 @@
"""Authenticated account tools (auth mocked, ESI mocked).""" """Authenticated account tools (auth mocked, ESI mocked), multi-character."""
import asyncio import asyncio
import httpx import httpx
import pytest import pytest
from conftest import route
from familiar import account, auth from familiar import account, auth
from familiar.server import mcp from familiar.server import mcp
from conftest import route
CHARS = [
{"character_id": 99, "character_name": "Alpha", "scopes": []},
{"character_id": 100, "character_name": "Beta", "scopes": []},
]
@pytest.fixture @pytest.fixture
def fake_auth(monkeypatch): def fake_auth(monkeypatch):
async def _stub(): monkeypatch.setattr(auth, "list_characters", lambda: list(CHARS))
return "fake-token", 99
monkeypatch.setattr(auth, "get_auth", _stub) async def get_auth(cid, **kw):
return 99 return f"token-{cid}", cid
monkeypatch.setattr(auth, "get_auth", get_auth)
def resolve(identifier):
for c in CHARS:
if (
str(c["character_id"]) == str(identifier)
or (c["character_name"] or "").lower() == str(identifier).lower()
):
return c["character_id"]
raise auth.NotAuthenticated("no such character")
monkeypatch.setattr(auth, "resolve_character", resolve)
return CHARS
def test_get_wallet(mock_esi, fake_auth): def test_wallet_all_characters(mock_esi, fake_auth):
mock_esi(lambda req: httpx.Response(200, json=1234567.89)) mock_esi(lambda req: httpx.Response(200, json=1000.0))
result = asyncio.run(account.get_wallet()) result = asyncio.run(account.get_wallet())
assert result["character_id"] == 99 assert len(result["characters"]) == 2
assert result["balance"] == 1234567.89 assert {c["character_name"] for c in result["characters"]} == {"Alpha", "Beta"}
assert all(c["balance"] == 1000.0 for c in result["characters"])
def test_get_skills_summarizes(mock_esi, fake_auth): def test_wallet_single_character_by_name(mock_esi, fake_auth):
mock_esi(lambda req: httpx.Response(200, json=500.0))
result = asyncio.run(account.get_wallet(character="Alpha"))
assert len(result["characters"]) == 1
assert result["characters"][0]["character_id"] == 99
def test_unknown_character_raises(mock_esi, fake_auth):
mock_esi(lambda req: httpx.Response(200, json=1.0))
with pytest.raises(auth.NotAuthenticated):
asyncio.run(account.get_wallet(character="Ghost"))
def test_not_logged_in_raises(monkeypatch):
monkeypatch.setattr(auth, "list_characters", lambda: [])
with pytest.raises(auth.NotAuthenticated):
asyncio.run(account.get_wallet())
def test_skills_summary(mock_esi, fake_auth):
body = {"total_sp": 5_000_000, "unallocated_sp": 10_000, "skills": [{}, {}, {}]} body = {"total_sp": 5_000_000, "unallocated_sp": 10_000, "skills": [{}, {}, {}]}
mock_esi(lambda req: route(req, {"/skills/": body})) mock_esi(lambda req: route(req, {"/skills/": body}))
result = asyncio.run(account.get_skills()) result = asyncio.run(account.get_skills(character=99))
assert result["total_sp"] == 5_000_000 assert result["characters"][0]["total_sp"] == 5_000_000
assert result["skills_trained"] == 3 assert result["characters"][0]["skills_trained"] == 3
def test_get_skill_queue_limits(mock_esi, fake_auth): def test_skill_queue_limit(mock_esi, fake_auth):
queue = [ queue = [{"skill_id": i, "finished_level": 5, "queue_position": i} for i in range(5)]
{"skill_id": i, "finished_level": 5, "queue_position": i, "finish_date": "2026"}
for i in range(5)
]
mock_esi(lambda req: route(req, {"/skillqueue/": queue})) mock_esi(lambda req: route(req, {"/skillqueue/": queue}))
result = asyncio.run(account.get_skill_queue(limit=2)) result = asyncio.run(account.get_skill_queue(character="Beta", limit=2))
assert result["in_queue"] == 5 entry = result["characters"][0]
assert len(result["upcoming"]) == 2 assert entry["in_queue"] == 5
assert len(entry["upcoming"]) == 2
def test_get_location_merges_endpoints(mock_esi, fake_auth): def test_location_merges_endpoints(mock_esi, fake_auth):
fixtures = { fixtures = {
"/location/": {"solar_system_id": 30000142, "station_id": 60003760}, "/location/": {"solar_system_id": 30000142, "station_id": 60003760},
"/ship/": {"ship_type_id": 670, "ship_name": "My Capsule"}, "/ship/": {"ship_type_id": 670, "ship_name": "My Capsule"},
"/online/": {"online": True, "last_login": "2026-06-27T00:00:00Z"}, "/online/": {"online": True, "last_login": "2026-06-27T00:00:00Z"},
} }
mock_esi(lambda req: route(req, fixtures)) mock_esi(lambda req: route(req, fixtures))
result = asyncio.run(account.get_location()) result = asyncio.run(account.get_location(character=99))
assert result["solar_system_id"] == 30000142 entry = result["characters"][0]
assert result["ship_name"] == "My Capsule" assert entry["solar_system_id"] == 30000142
assert result["online"] is True assert entry["ship_name"] == "My Capsule"
assert entry["online"] is True
def test_get_clones(mock_esi, fake_auth): def test_clones(mock_esi, fake_auth):
fixtures = { fixtures = {
"/clones/": {"home_location": {"location_id": 60003760}, "jump_clones": [{}, {}]}, "/clones/": {"home_location": {"location_id": 60003760}, "jump_clones": [{}, {}]},
"/implants/": [10000, 10001], "/implants/": [10000, 10001],
} }
mock_esi(lambda req: route(req, fixtures)) mock_esi(lambda req: route(req, fixtures))
result = asyncio.run(account.get_clones()) result = asyncio.run(account.get_clones(character=99))
assert result["jump_clones"] == 2 entry = result["characters"][0]
assert result["active_implants"] == [10000, 10001] assert entry["jump_clones"] == 2
assert entry["active_implants"] == [10000, 10001]
def test_get_assets_paginates(mock_esi, fake_auth): def test_assets_paginates(mock_esi, fake_auth):
# Two pages of assets; the client should follow X-Pages and concatenate.
def handler(req): def handler(req):
page = req.url.params.get("page", "1") page = req.url.params.get("page", "1")
if page == "1": type_id = 34 if page == "1" else 35
return httpx.Response(
200,
json=[{"type_id": 34, "quantity": 1, "location_id": 1}],
headers={"X-Pages": "2"},
)
return httpx.Response( return httpx.Response(
200, 200,
json=[{"type_id": 35, "quantity": 2, "location_id": 1}], json=[{"type_id": type_id, "quantity": 1, "location_id": 1}],
headers={"X-Pages": "2"}, headers={"X-Pages": "2"},
) )
mock_esi(handler) mock_esi(handler)
result = asyncio.run(account.get_assets(limit=20)) result = asyncio.run(account.get_assets(character=99, limit=20))
assert result["total_stacks"] == 2 entry = result["characters"][0]
assert {a["type_id"] for a in result["sample"]} == {34, 35} assert entry["total_stacks"] == 2
assert {a["type_id"] for a in entry["sample"]} == {34, 35}
def test_account_status_delegates(monkeypatch): def test_account_status_delegates(monkeypatch):
monkeypatch.setattr(auth, "status", lambda: {"authenticated": True, "character_id": 1}) monkeypatch.setattr(auth, "status", lambda: {"authenticated": True, "characters": CHARS})
result = asyncio.run(account.account_status()) result = asyncio.run(account.account_status())
assert result["authenticated"] is True assert result["authenticated"] is True
assert len(result["characters"]) == 2
def test_account_tools_registered(): def test_account_tools_registered():
+69 -37
View File
@@ -1,4 +1,4 @@
"""EVE SSO auth: PKCE, JWT claims, token store, exchange/refresh, get_auth.""" """EVE SSO auth: PKCE, JWT claims, multi-character store, exchange/refresh."""
import asyncio import asyncio
import base64 import base64
@@ -18,6 +18,19 @@ def tmp_store(tmp_path, monkeypatch):
return tmp_path return tmp_path
def _char(cid, name, **extra):
return {
"refresh_token": f"r-{cid}",
"access_token": f"a-{cid}",
"expires_at": 10_000,
"character_id": cid,
"character_name": name,
"scopes": [],
**extra,
}
# --- pure helpers ---------------------------------------------------------- #
def test_generate_pkce_challenge_matches_verifier(): def test_generate_pkce_challenge_matches_verifier():
verifier, challenge = auth.generate_pkce() verifier, challenge = auth.generate_pkce()
expected = ( expected = (
@@ -33,11 +46,10 @@ def test_build_authorize_url_has_required_params():
assert "code_challenge=the-challenge" in url assert "code_challenge=the-challenge" in url
assert "code_challenge_method=S256" in url assert "code_challenge_method=S256" in url
assert "state=the-state" in url assert "state=the-state" in url
assert "client_id=" in url
def test_decode_jwt_and_character_id(): def test_decode_jwt_and_character_id():
token = make_jwt({"sub": "CHARACTER:EVE:2112000001", "name": "Latte", "scp": ["a"]}) token = make_jwt({"sub": "CHARACTER:EVE:2112000001", "name": "Latte"})
claims = auth.decode_jwt_claims(token) claims = auth.decode_jwt_claims(token)
assert claims["name"] == "Latte" assert claims["name"] == "Latte"
assert auth.character_id_from_claims(claims) == 2112000001 assert auth.character_id_from_claims(claims) == 2112000001
@@ -53,28 +65,58 @@ def test_decode_malformed_token_raises():
auth.decode_jwt_claims("not-a-jwt") auth.decode_jwt_claims("not-a-jwt")
def test_token_store_round_trip(tmp_store): # --- multi-character store ------------------------------------------------- #
bundle = {"refresh_token": "r", "character_id": 7, "character_name": "X"} def test_store_starts_empty(tmp_store):
assert auth.load_tokens() is None assert auth.load_store() == {"characters": {}}
auth.save_tokens(bundle) assert auth.list_characters() == []
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): def test_save_and_list_multiple_characters(tmp_store):
auth.save_tokens({"refresh_token": "super-secret-value", "character_id": 1}) auth._save_character(_char(1, "Beta"))
auth._save_character(_char(2, "Alpha"))
chars = auth.list_characters()
assert [c["character_name"] for c in chars] == ["Alpha", "Beta"] # sorted by name
def test_legacy_single_bundle_is_migrated(tmp_store):
# Write a pre-multi-character file (single bundle at top level).
auth._write_store({"refresh_token": "r", "character_id": 42, "character_name": "Old"})
store = auth.load_store()
assert "42" in store["characters"]
assert auth.list_characters()[0]["character_name"] == "Old"
def test_remove_character_by_name_and_all(tmp_store):
auth._save_character(_char(1, "Beta"))
auth._save_character(_char(2, "Alpha"))
assert auth.remove_character("Alpha") is True
assert {c["character_id"] for c in auth.list_characters()} == {1}
assert auth.remove_character("nobody") is False
assert auth.remove_character() is True # removes all
assert auth.list_characters() == []
assert auth.remove_character() is False
def test_resolve_character_by_id_and_name(tmp_store):
auth._save_character(_char(7, "Gamma"))
assert auth.resolve_character(7) == 7
assert auth.resolve_character("gamma") == 7
with pytest.raises(auth.NotAuthenticated):
auth.resolve_character("ghost")
def test_stored_tokens_are_encrypted(tmp_store):
auth._save_character(_char(1, "X", refresh_token="super-secret-value"))
raw = (tmp_store / "token.enc").read_bytes() raw = (tmp_store / "token.enc").read_bytes()
assert b"super-secret-value" not in raw assert b"super-secret-value" not in raw
# --- token requests + get_auth -------------------------------------------- #
def _token_transport(payload): def _token_transport(payload):
return httpx.MockTransport(lambda req: httpx.Response(200, json=payload)) return httpx.MockTransport(lambda req: httpx.Response(200, json=payload))
def test_exchange_code_posts_and_parses(): def test_exchange_code_parses():
payload = {"access_token": "a", "refresh_token": "r", "expires_in": 1200} payload = {"access_token": "a", "refresh_token": "r", "expires_in": 1200}
out = asyncio.run(auth.exchange_code("code", "verifier", transport=_token_transport(payload))) out = asyncio.run(auth.exchange_code("code", "verifier", transport=_token_transport(payload)))
assert out["refresh_token"] == "r" assert out["refresh_token"] == "r"
@@ -86,41 +128,31 @@ def test_token_request_raises_on_error():
asyncio.run(auth.refresh_tokens("r", transport=transport)) asyncio.run(auth.refresh_tokens("r", transport=transport))
def test_get_auth_without_login_raises(tmp_store): def test_get_auth_unknown_character_raises(tmp_store):
with pytest.raises(auth.NotAuthenticated): with pytest.raises(auth.NotAuthenticated):
asyncio.run(auth.get_auth()) asyncio.run(auth.get_auth(123))
def test_get_auth_uses_unexpired_token(tmp_store): def test_get_auth_uses_unexpired_token(tmp_store):
auth.save_tokens( auth._save_character(_char(9, "Z", access_token="still-good", expires_at=10_000))
{ token, cid = asyncio.run(auth.get_auth(9, now=5_000))
"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 token == "still-good"
assert cid == 9 assert cid == 9
def test_get_auth_refreshes_when_expired(tmp_store): def test_get_auth_refreshes_when_expired(tmp_store):
auth.save_tokens( auth._save_character(_char(9, "Z", refresh_token="r0", expires_at=1_000))
{"access_token": "old", "refresh_token": "r0", "character_id": 9, "expires_at": 1_000} new_access = make_jwt({"sub": "CHARACTER:EVE:9", "name": "Z"})
)
new_access = make_jwt({"sub": "CHARACTER:EVE:9", "name": "Latte"})
payload = {"access_token": new_access, "refresh_token": "r1", "expires_in": 1200} 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))) token, cid = asyncio.run(auth.get_auth(9, now=2_000, transport=_token_transport(payload)))
assert token == new_access assert token == new_access
assert cid == 9 assert cid == 9
# The rotated refresh token is persisted. assert auth.load_store()["characters"]["9"]["refresh_token"] == "r1"
assert auth.load_tokens()["refresh_token"] == "r1"
def test_status_reflects_store(tmp_store): def test_status_lists_characters(tmp_store):
assert auth.status() == {"authenticated": False} assert auth.status() == {"authenticated": False, "characters": []}
auth.save_tokens({"refresh_token": "r", "character_id": 3, "character_name": "Y", "scopes": []}) auth._save_character(_char(3, "Y"))
info = auth.status() info = auth.status()
assert info["authenticated"] is True assert info["authenticated"] is True
assert info["character_id"] == 3 assert info["characters"][0]["character_id"] == 3