Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 39 additions & 16 deletions cli_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,11 @@

Called by pat_rotator._persist_token() every 10 minutes. Lightweight —
just swaps token values in existing files, no installs or script runs.

All writes are atomic (write to `.tmp`, then `os.replace`) so a Hermes / OpenCode
/ Codex invocation that reads the file mid-update sees the old token whole or
the new token whole — never a half-written file. Errors other than "file does
not exist" surface as warnings rather than being silently swallowed.
"""

import json
Expand All @@ -16,6 +21,20 @@
_HOME = "/app/python/source_code"


def _atomic_write_text(path, content):
"""Write `content` to `path` atomically via tmp file + rename.

Prevents the read-while-rewriting race that bit Hermes specifically:
Hermes reads `~/.hermes/config.yaml` on every invocation, so a bare
open(path, 'w') by the rotator could leave the file in a partial state
visible to a concurrent Hermes call → 403 Invalid access token.
"""
tmp = f"{path}.tmp"
with open(tmp, "w") as f:
f.write(content)
os.replace(tmp, path)


def update_cli_tokens(token):
"""Update the literal token in all CLI config files."""
_update_claude(token)
Expand All @@ -28,15 +47,16 @@ def update_cli_tokens(token):
def _update_claude(token):
"""Update ANTHROPIC_AUTH_TOKEN in ~/.claude/settings.json."""
path = os.path.join(_HOME, ".claude", "settings.json")
if not os.path.exists(path):
return # setup_claude.py hasn't run yet
try:
with open(path) as f:
settings = json.load(f)
if "env" in settings and "ANTHROPIC_AUTH_TOKEN" in settings["env"]:
settings["env"]["ANTHROPIC_AUTH_TOKEN"] = token
with open(path, "w") as f:
json.dump(settings, f, indent=2)
except (OSError, json.JSONDecodeError):
pass # file doesn't exist yet — initial setup hasn't run
_atomic_write_text(path, json.dumps(settings, indent=2))
except (OSError, json.JSONDecodeError) as e:
logger.warning("Failed to update Claude token in %s: %s", path, e)


def _update_codex(token):
Expand All @@ -48,6 +68,8 @@ def _update_codex(token):
def _update_opencode(token):
"""Update api_key values in ~/.local/share/opencode/auth.json."""
path = os.path.join(_HOME, ".local", "share", "opencode", "auth.json")
if not os.path.exists(path):
return # setup_opencode.py hasn't run yet
try:
with open(path) as f:
auth = json.load(f)
Expand All @@ -57,10 +79,9 @@ def _update_opencode(token):
provider["api_key"] = token
changed = True
if changed:
with open(path, "w") as f:
json.dump(auth, f, indent=2)
except (OSError, json.JSONDecodeError):
pass
_atomic_write_text(path, json.dumps(auth, indent=2))
except (OSError, json.JSONDecodeError) as e:
logger.warning("Failed to update OpenCode token in %s: %s", path, e)


def _update_gemini(token):
Expand All @@ -72,6 +93,8 @@ def _update_gemini(token):
def _update_hermes(token):
"""Update api_key lines in ~/.hermes/config.yaml."""
path = os.path.join(_HOME, ".hermes", "config.yaml")
if not os.path.exists(path):
return # setup_hermes.py hasn't run yet
try:
with open(path) as f:
content = f.read()
Expand All @@ -82,14 +105,15 @@ def _update_hermes(token):
flags=re.MULTILINE
)
if new_content != content:
with open(path, "w") as f:
f.write(new_content)
except OSError:
pass
_atomic_write_text(path, new_content)
except OSError as e:
logger.warning("Failed to update Hermes token in %s: %s", path, e)


def _replace_dotenv_key(path, key, value):
"""Replace a KEY=value line in a dotenv file."""
if not os.path.exists(path):
return # caller's setup script hasn't run yet
try:
with open(path) as f:
content = f.read()
Expand All @@ -100,7 +124,6 @@ def _replace_dotenv_key(path, key, value):
flags=re.MULTILINE
)
if new_content != content:
with open(path, "w") as f:
f.write(new_content)
except OSError:
pass
_atomic_write_text(path, new_content)
except OSError as e:
logger.warning("Failed to update %s in %s: %s", key, path, e)