fix: refuse a token store directory owned by another user or left writable (SEC-13)

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