mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
[F005,F006] grok auth: directory mount + atomic-write fallback
F005: the single-file bind mount of auth.json pinned the inode, so the orchestrator's atomic refresh (tmp+rename within ~/.grok) never reached a running grok container — a long-lived container hung at the login prompt when the original ~6h token expired. Mount the host ~/.grok DIRECTORY (ro) at /home/agent/.grok-auth-ro; the entrypoint symlinks ~/.grok/auth.json at that RO mount so grok + the --check backstop read the live credential (the directory mount sees the host-side rename) while grok's writable state (config.toml, sessions/) stays in the image's ~/.grok. F006: a rotated refresh_token is single-use — xAI invalidates the old one the instant it issues the new one. If the atomic write failed after the rotation, the file kept the now-dead old refresh_token and the credential was permanently lost on the next refresh. _atomic_write now falls back to a direct write when tmp+replace fails, so the rotated token always lands on disk (losing the write is catastrophic; losing atomicity is not). TDD: RED tests watched fail, then GREEN. ruff+mypy clean; 32 grok tests green, no regressions.
This commit is contained in:
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import pathlib
|
||||
from datetime import UTC, datetime
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
@@ -126,6 +127,38 @@ def test_refresh_failed_when_no_access_token(tmp_path: Path) -> None:
|
||||
assert ga.refresh_if_stale(path, post=lambda _u, _f: {"expires_in": 1}) == "failed"
|
||||
|
||||
|
||||
def test_refresh_persists_rotated_token_when_atomic_write_fails(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""F006: a rotated refresh_token is single-use — xAI invalidates the old one
|
||||
the moment it issues the new one. If the atomic write (tmp+replace) fails
|
||||
after the rotation, the file keeps the now-dead old refresh_token and the
|
||||
credential is permanently lost on the next refresh. The write must fall back
|
||||
to a direct write so the rotated refresh_token survives even when the atomic
|
||||
replace can't."""
|
||||
path = tmp_path / "auth.json"
|
||||
_write(path, _bundle(_PAST))
|
||||
|
||||
def _post(_url: str, _form: dict[str, str]) -> dict[str, Any]:
|
||||
return {
|
||||
"access_token": "new-access",
|
||||
"refresh_token": "rotated-rt",
|
||||
"expires_in": 21600,
|
||||
}
|
||||
|
||||
# Force the atomic tmp.replace to fail; the direct-write fallback must still
|
||||
# land the rotated refresh_token on disk.
|
||||
def _boom_replace(_self: pathlib.Path, _target: pathlib.Path) -> pathlib.Path:
|
||||
raise OSError("replace failed (simulated)")
|
||||
|
||||
monkeypatch.setattr(pathlib.Path, "replace", _boom_replace)
|
||||
|
||||
assert ga.refresh_if_stale(path, post=_post) == "refreshed"
|
||||
creds = next(iter(json.loads(path.read_text()).values()))
|
||||
assert creds["refresh_token"] == "rotated-rt" # survived the write failure
|
||||
assert creds["key"] == "new-access"
|
||||
|
||||
|
||||
def test_main_check_exit_codes(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
home = tmp_path / ".grok"
|
||||
home.mkdir()
|
||||
|
||||
@@ -241,7 +241,11 @@ async def test_grok_spawn_mounts_auth_when_present(_isolate_grok_auth: Path) ->
|
||||
) as exec_mock:
|
||||
await provider.spawn(_config())
|
||||
cmd = list(exec_mock.call_args.args)
|
||||
expected = f"{_isolate_grok_auth / 'auth.json'}:/home/agent/.grok/auth.json:ro"
|
||||
# F005: mount the host ~/.grok DIRECTORY (ro), not the single auth.json
|
||||
# file — a single-file bind mount pins the inode, so the orchestrator's
|
||||
# atomic auth.json refresh (rename) never reaches a running container.
|
||||
# The entrypoint symlinks ~/.grok/auth.json at this RO dir mount.
|
||||
expected = f"{_isolate_grok_auth}:/home/agent/.grok-auth-ro:ro"
|
||||
assert expected in cmd
|
||||
|
||||
|
||||
@@ -254,7 +258,7 @@ async def test_grok_spawn_omits_auth_mount_when_absent() -> None:
|
||||
) as exec_mock:
|
||||
await provider.spawn(_config())
|
||||
cmd = list(exec_mock.call_args.args)
|
||||
assert not any("/home/agent/.grok/auth.json" in c for c in cmd)
|
||||
assert not any("/home/agent/.grok-auth-ro" in c for c in cmd)
|
||||
|
||||
|
||||
async def test_grok_spawn_prompt_is_injection_safe() -> None:
|
||||
|
||||
@@ -82,7 +82,8 @@ def test_intake_grok_mounts_subscription_auth_when_present(
|
||||
cmd = AgentOrchestrator._build_intake_run_cmd(
|
||||
_intake_spec("grok", base_url="https://api.x.ai/v1", token="xai-key")
|
||||
)
|
||||
assert f"{grok_dir / 'auth.json'}:/home/agent/.grok/auth.json:ro" in cmd
|
||||
# F005: directory mount (ro), not the single-file inode-pinning mount.
|
||||
assert f"{grok_dir}:/home/agent/.grok-auth-ro:ro" in cmd
|
||||
|
||||
|
||||
def test_intake_anthropic_keeps_anthropic_env() -> None:
|
||||
|
||||
Reference in New Issue
Block a user