diff --git a/.env.example b/.env.example index 5382c23c77..c85bb043a6 100644 --- a/.env.example +++ b/.env.example @@ -154,6 +154,17 @@ SEARXNG_INSTANCE=http://localhost:8080 # if you intentionally want scheduled scripts to run remotely. # ODYSSEUS_SCRIPT_HOST=localhost +# Vault-backed DRM action guard. DRM run_local/run_script/ssh_command tasks +# fail closed unless this checkout contains a current heartbeat whose enabled +# actions and SHA-256 match automation/config/odysseus_actions.yaml. +# ODYSSEUS_DRM_VAULT_ROOT=/home/agent/projects/vault-remote-upkeep +# ODYSSEUS_DRM_HEARTBEAT_MAX_AGE_MINUTES=1800 + +# Scheduled LLM/research jobs emit Activity liveness every five minutes and +# time out after one hour by default. Both values remain bounded in code. +# ODYSSEUS_TASK_LIVENESS_SECONDS=300 +# ODYSSEUS_MODEL_TASK_TIMEOUT_SECONDS=3600 + # Chat / agent attachment size cap in bytes (default: 10 MB). # Raise this for local installs that need larger PDFs or text documents. # Example: 52428800 = 50 MB. diff --git a/src/drm_runtime_guard.py b/src/drm_runtime_guard.py new file mode 100644 index 0000000000..e350083232 --- /dev/null +++ b/src/drm_runtime_guard.py @@ -0,0 +1,90 @@ +"""Fail-closed heartbeat guard for vault-backed DRM action tasks.""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +_ACTION_RE = re.compile(r"^ ([A-Za-z0-9][A-Za-z0-9_-]*):\s*(?:#.*)?$") +_ENABLED_RE = re.compile(r"^ enabled:\s*(true|false)\b") + + +def _parse_time(value: Any) -> datetime | None: + try: + parsed = datetime.fromisoformat(str(value or "").replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _allowlist(path: Path) -> dict[str, bool]: + actions: dict[str, bool] = {} + in_actions = False + current = "" + for raw in path.read_text(encoding="utf-8").splitlines(): + line = raw.rstrip() + if line.startswith("actions:"): + in_actions = True + current = "" + continue + if in_actions and line and not line.startswith(" "): + break + if not in_actions: + continue + match = _ACTION_RE.match(line) + if match: + current = match.group(1) + actions[current] = False + continue + enabled = _ENABLED_RE.match(line) + if current and enabled: + actions[current] = enabled.group(1) == "true" + return actions + + +def _allowlist_hash(path: Path) -> str: + text = path.read_text(encoding="utf-8").replace("\r\n", "\n").replace("\r", "\n") + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def check_drm_runtime_guard(task_name: str, *, now: datetime | None = None) -> tuple[bool, str]: + """Return whether a DRM action may run under the configured vault heartbeat.""" + if not str(task_name or "").strip().lower().startswith("drm "): + return True, "not-a-drm-task" + vault_root = Path(os.environ.get("ODYSSEUS_DRM_VAULT_ROOT", "")).expanduser() + if not str(vault_root) or str(vault_root) == ".": + return False, "ODYSSEUS_DRM_VAULT_ROOT is not configured" + allowlist_path = vault_root / "automation/config/odysseus_actions.yaml" + heartbeat_dir = vault_root / "automation/review/routine-reports/odysseus-heartbeat" + if not allowlist_path.is_file(): + return False, f"DRM allowlist missing: {allowlist_path}" + candidates = sorted(heartbeat_dir.glob("*.heartbeat.json")) if heartbeat_dir.is_dir() else [] + if not candidates: + return False, f"DRM heartbeat missing: {heartbeat_dir}" + try: + payload = json.loads(candidates[-1].read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + return False, f"DRM heartbeat unreadable: {exc}" + if payload.get("report_kind") != "odysseus-heartbeat": + return False, "DRM heartbeat report_kind is invalid" + generated = _parse_time(payload.get("generated_at")) + if generated is None: + return False, "DRM heartbeat generated_at is invalid" + current = now or datetime.now(timezone.utc) + max_age_minutes = max(1, int(os.environ.get("ODYSSEUS_DRM_HEARTBEAT_MAX_AGE_MINUTES", "1800"))) + if (current.astimezone(timezone.utc) - generated).total_seconds() / 60 > max_age_minutes: + return False, "DRM heartbeat is stale" + if payload.get("actions_yaml_sha256") != _allowlist_hash(allowlist_path): + return False, "DRM heartbeat allowlist hash mismatch" + actions = _allowlist(allowlist_path) + enabled = payload.get("enabled_actions") + if not isinstance(enabled, list) or any(not actions.get(str(name), False) for name in enabled): + return False, "DRM heartbeat enabled_actions exceed the committed allowlist" + return True, "current heartbeat and allowlist verified" diff --git a/src/task_scheduler.py b/src/task_scheduler.py index 4b71ff8f64..84e636bf01 100644 --- a/src/task_scheduler.py +++ b/src/task_scheduler.py @@ -1,8 +1,10 @@ """Background scheduler for ScheduledTask execution.""" import asyncio +import contextlib import json import logging +import os import re import time import uuid @@ -12,6 +14,14 @@ logger = logging.getLogger(__name__) +def _model_task_timeout_seconds() -> int: + try: + value = int(os.environ.get("ODYSSEUS_MODEL_TASK_TIMEOUT_SECONDS", "3600")) + except ValueError: + value = 3600 + return max(300, value) + + def _utcnow() -> datetime: """Return naive UTC for task DB fields without using deprecated APIs.""" return datetime.now(timezone.utc).replace(tzinfo=None) @@ -274,6 +284,19 @@ def _set_run_progress(self, run_id: str, message: str): except Exception: logger.debug("Task progress update failed", exc_info=True) + async def _emit_run_liveness(self, run_id: str, task_name: str, interval_seconds: float | None = None): + """Persist a bounded progress heartbeat until the owning task cancels this coroutine.""" + if interval_seconds is None: + try: + interval_seconds = float(os.environ.get("ODYSSEUS_TASK_LIVENESS_SECONDS", "300")) + except ValueError: + interval_seconds = 300.0 + interval_seconds = max(0.01, interval_seconds) + while True: + await asyncio.sleep(interval_seconds) + stamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + self._set_run_progress(run_id, f"Running {task_name}; liveness {stamp}") + def _mark_run_aborted(self, task_id: str, run_id: str | None = None, message: str = "Stopped by user") -> bool: """Mark an active run as aborted. Used by stop/cancel paths.""" try: @@ -667,6 +690,7 @@ async def _execute_task_locked(self, task_id: str, run_id: str, *, release_execu from core.database import SessionLocal, ScheduledTask, TaskRun db = SessionLocal() + liveness_task = None try: task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first() if not task or task.status != "active": @@ -702,6 +726,8 @@ async def _execute_task_locked(self, task_id: str, run_id: str, *, release_execu db.add(run) db.commit() + liveness_task = asyncio.create_task(self._emit_run_liveness(run_id, task.name or task.id)) + task_type = task.task_type or "llm" from src.builtin_actions import TaskDeferred, TaskNoop @@ -718,12 +744,18 @@ async def _execute_task_locked(self, task_id: str, run_id: str, *, release_execu if not success: run.error = result elif task_type == "research": - result = await self._execute_research_task(task, db) + result = await asyncio.wait_for( + self._execute_research_task(task, db), + timeout=_model_task_timeout_seconds(), + ) run.status = "success" run.result = result else: # LLM task — use agent loop for tool access - result = await self._execute_llm_task(task, db) + result = await asyncio.wait_for( + self._execute_llm_task(task, db), + timeout=_model_task_timeout_seconds(), + ) run.status = "success" run.result = result # Record which model actually ran (resolved inside the executor). @@ -935,6 +967,10 @@ async def _execute_task_locked(self, task_id: str, run_id: str, *, release_execu except Exception: logger.exception("Task %s error-path failed unexpectedly", task_id) finally: + if liveness_task is not None: + liveness_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await liveness_task db.close() handle = self._task_handles.get(task_id) if handle is asyncio.current_task(): @@ -1015,6 +1051,12 @@ async def _execute_action(self, task, run_id: str | None = None) -> tuple: if not action_fn: return f"Unknown action: {task.action}", False + if task.action in ("run_local", "run_script", "ssh_command"): + from src.drm_runtime_guard import check_drm_runtime_guard + allowed, reason = check_drm_runtime_guard(task.name) + if not allowed: + return f"DRM runtime guard refused actuation: {reason}", False + from src.builtin_actions import TaskNoop try: # Pass task prompt as script/command for ssh_command/run_script actions. diff --git a/tests/test_drm_runtime_guard.py b/tests/test_drm_runtime_guard.py new file mode 100644 index 0000000000..eff6e82cc2 --- /dev/null +++ b/tests/test_drm_runtime_guard.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import asyncio +import hashlib +import json +from datetime import datetime, timezone + +import pytest + +from src.drm_runtime_guard import check_drm_runtime_guard +from src.task_scheduler import TaskScheduler, _model_task_timeout_seconds + + +ALLOWLIST = "version: 1\nactions:\n validate:\n enabled: true\n task-transition:\n enabled: false\n" +NOW = datetime(2026, 6, 22, 18, 0, tzinfo=timezone.utc) + + +def _vault(tmp_path, *, enabled=None, generated_at="2026-06-22T17:00:00Z"): + config = tmp_path / "automation/config" + heartbeats = tmp_path / "automation/review/routine-reports/odysseus-heartbeat" + config.mkdir(parents=True) + heartbeats.mkdir(parents=True) + (config / "odysseus_actions.yaml").write_text(ALLOWLIST, encoding="utf-8") + digest = hashlib.sha256(ALLOWLIST.encode("utf-8")).hexdigest() + payload = { + "report_kind": "odysseus-heartbeat", + "generated_at": generated_at, + "actions_yaml_sha256": digest, + "enabled_actions": ["validate"] if enabled is None else enabled, + } + (heartbeats / "2026-06-22.heartbeat.json").write_text(json.dumps(payload), encoding="utf-8") + return tmp_path + + +def test_guard_accepts_current_hash_matching_heartbeat(tmp_path, monkeypatch): + monkeypatch.setenv("ODYSSEUS_DRM_VAULT_ROOT", str(_vault(tmp_path))) + assert check_drm_runtime_guard("DRM Validate", now=NOW) == ( + True, + "current heartbeat and allowlist verified", + ) + + +@pytest.mark.parametrize( + "enabled,generated_at,reason", + [ + (["task-transition"], "2026-06-22T17:00:00Z", "enabled_actions exceed"), + (["validate"], "2026-06-20T00:00:00Z", "heartbeat is stale"), + ], +) +def test_guard_fails_closed_on_drift_or_staleness(tmp_path, monkeypatch, enabled, generated_at, reason): + monkeypatch.setenv("ODYSSEUS_DRM_VAULT_ROOT", str(_vault(tmp_path, enabled=enabled, generated_at=generated_at))) + allowed, message = check_drm_runtime_guard("DRM Validate", now=NOW) + assert not allowed + assert reason in message + + +def test_guard_requires_configuration_for_drm_only(monkeypatch): + monkeypatch.delenv("ODYSSEUS_DRM_VAULT_ROOT", raising=False) + assert check_drm_runtime_guard("Calendar Classify Events", now=NOW)[0] + assert not check_drm_runtime_guard("DRM Validate", now=NOW)[0] + + +def test_model_timeout_defaults_to_one_hour(monkeypatch): + monkeypatch.delenv("ODYSSEUS_MODEL_TASK_TIMEOUT_SECONDS", raising=False) + assert _model_task_timeout_seconds() == 3600 + monkeypatch.setenv("ODYSSEUS_MODEL_TASK_TIMEOUT_SECONDS", "900") + assert _model_task_timeout_seconds() == 900 + + +@pytest.mark.asyncio +async def test_liveness_emits_periodic_progress(): + scheduler = TaskScheduler.__new__(TaskScheduler) + messages = [] + scheduler._set_run_progress = lambda run_id, message: messages.append((run_id, message)) + task = asyncio.create_task(scheduler._emit_run_liveness("run-1", "Long task", interval_seconds=0.01)) + await asyncio.sleep(0.025) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert messages + assert messages[0][0] == "run-1" + assert "liveness" in messages[0][1]