From b4289200ba93fad96a2e1e87270e036613b14293 Mon Sep 17 00:00:00 2001 From: Que0x Date: Sat, 4 Jul 2026 04:18:00 +0300 Subject: [PATCH] fix(web-server): close OAuth token TOCTOU by writing 0o600 atomically MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_save_anthropic_oauth_creds` wrote the Anthropic OAuth token file with `os.replace(tmp, path)` followed by a post-hoc `chmod(0o600)`. Between the rename and the chmod the token file existed at the default umask (0o644 on most hosts) — a window in which another local user could read the access/refresh tokens. Write via `utils.atomic_json_write(..., mode=0o600)`, which creates the temp with mode 0o600 *before* any content is written, fsyncs, atomically replaces, preserves the existing file's owner, and cleans up its temp on failure. This matches the `atomic_json_write(mode=0o600)` call already used elsewhere in this module for the credential-pool write, and #56644's owner preservation. Tests updated for the new mechanism, plus a check that the write goes through `atomic_json_write(mode=0o600)` (mutation-verified). --- hermes_cli/web_server.py | 29 +++++----------- .../hermes_cli/test_web_server_oauth_write.py | 34 +++++++++++++++---- 2 files changed, 37 insertions(+), 26 deletions(-) diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 8232ab77c0a..d7ff480b390 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -7050,26 +7050,15 @@ def _save_anthropic_oauth_creds(access_token: str, refresh_token: str, expires_a "refreshToken": refresh_token, "expiresAt": expires_at_ms, } - oauth_file.parent.mkdir(parents=True, exist_ok=True) - tmp_path = oauth_file.with_name( - f"{oauth_file.name}.tmp.{os.getpid()}.{secrets.token_hex(8)}" - ) - try: - with tmp_path.open("w", encoding="utf-8") as handle: - handle.write(json.dumps(payload, indent=2)) - handle.flush() - os.fsync(handle.fileno()) - os.replace(tmp_path, oauth_file) - try: - oauth_file.chmod(stat.S_IRUSR | stat.S_IWUSR) - except OSError: - pass - finally: - try: - if tmp_path.exists(): - tmp_path.unlink() - except OSError: - pass + # atomic_json_write creates the temp with mode 0o600 (via mkstemp) *before* + # any content is written, then fsyncs and atomically replaces the target. + # The previous os.replace + post-hoc chmod left a TOCTOU window in which the + # OAuth token file was world-readable at the default umask (0o644 on most + # hosts) between the rename and the chmod. atomic_json_write also preserves + # the existing file's owner and cleans up its temp on failure. + from utils import atomic_json_write + + atomic_json_write(oauth_file, payload, indent=2, mode=0o600) # Best-effort credential-pool insert. Failure here doesn't invalidate # the file write — pool registration only matters for the rotation # strategy, not for runtime credential resolution. diff --git a/tests/hermes_cli/test_web_server_oauth_write.py b/tests/hermes_cli/test_web_server_oauth_write.py index 20200b9a73b..4289c781e1f 100644 --- a/tests/hermes_cli/test_web_server_oauth_write.py +++ b/tests/hermes_cli/test_web_server_oauth_write.py @@ -36,18 +36,40 @@ def test_dashboard_oauth_write_uses_owner_only_permissions(oauth_file): assert mode == 0o600 -def test_dashboard_oauth_write_uses_atomic_replace_and_cleans_temp_files(oauth_file, monkeypatch): - replace_calls = [] +def test_dashboard_oauth_write_is_atomic_and_cleans_temp_on_failure(oauth_file, monkeypatch): + """If the atomic replace fails, no partial file or temp file is left.""" + import utils def flaky_replace(src, dst): - replace_calls.append((src, dst)) raise OSError('simulated replace failure') - monkeypatch.setattr('hermes_cli.web_server.os.replace', flaky_replace) + monkeypatch.setattr(utils, 'atomic_replace', flaky_replace) with pytest.raises(OSError, match='simulated replace failure'): _save_anthropic_oauth_creds('access-token', 'refresh-token', 123456) - assert replace_calls, 'helper should attempt atomic os.replace()' assert not oauth_file.exists() - assert not list(oauth_file.parent.glob(f'{oauth_file.name}.tmp*')) + # atomic_json_write stages to ``._*.tmp`` and unlinks it on failure. + assert not list(oauth_file.parent.glob('*.tmp')) + + +def test_dashboard_oauth_write_uses_atomic_json_write_with_owner_only_mode(oauth_file, monkeypatch): + """The OAuth token file must be written 0o600 from creation via + ``atomic_json_write(mode=0o600)``, so it is never briefly world-readable + (the old ``os.replace`` + post-hoc ``chmod`` TOCTOU).""" + import utils + + calls = {} + real = utils.atomic_json_write + + def spy(path, data, **kwargs): + calls['mode'] = kwargs.get('mode') + return real(path, data, **kwargs) + + monkeypatch.setattr(utils, 'atomic_json_write', spy) + + _save_anthropic_oauth_creds('access-token', 'refresh-token', 123456) + + assert calls.get('mode') == 0o600, \ + 'OAuth creds must be written 0o600 atomically (no chmod-after-replace window)' + assert oauth_file.exists()