102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
"""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
|