Fix nine bugs across the executor, agents, and SQL layer

README.md

@@ -63,7 +63,9 @@ the lease lifetime (60 seconds by default), and
 unique `EXECUTOR_WORKER_ID` for a singleton worker deployment to let its next
 instance immediately recover commands left by the previous instance at
 startup; otherwise a process-unique identifier is generated and abandoned
-work is recovered when its lease expires.
+work is recovered when its lease expires. Each poll considers up to
+`COMMAND_CLAIM_CANDIDATES` sessions (10 by default) and takes the first whose
+per-session claim is free, so one busy session does not delay the others.
 Commands are parsed with `shlex.split` before execution, so quoting rules follow
 POSIX shells but features like glob expansion are not performed.
 
@@ -74,8 +76,13 @@ its PostgreSQL credentials, and the host outside `SHELL_ROOT` are trusted and
 must not be accessible to submitted commands. `run_subprocess` therefore builds
 a new environment containing only a fixed `PATH`, locale, the command account's
 identity variables, and values saved in `env_snapshot`. It never copies the
-worker environment, and rejects `DATABASE_URL` and `PG_CONN` even if a snapshot
-contains them.
+worker environment, and rejects reserved variables even if a snapshot contains
+them: the worker credentials `DATABASE_URL` and `PG_CONN`, every `LD_*`,
+`DYLD_*`, and `BASH_FUNC_*` name, and the glibc path overrides `GCONV_PATH`,
+`HOSTALIASES`, `LOCPATH`, `MALLOC_TRACE`, `NLSPATH`, and `RESOLV_HOST_CONF`.
+Those loader variables would otherwise run attacker-supplied code inside an
+allowlisted binary without executing anything new, defeating
+`EXECUTOR_ALLOWED_COMMANDS`. Matching is case-insensitive.
 
 Production deployments must create a dedicated, unprivileged OS account and
 configure an absolute-path executable allowlist. The allowlist is the required
@@ -106,7 +113,10 @@ command environment rather than inherited wholesale.
 You can run `cleanup_agent.py` periodically. Command retention applies only to
 terminal statuses: `done` and `failed` commands older than `CLEANUP_DAYS` are
 deleted, while recent terminal commands and all `pending` or `running` commands
-are retained. Use `replay_agent.py` for session replays. The optional
+are retained. Sessions untouched for `CLEANUP_DAYS` also have their `cwd` and
+`env` reset, so give the cleanup agent the same `SHELL_ROOT` as the executor
+(both default to `/home/sandbox`); a session reset to a `cwd` outside the
+executor's `SHELL_ROOT` cannot run anything. Use `replay_agent.py` for session replays. The optional
 `monitor_agent.py` emits usage metrics like
 command counts and average run time to stdout or CSV.
 

sql/html_fragments.sql

@@ -51,14 +51,14 @@ AS $$
         pg_shell_html_escape(result.status),
         pg_shell_html_escape(result.exit_code::TEXT)
       ),
-      '' ORDER BY result.row_number
+      '' ORDER BY result.id
     ),
     ''
   )::"text/html"
