From 6f7a3c0e5b4cc160deffd9622d2aa49ca93bc800 Mon Sep 17 00:00:00 2001 From: latte Date: Wed, 14 Oct 2026 17:36:21 +0200 Subject: [PATCH] fix: refuse a token store directory owned by another user or left writable (SEC-13) --- src/aegis_gitea_mcp/oauth/store.py | 35 ++++- tests/security/test_token_store.py | 221 +++++++++++++++++++++++++++++ 2 files changed, 252 insertions(+), 4 deletions(-) create mode 100644 tests/security/test_token_store.py diff --git a/src/aegis_gitea_mcp/oauth/store.py b/src/aegis_gitea_mcp/oauth/store.py index bc4587d..48accbd 100644 --- a/src/aegis_gitea_mcp/oauth/store.py +++ b/src/aegis_gitea_mcp/oauth/store.py @@ -13,7 +13,8 @@ What is stored: * sessions: the Gitea login and the user's Gitea access and refresh tokens; * our own access and refresh tokens, only as SHA-256 hashes. -File safety: the directory is created with mode 0700 and the file is written +File safety: the directory is created with mode 0700 (an existing one must be owned by the +running user and is tightened to 0700, otherwise startup stops) and the file is written with mode 0600 through a fresh ``O_EXCL`` temporary file and an atomic ``os.replace``. A symlink at the store path is refused. A file that cannot be decrypted or parsed stops startup with a clear message instead of being wiped, @@ -132,16 +133,40 @@ class TokenRecord: def _restrict_directory(directory: Path) -> None: - """Create the state directory with mode 0700, or tighten an existing one we own.""" + """Create the state directory with mode 0700, or check and tighten an existing one. + + Whoever controls the directory can swap the store for an older copy (which + resurrects revoked grants) or delete it, so an existing directory must belong + to the effective user and must not stay writable by group or others. + + Raises: + OAuthStoreError: The directory is owned by another user or remains writable + by group or others after an attempt to tighten it. + """ if not directory.exists(): directory.mkdir(mode=0o700, parents=True, exist_ok=True) return - mode = stat.S_IMODE(directory.stat().st_mode) + info = os.stat(directory) + if not stat.S_ISDIR(info.st_mode): + raise OAuthStoreError(f"{directory} is not a directory") + owner = getattr(os, "geteuid", lambda: info.st_uid)() + if info.st_uid != owner: + raise OAuthStoreError( + f"{directory} is owned by another user (uid {info.st_uid}, this process runs as " + f"uid {owner}); refusing to keep the OAuth token store there. Use a DATA_DIR " + "owned by the service user" + ) + mode = stat.S_IMODE(info.st_mode) if mode & 0o077: try: directory.chmod(0o700) except OSError: logger.warning("could not restrict the permissions of the state directory to 0700") + if stat.S_IMODE(os.stat(directory).st_mode) & 0o022: + raise OAuthStoreError( + f"{directory} is writable by group or others and could not be restricted; " + "refusing to keep the OAuth token store there. Run 'chmod 700' on it" + ) def _parse_document(document: Any) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: @@ -191,10 +216,12 @@ class OAuthStore: """Open the store, reading and decrypting the file when it exists. Raises: - OAuthStoreError: The path is a symlink or not a regular file, or the + OAuthStoreError: The directory belongs to another user or stays group or + world writable, the path is a symlink or not a regular file, or the file cannot be decrypted or parsed. """ store = cls(directory, secret, clock) + _restrict_directory(directory) store._read() store.prune() return store diff --git a/tests/security/test_token_store.py b/tests/security/test_token_store.py new file mode 100644 index 0000000..40107fa --- /dev/null +++ b/tests/security/test_token_store.py @@ -0,0 +1,221 @@ +"""The encrypted OAuth token store: permissions, encryption, key handling and ownership (SEC-13).""" + +from __future__ import annotations + +import os +import stat +from pathlib import Path +from typing import Any + +import pytest +from cryptography.fernet import Fernet +from pytest_httpx import HTTPXMock + +from aegis_gitea_mcp import transports +from aegis_gitea_mcp.oauth.store import ( + STORE_FILENAME, + OAuthStore, + OAuthStoreError, + SessionRecord, + derive_fernet_key, + token_hash, +) +from aegis_gitea_mcp.transports import run_http +from tests.test_oauth_helpers import ( + APP_CLIENT_SECRET, + ENCRYPTION_KEY, + USER_REFRESH, + USER_TOKEN, + browser, + build_oauth_app, + oauth_settings, + sign_in, +) + +OTHER_KEY = "another-key-0123456789abcdef0123456789" + + +@pytest.fixture(autouse=True) +def _env(mock_env: None) -> None: + """The audit logger reads the process-wide settings.""" + + +def a_session() -> SessionRecord: + return SessionRecord( + session_id="s1", + client_id="c1", + login="alice", + is_admin=False, + gitea_access_token=USER_TOKEN, + gitea_refresh_token=USER_REFRESH, + gitea_expires_at=None, + scopes=["gitea"], + resource="https://mcp.example.com/mcp", + created_at=1e12, + ) + + +async def test_file_is_0600_and_a_new_directory_is_0700(tmp_path: Path) -> None: + previous = os.umask(0) + try: + directory = tmp_path / "new" / "state" + store = OAuthStore.load(directory, ENCRYPTION_KEY) + store.sessions["s1"] = a_session() + await store.save() + finally: + os.umask(previous) + + assert stat.S_IMODE(directory.stat().st_mode) == 0o700 + assert stat.S_IMODE((directory / STORE_FILENAME).stat().st_mode) == 0o600 + assert [entry.name for entry in directory.iterdir()] == [STORE_FILENAME] + + +def test_a_symlinked_store_is_refused_and_its_target_untouched(tmp_path: Path) -> None: + target = tmp_path / "elsewhere" + target.write_bytes(b"keep me") + (tmp_path / STORE_FILENAME).symlink_to(target) + + with pytest.raises(OAuthStoreError, match="symlink"): + OAuthStore.load(tmp_path, ENCRYPTION_KEY) + assert target.read_bytes() == b"keep me" + + +async def test_a_symlink_swapped_in_later_is_not_written_through(tmp_path: Path) -> None: + store = OAuthStore.load(tmp_path, ENCRYPTION_KEY) + target = tmp_path / "elsewhere" + target.write_bytes(b"keep me") + (tmp_path / STORE_FILENAME).symlink_to(target) + + with pytest.raises(OAuthStoreError, match="symlink"): + await store.save() + assert target.read_bytes() == b"keep me" + + +async def test_the_file_holds_no_plaintext_and_our_tokens_only_as_hashes( + tmp_path: Path, httpx_mock: HTTPXMock +) -> None: + oauth = build_oauth_app(oauth_settings(data_dir=tmp_path)) + flow = sign_in(browser(oauth.app), httpx_mock, login="alice-the-user") + + raw = (tmp_path / STORE_FILENAME).read_bytes() + + for secret in ( + USER_TOKEN, + USER_REFRESH, + flow.client_secret, + flow.tokens["access_token"], + flow.tokens["refresh_token"], + "alice-the-user", + APP_CLIENT_SECRET, + ENCRYPTION_KEY, + ): + assert secret.encode() not in raw + plain = Fernet(derive_fernet_key(ENCRYPTION_KEY)).decrypt(raw).decode() + assert flow.tokens["access_token"] not in plain + assert flow.tokens["refresh_token"] not in plain + assert token_hash(flow.tokens["access_token"]) in plain + assert token_hash(flow.tokens["refresh_token"]) in plain + + +async def test_a_wrong_key_stops_startup_without_wiping_the_store( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + store = OAuthStore.load(tmp_path, ENCRYPTION_KEY) + store.sessions["s1"] = a_session() + await store.save() + before = (tmp_path / STORE_FILENAME).read_bytes() + served: list[Any] = [] + monkeypatch.setattr(transports.uvicorn, "run", lambda *a, **k: served.append(a)) + + code = run_http(oauth_settings(data_dir=tmp_path, token_encryption_key=OTHER_KEY)) + + assert code == 2 and served == [] + err = capsys.readouterr().err + assert "TOKEN_ENCRYPTION_KEY" in err and "move the file away" in err + assert OTHER_KEY not in err and ENCRYPTION_KEY not in err + assert (tmp_path / STORE_FILENAME).read_bytes() == before + + +def test_a_garbage_store_stops_startup_and_stays_on_disk(tmp_path: Path) -> None: + (tmp_path / STORE_FILENAME).write_bytes(b"not a fernet token") + + with pytest.raises(OAuthStoreError): + OAuthStore.load(tmp_path, ENCRYPTION_KEY) + assert (tmp_path / STORE_FILENAME).read_bytes() == b"not a fernet token" + + +# SEC-13: who owns the directory + + +def fake_owner(monkeypatch: pytest.MonkeyPatch, directory: Path) -> None: + """Report ``directory`` as owned by a different uid than the running process.""" + real_stat = os.stat + + def stat_other(path: Any, *args: Any, **kwargs: Any) -> os.stat_result: + info = real_stat(path, *args, **kwargs) + if Path(path) == directory: + fields = list(info) + fields[stat.ST_UID] = info.st_uid + 1 + return os.stat_result(fields) + return info + + monkeypatch.setattr(os, "stat", stat_other) + + +async def test_a_directory_owned_by_another_user_is_refused_on_load_and_save( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + directory = tmp_path / "state" + directory.mkdir(mode=0o700) + store = OAuthStore(directory, ENCRYPTION_KEY) + fake_owner(monkeypatch, directory) + + with pytest.raises(OAuthStoreError, match="owned by another user"): + OAuthStore.load(directory, ENCRYPTION_KEY) + with pytest.raises(OAuthStoreError, match="owned by another user"): + await store.save() + assert list(directory.iterdir()) == [] + + +async def test_a_group_or_world_writable_directory_that_cannot_be_tightened_is_refused( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + directory = tmp_path / "state" + directory.mkdir() + directory.chmod(0o777) + + def refuse(self: Path, mode: int) -> None: + raise PermissionError("not permitted") + + monkeypatch.setattr(Path, "chmod", refuse) + + with pytest.raises(OAuthStoreError, match="writable by group or others"): + OAuthStore.load(directory, ENCRYPTION_KEY) + with pytest.raises(OAuthStoreError, match="writable by group or others"): + await OAuthStore(directory, ENCRYPTION_KEY).save() + assert list(directory.iterdir()) == [] + + +async def test_a_loose_directory_we_own_is_tightened_not_refused(tmp_path: Path) -> None: + directory = tmp_path / "state" + directory.mkdir() + directory.chmod(0o777) + + await OAuthStore.load(directory, ENCRYPTION_KEY).save() + + assert stat.S_IMODE(directory.stat().st_mode) == 0o700 + + +def test_startup_exits_2_for_a_directory_owned_by_someone_else( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + directory = tmp_path / "state" + directory.mkdir(mode=0o700) + fake_owner(monkeypatch, directory) + served: list[Any] = [] + monkeypatch.setattr(transports.uvicorn, "run", lambda *a, **k: served.append(a)) + + assert run_http(oauth_settings(data_dir=directory)) == 2 + + assert served == [] + assert "owned by another user" in capsys.readouterr().err