diff --git a/agent_assembly/core/gateway_resolver.py b/agent_assembly/core/gateway_resolver.py index 56aa4f3..80d75d7 100644 --- a/agent_assembly/core/gateway_resolver.py +++ b/agent_assembly/core/gateway_resolver.py @@ -17,8 +17,10 @@ from __future__ import annotations +import logging import os import shutil +import stat import subprocess import time import warnings @@ -29,6 +31,8 @@ from agent_assembly.exceptions import ConfigurationError, GatewayError +logger = logging.getLogger(__name__) + DEFAULT_GATEWAY_URL = "http://localhost:7391" DEFAULT_HEALTHZ_PATH = "/healthz" DEFAULT_PROBE_TIMEOUT_SECONDS = 0.5 @@ -50,6 +54,30 @@ _warned_legacy_env: set[str] = set() +def _warn_if_world_readable(path: Path) -> None: + """Emit a warning when ``path`` is readable by group or other. + + Defense-in-depth for ``~/.aasm/config.yaml``, which may contain an + ``api_key``. Follows the AWS CLI / gcloud / docker convention of + warning-not-refusing: the read still proceeds, but the operator is + prompted to ``chmod 600``. Any stat failure (missing, unreadable) is + swallowed — the caller has already checked existence and will handle + the read error path itself. + """ + try: + mode = stat.S_IMODE(path.stat().st_mode) + except OSError: + return + if mode & 0o077: + logger.warning( + "Config file %s has group/other-readable permissions (mode %o); " + "run `chmod 600 %s` to secure it.", + path, + mode, + path, + ) + + def _resolve_env(canonical: str, legacy: str) -> str | None: """Read ``canonical`` first, falling back to the deprecated ``legacy`` var. @@ -127,6 +155,8 @@ def _load_config_file(path: str = DEFAULT_CONFIG_FILE_PATH) -> dict[str, Any]: if not resolved.is_file(): return {} + _warn_if_world_readable(resolved) + try: loaded = yaml.safe_load(resolved.read_text(encoding="utf-8")) except (OSError, yaml.YAMLError): diff --git a/test/unit/core/test_gateway_resolver.py b/test/unit/core/test_gateway_resolver.py index 287aba6..d69a0f7 100644 --- a/test/unit/core/test_gateway_resolver.py +++ b/test/unit/core/test_gateway_resolver.py @@ -2,6 +2,8 @@ from __future__ import annotations +import logging +import sys import warnings from pathlib import Path from unittest.mock import MagicMock, patch @@ -272,3 +274,35 @@ def test_returns_empty_string_default(self, monkeypatch: pytest.MonkeyPatch) -> monkeypatch.delenv(gateway_resolver.ENV_API_KEY, raising=False) with patch.object(gateway_resolver, "_load_config_file", return_value={}): assert gateway_resolver.resolve_api_key() == "" + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX permission bits are not meaningful on Windows") +class TestWarnIfWorldReadable: + """Defense-in-depth permission warning for ~/.aasm/config.yaml (AAASM-4323).""" + + def test_no_warning_when_mode_is_0600( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + cfg = tmp_path / "config.yaml" + cfg.write_text("agent:\n api_key: k\n", encoding="utf-8") + cfg.chmod(0o600) + + with caplog.at_level(logging.WARNING, logger=_RESOLVER_MOD): + gateway_resolver._warn_if_world_readable(cfg) + + assert caplog.records == [] + + def test_warns_when_mode_is_0644( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + cfg = tmp_path / "config.yaml" + cfg.write_text("agent:\n api_key: k\n", encoding="utf-8") + cfg.chmod(0o644) + + with caplog.at_level(logging.WARNING, logger=_RESOLVER_MOD): + gateway_resolver._warn_if_world_readable(cfg) + + assert len(caplog.records) == 1 + message = caplog.records[0].getMessage() + assert "chmod 600" in message + assert "644" in message