Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 30 additions & 0 deletions agent_assembly/core/gateway_resolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,10 @@

from __future__ import annotations

import logging
import os
import shutil
import stat
import subprocess
import time
import warnings
Expand All @@ -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
Expand All @@ -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.

Expand Down Expand Up @@ -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):
Expand Down
34 changes: 34 additions & 0 deletions test/unit/core/test_gateway_resolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

from __future__ import annotations

import logging
import sys
import warnings
from pathlib import Path
from unittest.mock import MagicMock, patch
Expand Down Expand Up @@ -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