diff --git a/cron/scheduler.py b/cron/scheduler.py index c625ba6f7b..9994fafc1b 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -2346,18 +2346,40 @@ def _send_media_via_adapter( loop, job: dict, platform=None, -) -> None: +) -> list: """Send extracted MEDIA files as native platform attachments via a live adapter. Routes each file to the appropriate adapter method (send_voice, send_image_file, send_video, send_document) based on file extension — mirroring the routing logic in ``BasePlatformAdapter._process_message_background``. + + Returns a list of per-file error strings (empty when every attachment + delivered). Callers surface these into the job's delivery errors so a + dropped attachment is visible in ``last_error``/run status instead of + only in the gateway log (the silent-drop half of the manual-run + attachment bug: text delivered, file vanished, job marked ok). """ from pathlib import Path from gateway.platforms.base import BasePlatformAdapter, should_send_media_as_audio + errors: list = [] + requested = [(str(p), v) for p, v in (media_files or [])] media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files) + # Report paths the safety filter dropped: the model referenced them in + # MEDIA: tags but they will never be sent (missing file, denied prefix, + # or strict-mode policy miss). + kept = {p for p, _ in media_files} + for raw_path, _v in requested: + try: + from gateway.platforms.base import validate_media_delivery_path + + if validate_media_delivery_path(raw_path) not in kept: + errors.append( + f"attachment dropped by media path policy: {raw_path}" + ) + except Exception: + errors.append(f"attachment dropped by media path policy: {raw_path}") for media_path, _is_voice in media_files: try: @@ -2375,23 +2397,27 @@ def _send_media_via_adapter( from agent.async_utils import safe_schedule_threadsafe future = safe_schedule_threadsafe(coro, loop) if future is None: - logger.warning( - "Job '%s': cannot send media %s, gateway loop unavailable", - job.get("id", "?"), media_path, - ) - return + msg = f"cannot send media {media_path}: gateway loop unavailable" + logger.warning("Job '%s': %s", job.get("id", "?"), msg) + errors.append(msg) + return errors try: result = future.result(timeout=30) except TimeoutError: future.cancel() raise if result and not getattr(result, "success", True): - logger.warning( - "Job '%s': media send failed for %s: %s", - job.get("id", "?"), media_path, getattr(result, "error", "unknown"), + msg = ( + f"media send failed for {media_path}: " + f"{getattr(result, 'error', 'unknown')}" ) + logger.warning("Job '%s': %s", job.get("id", "?"), msg) + errors.append(msg) except Exception as e: - logger.warning("Job '%s': failed to send media %s: %s", job.get("id", "?"), media_path, e) + msg = f"failed to send media {media_path}: {e}" + logger.warning("Job '%s': %s", job.get("id", "?"), msg) + errors.append(msg) + return errors def _confirm_adapter_delivery(send_result) -> bool: @@ -2533,8 +2559,34 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option # Extract MEDIA: tags so attachments are forwarded as files, not raw text from gateway.platforms.base import BasePlatformAdapter + + # Bridge gateway media-policy config (strict / allow_dirs / trust_recent) + # into the env vars the path validator reads. Gateway startup does this + # at boot; a standalone process (manual `hermes cron run` from the CLI, + # a cron tick without the gateway) historically did NOT — so manual runs + # filtered attachment paths under a DIFFERENT policy than scheduled runs + # and silently dropped files the gateway would deliver. Idempotent, + # env-wins, never raises. + from gateway.media_policy import apply_media_policy_env + + apply_media_policy_env(user_cfg) + media_files, cleaned_delivery_content = BasePlatformAdapter.extract_media(delivery_content) + requested_media = [(str(p), v) for p, v in media_files] media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files) + # Attachments the policy filter dropped will never be sent on ANY lane — + # record them up front so the run status says so (previously one + # stderr WARNING was the only trace: text delivered, file vanished). + _policy_dropped = len(requested_media) - len(media_files) + policy_drop_errors = ( + [ + f"{_policy_dropped} media attachment(s) dropped by media path " + "policy (missing file, denied prefix, or strict-mode miss); " + "see gateway.strict / media_delivery_allow_dirs in config.yaml" + ] + if _policy_dropped > 0 + else [] + ) # Resolve the delivery-mirror gate ONCE (default off). When on, each # successful delivery is also appended to the target chat's gateway session @@ -2974,7 +3026,7 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option routed_media_metadata["user_id"] = logical_home.user_id if logical_home.scope_id: routed_media_metadata["scope_id"] = logical_home.scope_id - _send_media_via_adapter( + _media_errors = _send_media_via_adapter( runtime_adapter, chat_id, media_files, @@ -2983,6 +3035,12 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option job, platform=platform, ) + # Surface per-file failures into the run status (parity + # with the standalone lane): text delivered but an + # attachment didn't is a visible partial failure, not ok. + for _me in _media_errors: + _msg = f"{_me} (target {platform_name}:{chat_id})" + delivery_errors.append(_msg) elif timed_out and media_files: msg = ( f"{len(media_files)} media attachment(s) not delivered to " @@ -3120,6 +3178,19 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option delivery_errors.extend(target_errors) continue + # Standalone senders report per-file attachment failures in + # ``warnings`` while still returning success (the text leg + # delivered). Surface them: a cron whose PDF/image silently + # vanished used to mark the run ok with no trace — the exact + # "manual run delivers text but no attachment" field report. + _sender_warnings = ( + result.get("warnings") if isinstance(result, dict) else None + ) or [] + for _w in _sender_warnings: + msg = f"delivery warning: {_w} (target {platform_name}:{chat_id})" + logger.error("Job '%s': %s", job["id"], msg) + delivery_errors.append(msg) + logger.info("Job '%s': delivered to %s:%s", job["id"], platform_name, chat_id) _maybe_mirror_cron_delivery( job, platform_name, chat_id, mirror_text, @@ -3127,6 +3198,9 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option enabled=mirror_this_target and not thread_seeded, ) + if policy_drop_errors: + # Filter-time drops apply to every target; report them once. + delivery_errors.extend(policy_drop_errors) if delivery_errors: return "; ".join(delivery_errors) return None diff --git a/gateway/media_policy.py b/gateway/media_policy.py new file mode 100644 index 0000000000..5b4d671aae --- /dev/null +++ b/gateway/media_policy.py @@ -0,0 +1,88 @@ +"""Shared config→env bridge for media-delivery policy. + +``validate_media_delivery_path`` (gateway/platforms/base.py) reads its policy +from environment variables: + + - ``HERMES_MEDIA_DELIVERY_STRICT`` <- gateway.strict + - ``HERMES_MEDIA_ALLOW_DIRS`` <- gateway.media_delivery_allow_dirs + - ``HERMES_MEDIA_TRUST_RECENT_FILES`` <- gateway.trust_recent_files + +Historically the config.yaml -> env translation ran ONLY in gateway startup +(gateway/run.py), so any process that delivers media without booting the +gateway — a manual ``hermes cron run`` in the CLI, ``hermes send``, a +standalone cron tick — filtered MEDIA paths under DIFFERENT policy than the +gateway's scheduled deliveries. In strict/allowlisted enterprise deployments +that divergence silently dropped attachments from manual cron runs while +scheduled runs delivered them (text is unaffected — only media goes through +path validation). + +``apply_media_policy_env()`` is that same translation as a shared, idempotent +helper. Gateway startup calls it, and every standalone delivery entrypoint +calls it immediately before filtering media paths. + +Precedence: an explicitly-set environment variable WINS over config.yaml. +This preserves both the operator contract (env overrides are how deployments +pin behavior) and gateway/run.py's historical shape (it only wrote the env +var when the config key was present; we additionally refuse to overwrite a +pre-existing env value so a shell-exported override survives). +""" + +from __future__ import annotations + +import logging +import os +from typing import Any, Dict, Optional + +logger = logging.getLogger(__name__) + +_STRICT_ENV = "HERMES_MEDIA_DELIVERY_STRICT" +_ALLOW_DIRS_ENV = "HERMES_MEDIA_ALLOW_DIRS" +_TRUST_RECENT_ENV = "HERMES_MEDIA_TRUST_RECENT_FILES" + + +def _load_gateway_cfg(config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + if config is None: + try: + from hermes_cli.config import load_config + + config = load_config() or {} + except Exception: + return {} + gateway_cfg = config.get("gateway", {}) + return gateway_cfg if isinstance(gateway_cfg, dict) else {} + + +def apply_media_policy_env(config: Optional[Dict[str, Any]] = None) -> None: + """Bridge gateway media-policy settings from config.yaml into the env. + + Idempotent and env-wins: a variable already present in the environment is + never overwritten, so gateway startup (which runs this same helper) and + operator shell exports keep precedence. Never raises — a policy-bridge + failure must not break delivery; the validator falls back to its + defaults exactly as before. + """ + try: + gateway_cfg = _load_gateway_cfg(config) + if not gateway_cfg: + return + + strict = gateway_cfg.get("strict") + if strict is not None and not os.environ.get(_STRICT_ENV): + os.environ[_STRICT_ENV] = "1" if strict else "0" + + allow_dirs = gateway_cfg.get("media_delivery_allow_dirs") + if allow_dirs and not os.environ.get(_ALLOW_DIRS_ENV): + if isinstance(allow_dirs, str): + allow_dirs_str = allow_dirs + elif isinstance(allow_dirs, (list, tuple)): + allow_dirs_str = os.pathsep.join(str(p) for p in allow_dirs if p) + else: + allow_dirs_str = "" + if allow_dirs_str: + os.environ[_ALLOW_DIRS_ENV] = allow_dirs_str + + trust_recent = gateway_cfg.get("trust_recent_files") + if trust_recent is not None and not os.environ.get(_TRUST_RECENT_ENV): + os.environ[_TRUST_RECENT_ENV] = "1" if trust_recent else "0" + except Exception: # noqa: BLE001 - policy bridge must never break delivery + logger.debug("apply_media_policy_env failed", exc_info=True) diff --git a/gateway/run.py b/gateway/run.py index 17ef3b6409..657f706fd1 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2380,28 +2380,14 @@ if _config_path.exists(): if _redact is not None: os.environ["HERMES_REDACT_SECRETS"] = str(_redact).lower() # Gateway settings (media delivery allowlist + recency trust + strict mode) + # Delegated to the shared bridge so standalone delivery entrypoints + # (manual `hermes cron run`, ticks without the gateway) apply the SAME + # policy translation — process parity for attachment filtering. _gateway_cfg = _cfg.get("gateway", {}) if isinstance(_gateway_cfg, dict): - _strict = _gateway_cfg.get("strict") - if _strict is not None: - os.environ["HERMES_MEDIA_DELIVERY_STRICT"] = ( - "1" if _strict else "0" - ) - _allow_dirs = _gateway_cfg.get("media_delivery_allow_dirs") - if _allow_dirs: - if isinstance(_allow_dirs, str): - _allow_dirs_str = _allow_dirs - elif isinstance(_allow_dirs, (list, tuple)): - _allow_dirs_str = os.pathsep.join(str(p) for p in _allow_dirs if p) - else: - _allow_dirs_str = "" - if _allow_dirs_str: - os.environ["HERMES_MEDIA_ALLOW_DIRS"] = _allow_dirs_str - _trust_recent = _gateway_cfg.get("trust_recent_files") - if _trust_recent is not None: - os.environ["HERMES_MEDIA_TRUST_RECENT_FILES"] = ( - "1" if _trust_recent else "0" - ) + from gateway.media_policy import apply_media_policy_env + + apply_media_policy_env(_cfg) _trust_recent_seconds = _gateway_cfg.get("trust_recent_files_seconds") if _trust_recent_seconds is not None: os.environ["HERMES_MEDIA_TRUST_RECENT_SECONDS"] = str(_trust_recent_seconds) diff --git a/tests/cron/test_media_delivery_parity.py b/tests/cron/test_media_delivery_parity.py new file mode 100644 index 0000000000..8e6a6dd70d --- /dev/null +++ b/tests/cron/test_media_delivery_parity.py @@ -0,0 +1,330 @@ +"""Cron media-delivery parity — manual runs must not silently drop attachments. + +Field report (enterprise, v0.20.0/2026.8.3): cron jobs whose output carries +PDF/image MEDIA attachments deliver text+attachment on scheduled ticks but +text-only on manual ``hermes cron run ``. Same box, same token, same +scopes — the divergence is process context and error visibility, not +credentials. + +Three defects, each pinned here: + +1. STANDALONE LANE SWALLOWS WARNINGS: platform standalone senders (Slack, + Discord, ...) report per-file upload failures in ``result["warnings"]`` + while returning ``success: True``. ``_deliver_result`` only checks + ``result["error"]`` — an attachment failure vanishes: job marked ok, no + delivery error, text delivered, file gone. + +2. LIVE-ADAPTER LANE SWALLOWS MEDIA FAILURES: ``_send_media_via_adapter`` + logs failed/skipped sends at WARNING and returns None — the caller + cannot surface them into delivery_errors either. + +3. MEDIA-POLICY ENV BRIDGE IS GATEWAY-ONLY: ``gateway.strict`` / + ``media_delivery_allow_dirs`` / ``trust_recent_files`` from config.yaml + are exported to the env vars ``validate_media_delivery_path`` reads ONLY + in gateway startup (gateway/run.py). A CLI-process manual run filters + MEDIA paths under different policy than the gateway's scheduled tick. +""" + +import os +from pathlib import Path + +import pytest + +from cron.scheduler import _deliver_result, _send_media_via_adapter + + +@pytest.fixture() +def media_file(tmp_path): + p = tmp_path / "report.pdf" + p.write_bytes(b"%PDF-1.4 test") + return str(p) + + +@pytest.fixture() +def slack_job(): + return { + "id": "job-media-1", + "name": "media-parity", + "deliver": "slack:D0123456789", + "origin": {"platform": "slack", "chat_id": "D0123456789"}, + } + + +def _install_fake_slack_sender(monkeypatch, result_factory): + """Register a fake slack standalone sender through the platform registry.""" + calls = [] + + async def fake_sender(pconfig, chat_id, message, *, thread_id=None, + media_files=None, force_document=False, caption=None): + calls.append({ + "chat_id": chat_id, + "message": message, + "media_files": list(media_files or []), + "caption": caption, + }) + return result_factory(calls[-1]) + + import gateway.platform_registry as reg_mod + + entry = reg_mod.platform_registry.get("slack") + if entry is None: + # Populate the registry the same way tools/send_message_tool does. + import hermes_cli.plugins as hp_boot + + hp_boot.discover_plugins() + entry = reg_mod.platform_registry.get("slack") + if entry is None: + pytest.skip("slack platform entry not registered in this environment") + monkeypatch.setattr(entry, "standalone_sender_fn", fake_sender) + # Keep plugin discovery from replacing our fake mid-test. + import hermes_cli.plugins as hp + + monkeypatch.setattr(hp, "discover_plugins", lambda *a, **k: None) + return calls + + +@pytest.fixture() +def slack_platform_config(monkeypatch, tmp_path): + home = tmp_path / "hermes-home" + home.mkdir() + (home / "config.yaml").write_text( + "platforms:\n slack:\n enabled: true\n token: xoxb-test\n" + ) + monkeypatch.setenv("HERMES_HOME", str(home)) + # Config caches are process-global; clear them so the temp HERMES_HOME wins. + try: + from gateway import config as gwconfig + + for attr in ("_config_cache", "_CONFIG_CACHE", "_cached_config"): + if hasattr(gwconfig, attr): + monkeypatch.setattr(gwconfig, attr, None) + except Exception: + pass + return home + + +class TestStandaloneWarningsSurfaced: + """Defect 1: sender warnings must become delivery errors.""" + + def test_upload_warning_becomes_delivery_error( + self, monkeypatch, slack_platform_config, slack_job, media_file + ): + _install_fake_slack_sender( + monkeypatch, + lambda call: { + "success": True, + "platform": "slack", + "chat_id": call["chat_id"], + "message_id": "1.2", + "warnings": [ + f"Failed to send media {media_file}: Slack API error: missing_scope" + ], + }, + ) + err = _deliver_result( + slack_job, f"Report ready.\n\nMEDIA:{media_file}", adapters=None, loop=None + ) + assert err is not None, ( + "an attachment-upload failure reported via warnings must surface " + "as a delivery error, not vanish" + ) + assert "missing_scope" in err or "Failed to send media" in err + + def test_clean_delivery_still_returns_none( + self, monkeypatch, slack_platform_config, slack_job, media_file + ): + _install_fake_slack_sender( + monkeypatch, + lambda call: { + "success": True, + "platform": "slack", + "chat_id": call["chat_id"], + "message_id": "1.2", + }, + ) + err = _deliver_result( + slack_job, f"Report ready.\n\nMEDIA:{media_file}", adapters=None, loop=None + ) + assert err is None + + def test_media_actually_reaches_sender( + self, monkeypatch, slack_platform_config, slack_job, media_file + ): + calls = _install_fake_slack_sender( + monkeypatch, + lambda call: {"success": True, "chat_id": call["chat_id"], "message_id": "1.2"}, + ) + _deliver_result( + slack_job, f"Report ready.\n\nMEDIA:{media_file}", adapters=None, loop=None + ) + sent_media = [m for c in calls for m in c["media_files"]] + assert any(media_file in m[0] for m in sent_media) + + +class TestLiveAdapterMediaFailuresSurfaced: + """Defect 2: _send_media_via_adapter must report failures to its caller.""" + + def test_failed_media_send_returns_errors(self, media_file, slack_job): + import asyncio + + class FailingAdapter: + platform = None + + async def send_document(self, chat_id, file_path, metadata=None): + from types import SimpleNamespace + + return SimpleNamespace(success=False, error="upload rejected") + + loop = asyncio.new_event_loop() + try: + import threading + + t = threading.Thread(target=loop.run_forever, daemon=True) + t.start() + errors = _send_media_via_adapter( + FailingAdapter(), "D01", [(media_file, False)], None, loop, slack_job + ) + assert errors, ( + "a failed media send must be returned to the caller, not only logged" + ) + assert any("upload rejected" in e for e in errors) + finally: + loop.call_soon_threadsafe(loop.stop) + t.join(timeout=5) + loop.close() + + def test_dropped_unsafe_path_is_reported(self, slack_job): + import asyncio + + class NeverCalledAdapter: + platform = None + + async def send_document(self, chat_id, file_path, metadata=None): + raise AssertionError("should not be called for a dropped path") + + loop = asyncio.new_event_loop() + try: + import threading + + t = threading.Thread(target=loop.run_forever, daemon=True) + t.start() + errors = _send_media_via_adapter( + NeverCalledAdapter(), "D01", + [("/nonexistent/definitely-missing.pdf", False)], + None, loop, slack_job, + ) + assert errors, "a filtered-out MEDIA path must be reported, not silent" + finally: + loop.call_soon_threadsafe(loop.stop) + t.join(timeout=5) + loop.close() + + +class TestMediaPolicyEnvBridge: + """Defect 3: media-policy config must apply outside the gateway process.""" + + def test_bridge_helper_exists_and_applies_config(self, monkeypatch, tmp_path): + home = tmp_path / "hermes-home" + home.mkdir() + allow_dir = tmp_path / "reports" + allow_dir.mkdir() + (home / "config.yaml").write_text( + "gateway:\n" + " strict: true\n" + f" media_delivery_allow_dirs: [{str(allow_dir)!r}]\n" + " trust_recent_files: false\n" + ) + monkeypatch.setenv("HERMES_HOME", str(home)) + for var in ( + "HERMES_MEDIA_DELIVERY_STRICT", + "HERMES_MEDIA_ALLOW_DIRS", + "HERMES_MEDIA_TRUST_RECENT_FILES", + ): + monkeypatch.delenv(var, raising=False) + + from gateway.media_policy import apply_media_policy_env + + apply_media_policy_env() + + assert os.environ.get("HERMES_MEDIA_DELIVERY_STRICT") == "1" + assert str(allow_dir) in os.environ.get("HERMES_MEDIA_ALLOW_DIRS", "") + assert os.environ.get("HERMES_MEDIA_TRUST_RECENT_FILES") == "0" + + def test_standalone_filter_honors_bridged_allowlist(self, monkeypatch, tmp_path): + """End-to-end: strict-mode file inside allow_dirs passes validation + in a process that never ran gateway startup.""" + home = tmp_path / "hermes-home" + home.mkdir() + allow_dir = tmp_path / "reports" + allow_dir.mkdir() + media = allow_dir / "report.pdf" + media.write_bytes(b"%PDF-1.4 x") + # Make the file OLD so recency-trust cannot save it: only the + # allowlist can accept it, proving the bridge ran. + old = 1_600_000_000 + os.utime(media, (old, old)) + (home / "config.yaml").write_text( + "gateway:\n" + " strict: true\n" + f" media_delivery_allow_dirs: [{str(allow_dir)!r}]\n" + ) + monkeypatch.setenv("HERMES_HOME", str(home)) + for var in ("HERMES_MEDIA_DELIVERY_STRICT", "HERMES_MEDIA_ALLOW_DIRS"): + monkeypatch.delenv(var, raising=False) + + from gateway.media_policy import apply_media_policy_env + + apply_media_policy_env() + + from gateway.platforms.base import BasePlatformAdapter + + kept = BasePlatformAdapter.filter_media_delivery_paths([(str(media), False)]) + assert kept, ( + "with the policy bridged, an allowlisted file must survive strict " + "filtering in a non-gateway process" + ) + + def test_deliver_result_runs_bridge(self, monkeypatch, tmp_path, media_file): + """_deliver_result itself must apply the bridge before filtering. + + Realistic enterprise shape: HERMES_MEDIA_DELIVERY_STRICT=1 arrives via + .env (loaded by BOTH processes), but the allowlist lives in + config.yaml's gateway block — bridged only at gateway boot. Without + the bridge, a CLI manual run is strict WITHOUT the allowlist and + silently drops the attachment the scheduled (gateway) run delivers. + """ + home = tmp_path / "hermes-home" + home.mkdir() + media_dir = str(Path(media_file).parent) + (home / "config.yaml").write_text( + "platforms:\n slack:\n enabled: true\n token: xoxb-test\n" + "gateway:\n" + " strict: true\n" + f" media_delivery_allow_dirs: [{media_dir!r}]\n" + " trust_recent_files: false\n" + ) + monkeypatch.setenv("HERMES_HOME", str(home)) + # Strict comes from the shared .env in both processes... + monkeypatch.setenv("HERMES_MEDIA_DELIVERY_STRICT", "1") + monkeypatch.setenv("HERMES_MEDIA_TRUST_RECENT_FILES", "0") + # ...but the allowlist is config-only (gateway-boot bridge). + monkeypatch.delenv("HERMES_MEDIA_ALLOW_DIRS", raising=False) + old = 1_600_000_000 + os.utime(media_file, (old, old)) + + calls = _install_fake_slack_sender( + monkeypatch, + lambda call: {"success": True, "chat_id": call["chat_id"], "message_id": "1.2"}, + ) + job = { + "id": "job-media-3", + "name": "bridge", + "deliver": "slack:D0123456789", + "origin": {"platform": "slack", "chat_id": "D0123456789"}, + } + _deliver_result(job, f"Report.\n\nMEDIA:{media_file}", adapters=None, loop=None) + sent_media = [m for c in calls for m in c["media_files"]] + assert any(media_file in m[0] for m in sent_media), ( + "manual-run delivery must filter media under the same configured " + "policy as the gateway process (allowlisted file was dropped)" + )