-  FROM (
-    SELECT row_number() OVER () AS row_number, output.*
-      FROM latest_output(p_user_id, p_session_id, p_since_id) AS output
-  ) AS result
+  -- Ordering by id keeps the fragment a chronological transcript. A bare
+  -- row_number() OVER () would instead depend on the order the planner
+  -- happened to emit latest_output's rows in.
+  FROM latest_output(p_user_id, p_session_id, p_since_id) AS result
 $$;
 
 CREATE OR REPLACE FUNCTION submit_command_html(

sql/install.sql

@@ -4,7 +4,8 @@
 \i sql/migrate_replay_provenance_retention.sql
 \i sql/migrate_command_leases.sql
 \i sql/submit_command.sql
-\i sql/latest_output.sql -- updated latest_output with optional since_id filter
+-- latest_output takes an optional since_id filter
+\i sql/latest_output.sql
 \i sql/html_fragments.sql
 \i sql/fork_session.sql
 \i sql/replay_session.sql

sql/latest_output.sql

@@ -18,18 +18,30 @@ RETURNS TABLE(
     completed_at TIMESTAMP
 ) LANGUAGE plpgsql AS $$
 BEGIN
+    -- Rows are always returned oldest-first so a caller can render them as a
+    -- transcript. The p_since_id = 0 backfill still selects the *newest* 20
+    -- commands (SPEC.md: "last N commands"), but the DESC + LIMIT is confined
+    -- to an inner query and the result is re-sorted ascending. Returning that
+    -- page newest-first would make the initial page load read bottom-to-top
+    -- and then flip as soon as incremental polling took over.
+    --
     -- Columns are qualified with the table alias so they are not confused
     -- with the identically named RETURNS TABLE output columns, which would
     -- otherwise raise "column reference \"id\" is ambiguous".
     IF p_since_id = 0 THEN
         RETURN QUERY
-        SELECT c.id, c.command, c.output, c.exit_code, c.status,
-               c.submitted_at, c.completed_at
-        FROM commands c
-        WHERE c.user_id = p_user_id AND c.session_id = p_session_id
-          AND c.id > p_since_id
-        ORDER BY c.id DESC
-        LIMIT 20;
+        SELECT recent.id, recent.command, recent.output, recent.exit_code,
+               recent.status, recent.submitted_at, recent.completed_at
+        FROM (
+            SELECT c.id, c.command, c.output, c.exit_code, c.status,
+                   c.submitted_at, c.completed_at
+            FROM commands c
+            WHERE c.user_id = p_user_id AND c.session_id = p_session_id
+              AND c.id > p_since_id
+            ORDER BY c.id DESC
+            LIMIT 20
+        ) AS recent
+        ORDER BY recent.id ASC;
     ELSE
         RETURN QUERY
         SELECT c.id, c.command, c.output, c.exit_code, c.status,

tests/test_cleanup_agent.py

@@ -107,6 +107,83 @@ def conn():
     conn.close()
 
 
+def test_main_retries_after_a_connection_failure(monkeypatch, caplog):
+    """A database outage must not silently end the retention daemon."""
+    attempts = []
+
+    def flaky_get_conn():
+        attempts.append(len(attempts))
+        if len(attempts) < 3:
+            raise RuntimeError("connection refused")
+        raise KeyboardInterrupt  # stand in for operator shutdown
+
+    sleeps = []
+    monkeypatch.setattr(cleanup_agent, "get_conn", flaky_get_conn)
+    monkeypatch.setattr(cleanup_agent.time, "sleep", sleeps.append)
+    monkeypatch.setattr(sys, "argv", ["cleanup_agent.py", "--interval", "3600"])
+
+    with pytest.raises(KeyboardInterrupt):
+        cleanup_agent.main()
+
+    assert len(attempts) == 3, "daemon gave up instead of retrying"
+    # Retries back off by RETRY_DELAY_SECONDS, not the full 3600s interval.
+    assert sleeps == [cleanup_agent.RETRY_DELAY_SECONDS] * 2
+
+
+def test_main_retries_after_a_failed_cleanup_run(monkeypatch):
+    """An error mid-run must not end the daemon either."""
+    closed = []
+
+    class FakeConn:
+        def close(self):
+            closed.append(True)
+
+    runs = []
+
+    def flaky_cleanup_once(conn, days):
+        runs.append(days)
+        if len(runs) < 2:
+            raise psycopg2.OperationalError("server closed the connection")
+        raise KeyboardInterrupt
+
+    monkeypatch.setattr(cleanup_agent, "get_conn", FakeConn)
+    monkeypatch.setattr(cleanup_agent, "cleanup_once", flaky_cleanup_once)
+    monkeypatch.setattr(cleanup_agent.time, "sleep", lambda seconds: None)
+    monkeypatch.setattr(sys, "argv", ["cleanup_agent.py"])
+
+    with pytest.raises(KeyboardInterrupt):
+        cleanup_agent.main()
+
+    assert len(runs) == 2
+    assert len(closed) == 2, "the connection must be closed on every path"
+
+
+@pytest.mark.parametrize(
+    "failure", [RuntimeError("no dsn"), psycopg2.OperationalError("boom")]
+)
+def test_main_once_reports_failure_through_exit_status(monkeypatch, failure):
+    """--once must exit non-zero so a supervisor or cron notices."""
+    if isinstance(failure, RuntimeError):
+        monkeypatch.setattr(
+            cleanup_agent, "get_conn", lambda: (_ for _ in ()).throw(failure)
+        )
+    else:
+
+        class FakeConn:
+            def close(self):
+                pass
+
+        monkeypatch.setattr(cleanup_agent, "get_conn", FakeConn)
+        monkeypatch.setattr(
+            cleanup_agent,
+            "cleanup_once",
+            lambda conn, days: (_ for _ in ()).throw(failure),
+        )
+    monkeypatch.setattr(sys, "argv", ["cleanup_agent.py", "--once"])
+
+    assert cleanup_agent.main() == 1
+
+
 def test_cleanup_once_deletes_only_old_terminal_commands(conn):
     uid_str = str(uuid.uuid4())
     with conn.cursor() as cur:
@@ -266,7 +343,37 @@ def test_cleanup_once_resets_env(conn):
         cur.execute("SELECT cwd, env FROM environments WHERE user_id = %s", (uid_str,))
         cwd, env = cur.fetchone()
 
-    assert cwd == "/home/sandbox"
+    assert cwd == cleanup_agent.DEFAULT_SHELL_ROOT
+    assert env == {}
+
+
+def test_cleanup_once_resets_env_to_configured_shell_root(conn, monkeypatch):
+    """A reset session must land inside the executor's SHELL_ROOT.
+
+    The executor hands environments.cwd straight to the subprocess, so
+    resetting to a hardcoded /home/sandbox in a deployment that configured a
+    different SHELL_ROOT leaves the session unable to run anything.
+    """
+    monkeypatch.setenv("SHELL_ROOT", "/home/pg-shell-command")
+    uid_str = str(uuid.uuid4())
+    with conn.cursor() as cur:
+        cur.execute(
+            "INSERT INTO users (id, username) VALUES (%s, %s)",
+            (uid_str, "shell-root-user"),
+        )
+        cur.execute(
+            "INSERT INTO environments (user_id, cwd, env, updated_at) "
+            "VALUES (%s, %s, %s::jsonb, now() - interval '100 days')",
+            (uid_str, "/home/pg-shell-command/deep", '{"k":1}'),
+        )
+
+    cleanup_once(conn, 90)
+
+    with conn.cursor() as cur:
+        cur.execute("SELECT cwd, env FROM environments WHERE user_id = %s", (uid_str,))
+        cwd, env = cur.fetchone()
+
+    assert cwd == "/home/pg-shell-command"
     assert env == {}
 
 

tests/test_executor_agent.py

@@ -123,14 +123,61 @@ def test_run_subprocess_does_not_expose_worker_secrets(monkeypatch, tmp_path):
     assert output.splitlines() == ["None", "None"]
 
 
-@pytest.mark.parametrize("reserved", ["DATABASE_URL", "PG_CONN"])
+@pytest.mark.parametrize(
+    "reserved",
+    [
+        "DATABASE_URL",
+        "PG_CONN",
+        # Dynamic loader controls: these run attacker-supplied code inside an
+        # allowlisted binary, so they would defeat EXECUTOR_ALLOWED_COMMANDS.
+        "LD_PRELOAD",
+        "LD_LIBRARY_PATH",
+        "LD_AUDIT",
+        "DYLD_INSERT_LIBRARIES",
+        "BASH_FUNC_ls%%",
+        # glibc path overrides with the same effect.
+        "GCONV_PATH",
+        "LOCPATH",
+        "NLSPATH",
+        "HOSTALIASES",
+        "RESOLV_HOST_CONF",
+        "MALLOC_TRACE",
+        # Names are matched case-insensitively; the loader does the same.
+        "ld_preload",
+    ],
+)
 def test_run_subprocess_rejects_reserved_snapshot_variables(
     monkeypatch, tmp_path, reserved
 ):
     with pytest.raises(ValueError, match="reserved environment variable"):
         run_subprocess("python3 -c 'pass'", str(tmp_path), {reserved: "secret"})
 
 
+def test_run_subprocess_rejects_ld_preload_before_launching(monkeypatch, tmp_path):
+    """LD_PRELOAD must be refused rather than handed to the dynamic loader."""
+
+    def fail_popen(*args, **kwargs):  # pragma: no cover - must not be reached
+        pytest.fail("subprocess launched with an attacker-controlled loader")
+
+    monkeypatch.setattr(workers.executor_agent.subprocess, "Popen", fail_popen)
+
+    with pytest.raises(ValueError, match="LD_PRELOAD"):
+        run_subprocess(
+            "python3 -c 'pass'",
+            str(tmp_path),
+            {"LD_PRELOAD": str(tmp_path / "evil.so")},
+        )
+
+
+def test_run_subprocess_allows_unreserved_snapshot_variables(tmp_path):
+    """The prefix rule must not swallow ordinary variables."""
+    cmd = "python3 -c 'import os;print(os.environ[\"LDAP_URL\"])'"
+    exit_code, output = run_subprocess(cmd, str(tmp_path), {"LDAP_URL": "ldap://x"})
+
+    assert exit_code == 0
+    assert "ldap://x" in output
+
+
 def test_handle_command_uses_combined_output(monkeypatch):
     captured = {}