diff --git a/docs/audit-logging.md b/docs/audit-logging.md index 4494ceb..b6a6068 100644 --- a/docs/audit-logging.md +++ b/docs/audit-logging.md @@ -63,6 +63,10 @@ keep elsewhere. The log is written to `AUDIT_LOG_PATH`, default `DATA_DIR/audit.log` (`/var/lib/aegis-mcp` in the image). Keep it on durable storage, readable and writable only by the service account. +The server creates a missing state directory with mode 0700 and the log with mode 0600, tightens +an older log it owns that is group or world readable, and refuses to open the log through a +symlink (startup exits with code 2). A keyed (HMAC) chain is not offered: the chain is +tamper-evident only, so ship the log off the host as the integrity control. The server opens the file for append at startup and exits with code 2 (message on stderr) when that fails, so it never runs unaudited. The check does not stop at startup: each call to a known diff --git a/src/aegis_gitea_mcp/audit.py b/src/aegis_gitea_mcp/audit.py index 78ed295..60d2028 100644 --- a/src/aegis_gitea_mcp/audit.py +++ b/src/aegis_gitea_mcp/audit.py @@ -5,6 +5,8 @@ from __future__ import annotations import contextlib import hashlib import json +import os +import stat import threading import uuid from datetime import datetime, timezone @@ -33,12 +35,37 @@ class AuditLogger: """ self.settings = get_settings() self.log_path = log_path or self.settings.resolved_audit_log_path - self.log_path.parent.mkdir(parents=True, exist_ok=True) + # A new state directory is private (0700); an existing one is left alone. + self.log_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) self._lock = threading.Lock() # Held open for the process lifetime; closed in close(). - self._log_file = open(self.log_path, "a+", encoding="utf-8") # noqa: SIM115 + self._log_file = self._open_log() self._last_hash = self._read_last_hash() + def _open_log(self) -> Any: + """Open the log for append: mode 0600 when created, and never through a symlink. + + Raises: + OSError: The path cannot be opened (a symlink at the path gives ``ELOOP``). + """ + flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(self.log_path, flags, 0o600) + try: + info = os.fstat(descriptor) + if not stat.S_ISREG(info.st_mode): + raise OSError(f"{self.log_path} is not a regular file") + if ( + info.st_uid == getattr(os, "geteuid", lambda: info.st_uid)() + and stat.S_IMODE(info.st_mode) & 0o077 + ): + # An older log created under a permissive umask: logins and paths of + # every call must not stay readable by other local users. + os.fchmod(descriptor, 0o600) + return os.fdopen(descriptor, "a", encoding="utf-8") + except BaseException: + os.close(descriptor) + raise + def _read_last_hash(self) -> str: """Read the previous hash from the last log entry.""" try: diff --git a/src/aegis_gitea_mcp/transports.py b/src/aegis_gitea_mcp/transports.py index 8506737..6af75bb 100644 --- a/src/aegis_gitea_mcp/transports.py +++ b/src/aegis_gitea_mcp/transports.py @@ -8,6 +8,7 @@ actionable message on stderr (never on stdout, which carries JSON-RPC on stdio). from __future__ import annotations import asyncio +import errno import logging import sys import time @@ -428,7 +429,9 @@ def _audit_startup_error(settings: Settings) -> str | None: try: get_audit_logger() except OSError as exc: - reason = exc.strerror or type(exc).__name__ + reason = exc.strerror or str(exc) or type(exc).__name__ + if exc.errno == errno.ELOOP: + reason = "the path is a symbolic link, which is refused" return ( f"cannot open the audit log {settings.resolved_audit_log_path} for append " f"({reason}). Fix the permissions or point AUDIT_LOG_PATH (or DATA_DIR) at a " diff --git a/tests/security/test_audit_file.py b/tests/security/test_audit_file.py new file mode 100644 index 0000000..83e530a --- /dev/null +++ b/tests/security/test_audit_file.py @@ -0,0 +1,101 @@ +"""The audit file is private and is never reached through a symlink (SEC-11).""" + +from __future__ import annotations + +import errno +import os +import stat +from pathlib import Path + +import pytest +from pytest_httpx import HTTPXMock + +from aegis_gitea_mcp import transports +from aegis_gitea_mcp.audit import AuditLogger, get_audit_logger, validate_audit_log_integrity +from aegis_gitea_mcp.transports import run_http +from tests.test_server_helpers import API, connected, mock_user, server_settings + + +@pytest.fixture(autouse=True) +def _env(mock_env: None, monkeypatch: pytest.MonkeyPatch) -> None: + """The audit logger reads the process-wide settings; a startup that passes must not serve.""" + + def refuse(*args: object, **kwargs: object) -> None: + raise AssertionError("uvicorn.run must not be reached") + + monkeypatch.setattr(transports.uvicorn, "run", refuse) + + +@pytest.fixture +def open_umask() -> object: + """A permissive umask: what an unprepared host or container would hand out.""" + previous = os.umask(0) + yield None + os.umask(previous) + + +def mode_of(path: Path) -> int: + return stat.S_IMODE(path.stat().st_mode) + + +async def test_audit_log_is_0600_and_a_new_state_directory_is_0700( + httpx_mock: HTTPXMock, open_umask: object +) -> None: + mock_user(httpx_mock) + httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={}) + async with connected() as client: + await client.call_tool("gitea_repo_get", {"owner": "octo", "repo": "demo"}) + + path = get_audit_logger().log_path + assert mode_of(path) == 0o600 + assert mode_of(path.parent) == 0o700 + assert validate_audit_log_integrity(path) == (True, []) + + +def test_an_older_world_readable_log_is_tightened(tmp_path: Path) -> None: + path = tmp_path / "audit.log" + path.write_text("") + path.chmod(0o644) + + AuditLogger(path).close() + + assert mode_of(path) == 0o600 + + +def test_a_symlink_at_the_audit_path_is_not_followed(tmp_path: Path) -> None: + target = tmp_path / "elsewhere.log" + target.write_text("") + link = tmp_path / "audit.log" + link.symlink_to(target) + + with pytest.raises(OSError) as caught: + AuditLogger(link) + assert caught.value.errno == errno.ELOOP + assert target.read_text() == "" + + +def test_startup_exits_2_for_a_symlinked_audit_path( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + target = tmp_path / "elsewhere.log" + target.write_text("") + link = tmp_path / "state" / "audit.log" + link.parent.mkdir() + link.symlink_to(target) + monkeypatch.setenv("AUDIT_LOG_PATH", str(link)) + + assert run_http(server_settings(audit_log_path=link)) == 2 + + assert "symbolic link" in capsys.readouterr().err + assert target.read_text() == "" + + +def test_startup_exits_2_for_an_audit_path_that_is_a_directory( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + directory = tmp_path / "audit.log" + directory.mkdir() + monkeypatch.setenv("AUDIT_LOG_PATH", str(directory)) + + assert run_http(server_settings(audit_log_path=directory)) == 2 + assert "audit log" in capsys.readouterr().err