fix: create the audit log 0600 in a 0700 directory and refuse symlinks (SEC-11)

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

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:

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 "

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