security: 17 hardening fixes across Phases 1-4

.gitignore

@@ -19,6 +19,11 @@ data/*
 !data/spreadsheetbench_id_split/**
 !data/alfworld_path_split/
 !data/alfworld_path_split/**
+# ─── Internal analysis docs (private, not for open-source release) ───────────
+doc-internals/
+
+# ─── Build / local output dirs ───────────────────────────────────────────────
+out/
 outputs/
 logs/
 external/

docs/sleep/README.md

@@ -17,7 +17,7 @@ normal agent requests.
 One "night":
 
 ```
-harvest Claude Code / Codex / Cursor transcripts → mine recurring tasks → replay in isolated model calls
+harvest Claude Code / Codex / Cursor transcripts → mine recurring tasks → replay via the configured backend (isolation varies by backend; mock/handoff make no network calls)
    → consolidate (reflect → bounded edit → GATE on real held-out tasks)
    → stage proposal → (you) adopt
 ```

scripts/train.py

@@ -383,8 +383,24 @@ def parse_args() -> argparse.Namespace:
 
 def load_config(args: argparse.Namespace) -> dict:
     """Load config with _base_ inheritance, then apply CLI overrides."""
+    import warnings
     from skillopt.config import load_config as _load, flatten_config, is_structured
 
+    # F08: Warn when API keys are supplied on the CLI — prefer env vars or managed identity.
+    for _cli_key in (
+        "azure_api_key", "azure_openai_api_key",
+        "optimizer_azure_openai_api_key", "target_azure_openai_api_key",
+        "minimax_api_key",
+    ):
+        if getattr(args, _cli_key, None):
+            warnings.warn(
+                f"--{_cli_key} is deprecated: provide credentials via the "
+                f"AZURE_OPENAI_API_KEY environment variable or set "
+                f"--azure_openai_auth_mode=managed_identity instead.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
     cfg = _load(args.config, overrides=args.cfg_options)
     structured = is_structured(cfg)
 

skillopt/envs/spreadsheetbench/codegen_agent.py

@@ -260,17 +260,34 @@ def _build_codex_driver() -> str:
     return (
         "import pathlib\n"
         "import re\n"
-        "import sys\n"
-        "import traceback\n\n"
+        "import subprocess\n"
+        "import sys\n\n"
         'INPUT_PATH = "input.xlsx"\n'
         'OUTPUT_PATH = "output.xlsx"\n'
         "code = pathlib.Path('solution.py').read_text(encoding='utf-8')\n"
         "code = re.sub(r'^\\s*(INPUT_PATH|OUTPUT_PATH)\\s*=\\s*.+$', '', code, flags=re.MULTILINE)\n"
-        "globals_dict = {'__name__': '__main__', 'INPUT_PATH': INPUT_PATH, 'OUTPUT_PATH': OUTPUT_PATH}\n"
-        "try:\n"
-        "    exec(compile(code, 'solution.py', 'exec'), globals_dict, globals_dict)\n"
-        "except Exception:\n"
-        "    traceback.print_exc()\n"
+        "# Write patched code to a temporary file and run it in a clean subprocess\n"
+        "# (avoids exec/compile of untrusted LLM-generated code in the current process).\n"
+        "_patched = pathlib.Path('_driver_runner.py')\n"
+        "_patched.write_text(\n"
+        "    f'INPUT_PATH = {INPUT_PATH!r}\\nOUTPUT_PATH = {OUTPUT_PATH!r}\\n' + code,\n"
+        "    encoding='utf-8',\n"
+        ")\n"
+        "import os as _os\n"
+        "_safe_env = {\n"
+        "    'PATH': _os.environ.get('PATH', '/usr/bin:/bin'),\n"
+        "    'HOME': str(pathlib.Path('_driver_runner.py').parent.resolve()),\n"
+        "}\n"
+        "if _os.name == 'nt':\n"
+        "    _safe_env['SYSTEMROOT'] = _os.environ.get('SYSTEMROOT', '')\n"
+        "    _safe_env['TEMP'] = _safe_env['HOME']\n"
+        "    _safe_env['TMP'] = _safe_env['HOME']\n"
+        "_safe_env = {k: v for k, v in _safe_env.items() if v}\n"
+        "_res = subprocess.run([sys.executable, str(_patched)], capture_output=True, text=True, env=_safe_env)\n"
+        "if _res.returncode != 0:\n"
+        "    import traceback as _tb\n"
+        "    print(_res.stdout, end='')\n"
+        "    print(_res.stderr, end='')\n"
         "    sys.exit(2)\n"
     )
 

skillopt/envs/spreadsheetbench/executor.py

@@ -46,12 +46,27 @@ def run_generated_code(code: str, input_path: str, output_path: str, timeout: in
     with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as f:
         f.write(script)
         tmp = f.name
+    # Build a minimal environment so the generated code cannot read API keys,
+    # cloud credentials, or other secrets from the current process environment.
+    import platform as _platform
+    _safe_env: dict[str, str] = {
+        "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
+        "HOME": os.path.dirname(output_path),
+        "TMPDIR": tempfile.gettempdir(),
+    }
+    if _platform.system() == "Windows":
+        _safe_env["SYSTEMROOT"] = os.environ.get("SYSTEMROOT", "")
+        _safe_env["TEMP"] = tempfile.gettempdir()
+        _safe_env["TMP"] = tempfile.gettempdir()
+    # Drop empty entries (env dict values must be non-empty strings)
+    _safe_env = {k: v for k, v in _safe_env.items() if v}
     try:
         proc = subprocess.run(
             [sys.executable, tmp],
             capture_output=True,
             text=True,
             timeout=timeout if timeout and timeout > 0 else None,
+            env=_safe_env,
         )
         if proc.returncode != 0:
             return False, (proc.stdout + "\n" + proc.stderr).strip()

skillopt/envs/spreadsheetbench/react_agent.py

@@ -9,6 +9,7 @@
 
 import json
 import os
+import shlex
 import subprocess
 
 from skillopt.model import chat_target_messages
@@ -248,13 +249,28 @@ def _auto_verify(work_dir: str) -> str:
         return f"\n\n[AUTO-VERIFY] Could not inspect output: {e}"
 
 
+# Commands that the ReAct agent is allowed to run (benchmark only needs Python).
+_CMD_ALLOW = {"python", "python3"}
+
+
 # ── Bash execution ────────────────────────────────────────────────────────────
 
 def _run_bash(cmd: str, work_dir: str, timeout: int = 60) -> str:
     try:
+        parts = shlex.split(cmd)
+        if not parts:
+            return "[error: empty command]"
+        exe_name = os.path.basename(parts[0]).lower()
+        # Strip .exe suffix on Windows (python.exe → python)
+        exe_stem = exe_name.split(".")[0]
+        if exe_stem not in _CMD_ALLOW:
+            return (
+                f"[blocked: '{parts[0]}' not in allow-list {sorted(_CMD_ALLOW)}; "
+                "use Python to manipulate spreadsheets]"
+            )
         proc = subprocess.run(
-            cmd,
-            shell=True,
+            parts,
+            shell=False,
             capture_output=True,
             text=True,
             timeout=timeout,

skillopt_sleep/backend.py

@@ -1082,7 +1082,7 @@ def _call(self, prompt: str, *, max_tokens: int = 1024) -> str:
             "--stream", "off",
             "--no-color",
             "--log-level", "none",
-            "--allow-all-tools",
+            "--allowed-tools", os.environ.get("COPILOT_ALLOWED_TOOLS", "Bash"),
             "-C", clean_cwd,
         ]
         if not self.full_env:
@@ -1197,7 +1197,7 @@ def attempt_with_tools(self, task, skill, memory, tools):
                 "--stream", "off",
                 "--no-color",
                 "--log-level", "none",
-                "--allow-all-tools",
+                "--allowed-tools", os.environ.get("COPILOT_ALLOWED_TOOLS", "Bash"),
                 "-C", work,
             ]
             if not self.full_env:

skillopt_sleep/cycle.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import os
+import re
 import shutil
 import sys
 from dataclasses import dataclass
@@ -30,6 +31,56 @@
 from skillopt_sleep.types import SessionDigest, SleepReport, TaskRecord
 
 
+# ── Auto-adopt safety gate ────────────────────────────────────────────────────────────
+# Block auto-adoption if any proposed edit content contains high-risk patterns.
+_DANGEROUS_CONTENT = re.compile(
+    r"(?i)(api[_\-]?key|credential|exfil|leak|bypass|ignore\s*instruction"
+    r"|x-bypass|token\s*=\s*[A-Za-z0-9])"
+)
+
+
+def _safe_edits(edits: list) -> bool:
+    """Return False (and log) if any edit contains a dangerous-looking pattern."""
+    import logging as _logging
+    _log = _logging.getLogger(__name__)
+    for e in edits:
+        content = getattr(e, "content", "") or ""
+        if _DANGEROUS_CONTENT.search(content):
+            _log.warning(
+                "auto_adopt BLOCKED: dangerous pattern detected in edit content: %s",
+                content[:120],
+            )
+            return False
+    return True
+
+
+def _make_model_key(cfg: SleepConfig) -> str:
+    """Stable string identifying the backend+model combination for this cycle."""
+    return "{}::{}".format(
+        cfg.get("backend", "mock"),
+        cfg.get("model", ""),
+    )
+
+
+def _check_model_change(cfg: SleepConfig, state: SleepState) -> None:
+    """Warn when the backend/model has changed since the last night.
+
+    Skill text is backend-specific; adopting edits from a different model's
+    reflections into a new model's skill file can cause regressions.
+    This is advisory only — the cycle continues either way.
+    """
+    current_key = _make_model_key(cfg)
+    prior_key = state.last_model_key
+    if prior_key and prior_key != current_key:
+        print(
+            f"[sleep] WARNING: model changed since last night "
+            f"(was {prior_key!r}, now {current_key!r}). "
+            "Learned skill text may not transfer cleanly. "
+            "Consider starting from a fresh skill document.",
+            file=sys.stderr,
+        )
+
+
 @dataclass
 class CycleOutcome:
     report: SleepReport
@@ -127,6 +178,7 @@ def run_sleep_cycle(
     """
     cfg = cfg or load_config()
     state = SleepState.load(cfg.state_path)
+    _check_model_change(cfg, state)  # F16: warn if model changed between nights
     night = state.begin_night(clock)
     project = _project_paths(cfg)
     started = _now_iso(clock)
@@ -409,10 +461,12 @@ def run_sleep_cycle(
             "baseline": result.baseline_score, "candidate": result.candidate_score,
             "n_tasks": len(tasks), "staging": staging_dir,
         })
+        state.set_last_model_key(_make_model_key(cfg))  # F16: track model for next night
         # ── 6. adopt (opt-in) ────────────────────────────────────────────
         if cfg.get("auto_adopt") and result.accepted:
-            adopted_paths = adopt_staging(staging_dir)
-            adopted = bool(adopted_paths)
+            if _safe_edits(result.applied_edits):
+                adopted_paths = adopt_staging(staging_dir)
+                adopted = bool(adopted_paths)
         state.save()
 
     if ev is not None:

skillopt_sleep/evidence.py

@@ -27,6 +27,8 @@
 """
 from __future__ import annotations
 
+import hashlib
+import hmac as _hmac_mod
 import json
 import os
 import threading
@@ -40,6 +42,11 @@ def _now_iso() -> str:
     return time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime())
 
 
+def _os_user() -> str:
+    """Return the current OS username for audit purposes (best-effort)."""
+    return os.environ.get("USERNAME") or os.environ.get("USER") or "unknown"
+
+
 class EvidenceLog:
     """Append-only, thread-safe JSONL logger for one sleep night."""
 
@@ -71,13 +78,24 @@ def _clean(self, value: Any) -> Any:
 
     # ── the one write path ────────────────────────────────────────────────
     def log(self, stage: str, event: str, **data: Any) -> None:
-        record = {"ts": _now_iso(), "stage": stage, "event": event}
+        record = {"ts": _now_iso(), "stage": stage, "event": event,
+                  "os_user": _os_user()}
         record.update(self._clean(data))
         with self._lock:
             self._seq += 1
             record["seq"] = self._seq
             try:
                 line = json.dumps(record, ensure_ascii=False, default=str)
+                # Optional per-record HMAC for tamper detection (F15).
+                # Set SKILLOPT_EVIDENCE_HMAC_KEY to enable.
+                _hmac_key = os.environ.get("SKILLOPT_EVIDENCE_HMAC_KEY", "")
+                if _hmac_key:
+                    mac = _hmac_mod.new(
+                        _hmac_key.encode(), line.encode(), hashlib.sha256
+                    ).hexdigest()
+                    # Append the MAC as a separate audit line so the JSONL
+                    # structure of the payload line is unchanged.
+                    line = line + "\n" + json.dumps({"_audit_mac": mac, "seq": self._seq})
                 with open(self.path, "a", encoding="utf-8") as f:
                     f.write(line + "\n")
             except Exception:

skillopt_sleep/gate.py

@@ -8,6 +8,7 @@
 """
 from __future__ import annotations
 
+import re
 from dataclasses import dataclass
 
 
@@ -21,6 +22,9 @@ class GateResult:
     best_step: int
 
 
+# \u2500\u2500 Content safety (F14) \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n# Reject candidate skills that contain high-risk content before they can ever\n# be accepted by the gate, regardless of score.\n_DANGEROUS = re.compile(\n    r\"(?i)(api[_\\-]?key|credential|exfil|leak|bypass\\s*security\"\n    r\"|ignore\\s*instruction|x-bypass|token\\s*=\\s*[A-Za-z0-9])\"\n)\n\n\ndef _content_safe(candidate_skill: str) -> bool:\n    \"\"\"Return False if the candidate skill text contains dangerous patterns.\"\"\"\n    return not bool(_DANGEROUS.search(candidate_skill))
+
+
 def select_gate_score(hard: float, soft: float, metric: str = "hard",
                       mixed_weight: float = 0.5) -> float:
     """Project (hard, soft) onto a single comparison metric."""
@@ -39,6 +43,10 @@ def evaluate_gate(candidate_skill: str, cand_hard: float, current_skill: str,
                   best_step: int, global_step: int, *, cand_soft: float = 0.0,
                   metric: str = "hard", mixed_weight: float = 0.5) -> GateResult:
     """Pure gate decision: compare candidate score to current/best."""
+    # Content safety: reject before score comparison if skill text is unsafe.
+    if not _content_safe(candidate_skill):
+        return GateResult("reject", current_skill, current_score,
+                          best_skill, best_score, best_step)
     cand_score = select_gate_score(cand_hard, cand_soft, metric, mixed_weight)
     if cand_score > current_score:
         if cand_score > best_score:

skillopt_sleep/harvest.py

@@ -15,6 +15,8 @@
 """
 from __future__ import annotations
 
+import hashlib
+import hmac as _hmac_mod
 import json
 import os
 from datetime import datetime
@@ -23,6 +25,9 @@
 from skillopt_sleep.types import SessionDigest
 
 
+# \u2500\u2500 Optional HMAC session integrity (F09) \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n# When SKILLOPT_SESSION_HMAC_KEY is set, each session file is signed on first\n# harvest (a .sig companion is written) and verified on subsequent harvests.\n# Tampered or injected session files are skipped.  Opt-in: if the env var is\n# unset (the default), signing and verification are both no-ops.\n\ndef _hmac_key() -> bytes:\n    \"\"\"Return the configured HMAC key, or b'' if none configured.\"\"\"\n    raw = os.environ.get(\"SKILLOPT_SESSION_HMAC_KEY\", \"\")\n    return raw.encode() if raw else b\"\"\n\n\ndef _sign_session(path: str) -> None:\n    \"\"\"Write a .sig file for *path* if a key is configured and no sig exists.\"\"\"\n    key = _hmac_key()\n    if not key:\n        return\n    sig_path = path + \".sig\"\n    if os.path.exists(sig_path):\n        return\n    try:\n        with open(path, \"rb\") as f:\n            content = f.read()\n        digest = _hmac_mod.new(key, content, hashlib.sha256).hexdigest()\n        with open(sig_path, \"w\", encoding=\"utf-8\") as f:\n            f.write(digest + \"\\n\")\n    except OSError:\n        pass  # read-only fs or permission issue — skip silently\n\n\ndef _verify_session(path: str) -> bool:\n    \"\"\"Return True if HMAC verification passes (or no key is configured).\"\"\"\n    key = _hmac_key()\n    if not key:\n        return True\n    sig_path = path + \".sig\"\n    if not os.path.exists(sig_path):\n        # No sig yet — sign on this first harvest and accept it.\n        _sign_session(path)\n        return True\n    try:\n        with open(sig_path, encoding=\"utf-8\") as f:\n            expected = f.read().strip()\n        with open(path, \"rb\") as f:\n            content = f.read()\n        actual = _hmac_mod.new(key, content, hashlib.sha256).hexdigest()\n        return _hmac_mod.compare_digest(expected, actual)\n    except OSError:\n        return False  # unreadable sig or session file \u2014 skip
+
+
 # Heuristic phrases that signal the user (dis)approving of prior output.
 # English-only by default. Users whose sessions are in another language can add
 # their own phrases via the SKILLOPT_SLEEP_NEG_FEEDBACK / _POS_FEEDBACK env vars
@@ -322,6 +327,11 @@ def harvest(
     paths.sort(key=lambda p: os.path.getmtime(p), reverse=True)
 
     for p in paths:
+        if not _verify_session(p):
+            import logging as _logging
+            _logging.getLogger(__name__).warning(
+                "harvest: HMAC verification failed for %s — skipping", p)
+            continue
         d = digest_transcript(p)
         if d is None:
             continue