From 3130d95cc5027003e129934efc72bc9f6b4f91a3 Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Wed, 10 Jun 2026 21:15:30 -0400 Subject: [PATCH 1/8] =?UTF-8?q?feat:=20sandbox=20primitives=20=E2=80=94=20?= =?UTF-8?q?Bind,=20SandboxPolicy,=20is=5Favailable?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Core sandbox infrastructure for isolating harness sessions. Bind defines read-only/writable path bindings, SandboxPolicy specifies filesystem and network isolation modes, and is_available() is the single source of truth for bubblewrap availability across the application. Co-Authored-By: Claude Opus 4.8 --- src/hqt/sandbox.py | 33 +++++++++++++++++++++++++++++++++ tests/test_sandbox.py | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+) create mode 100644 src/hqt/sandbox.py create mode 100644 tests/test_sandbox.py diff --git a/src/hqt/sandbox.py b/src/hqt/sandbox.py new file mode 100644 index 0000000..f1d5219 --- /dev/null +++ b/src/hqt/sandbox.py @@ -0,0 +1,33 @@ +import os # noqa: F401 +import shutil +import sys +from dataclasses import dataclass +from pathlib import Path + + +@dataclass(frozen=True) +class Bind: + """A host path to expose inside the sandbox (read-only unless writable).""" + + src: Path + writable: bool = False + + +@dataclass(frozen=True) +class SandboxPolicy: + """Per-session sandbox settings. + + fs: "rw" binds the project dir writable; "ro" binds it read-only. + net: True shares the host network namespace; False = no connectivity. + """ + + fs: str + net: bool + + +def is_available() -> bool: + """True when bubblewrap can be used: Linux with `bwrap` on PATH. + + The single source of truth for both `hqt doctor` and the New Session dialog. + """ + return sys.platform.startswith("linux") and shutil.which("bwrap") is not None diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py new file mode 100644 index 0000000..52884f5 --- /dev/null +++ b/tests/test_sandbox.py @@ -0,0 +1,34 @@ +from pathlib import Path + +from hqt import sandbox +from hqt.sandbox import Bind, SandboxPolicy + + +def test_bind_defaults_readonly(): + b = Bind(Path("/home/u/.claude")) + assert b.src == Path("/home/u/.claude") + assert b.writable is False + + +def test_sandbox_policy_fields(): + p = SandboxPolicy(fs="rw", net=True) + assert p.fs == "rw" + assert p.net is True + + +def test_is_available_true_on_linux_with_bwrap(monkeypatch): + monkeypatch.setattr(sandbox.sys, "platform", "linux") + monkeypatch.setattr(sandbox.shutil, "which", lambda name: "/usr/bin/bwrap") + assert sandbox.is_available() is True + + +def test_is_available_false_without_bwrap(monkeypatch): + monkeypatch.setattr(sandbox.sys, "platform", "linux") + monkeypatch.setattr(sandbox.shutil, "which", lambda name: None) + assert sandbox.is_available() is False + + +def test_is_available_false_off_linux(monkeypatch): + monkeypatch.setattr(sandbox.sys, "platform", "darwin") + monkeypatch.setattr(sandbox.shutil, "which", lambda name: "/usr/bin/bwrap") + assert sandbox.is_available() is False From 9db79c68f4edd109eb57c924308f812345d5b384 Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Wed, 10 Jun 2026 21:16:44 -0400 Subject: [PATCH 2/8] feat: sandbox.wrap builds bwrap argv from policy and binds Co-Authored-By: Claude Opus 4.8 --- src/hqt/sandbox.py | 45 +++++++++++++++++++++++++++- tests/test_sandbox.py | 70 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 114 insertions(+), 1 deletion(-) diff --git a/src/hqt/sandbox.py b/src/hqt/sandbox.py index f1d5219..1763d12 100644 --- a/src/hqt/sandbox.py +++ b/src/hqt/sandbox.py @@ -1,4 +1,4 @@ -import os # noqa: F401 +import os import shutil import sys from dataclasses import dataclass @@ -31,3 +31,46 @@ def is_available() -> bool: The single source of truth for both `hqt doctor` and the New Session dialog. """ return sys.platform.startswith("linux") and shutil.which("bwrap") is not None + + +# System paths bound read-only so binaries and shared libraries resolve. +_SYSTEM_RO_PATHS = ["/usr", "/bin", "/sbin", "/lib", "/lib64", "/etc", "/opt"] + +# Environment variables passed through into the jail. +_PASSTHROUGH_ENV = ["PATH", "HOME", "TERM", "LANG", "LC_ALL", "USER", "SHELL"] + + +def wrap( + command: list[str], + cwd: Path, + policy: SandboxPolicy, + binds: list[Bind], +) -> list[str]: + """Build the `bwrap … -- ` argv for a sandboxed harness. + + Minimal-allowlist model: $HOME and other user data are NOT exposed. System + dirs are read-only; the project dir (cwd) and each existing harness bind are + bound explicitly. Network is all-or-nothing: shared only when policy.net. + Binds whose source does not exist are skipped (bwrap would otherwise error). + """ + args: list[str] = ["bwrap", "--die-with-parent", "--unshare-all"] + if policy.net: + args.append("--share-net") + for path in _SYSTEM_RO_PATHS: + if Path(path).exists(): + args += ["--ro-bind", path, path] + args += ["--proc", "/proc", "--dev", "/dev", "--tmpfs", "/tmp"] + for bind in binds: + if not bind.src.exists(): + continue + flag = "--bind" if bind.writable else "--ro-bind" + args += [flag, str(bind.src), str(bind.src)] + cwd_flag = "--bind" if policy.fs == "rw" else "--ro-bind" + args += [cwd_flag, str(cwd), str(cwd)] + for name in _PASSTHROUGH_ENV: + value = os.environ.get(name) + if value is not None: + args += ["--setenv", name, value] + args.append("--") + args += command + return args diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py index 52884f5..b39e650 100644 --- a/tests/test_sandbox.py +++ b/tests/test_sandbox.py @@ -32,3 +32,73 @@ def test_is_available_false_off_linux(monkeypatch): monkeypatch.setattr(sandbox.sys, "platform", "darwin") monkeypatch.setattr(sandbox.shutil, "which", lambda name: "/usr/bin/bwrap") assert sandbox.is_available() is False + + +def _split(args, flag): + """Return list of (src, dst) pairs that follow each occurrence of flag.""" + out = [] + for i, a in enumerate(args): + if a == flag: + out.append((args[i + 1], args[i + 2])) + return out + + +def test_wrap_command_after_double_dash(tmp_path): + cmd = ["claude", "--session-id", "x"] + args = sandbox.wrap(cmd, tmp_path, SandboxPolicy(fs="rw", net=True), []) + assert args[0] == "bwrap" + assert "--" in args + assert args[args.index("--") + 1 :] == cmd + + +def test_wrap_base_flags(tmp_path): + args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), []) + assert "--die-with-parent" in args + assert "--unshare-all" in args + assert "--proc" in args and "--dev" in args + + +def test_wrap_net_on_shares_net(tmp_path): + args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), []) + assert "--share-net" in args + + +def test_wrap_net_off_does_not_share(tmp_path): + args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=False), []) + assert "--share-net" not in args + + +def test_wrap_cwd_rw_is_bind(tmp_path): + args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), []) + assert (str(tmp_path), str(tmp_path)) in _split(args, "--bind") + + +def test_wrap_cwd_ro_is_ro_bind(tmp_path): + args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="ro", net=True), []) + assert (str(tmp_path), str(tmp_path)) in _split(args, "--ro-bind") + + +def test_wrap_writable_bind_spliced(tmp_path): + cred = tmp_path / "cred" + cred.mkdir() + args = sandbox.wrap( + ["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [Bind(cred, writable=True)] + ) + assert (str(cred), str(cred)) in _split(args, "--bind") + + +def test_wrap_readonly_bind_spliced(tmp_path): + cred = tmp_path / "cred" + cred.mkdir() + args = sandbox.wrap( + ["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [Bind(cred)] + ) + assert (str(cred), str(cred)) in _split(args, "--ro-bind") + + +def test_wrap_skips_missing_binds(tmp_path): + missing = tmp_path / "nope" + args = sandbox.wrap( + ["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [Bind(missing)] + ) + assert str(missing) not in args From 4c3638504559c328a186c166bc4610575ff2f471 Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Wed, 10 Jun 2026 21:20:50 -0400 Subject: [PATCH 3/8] feat: configurators declare sandbox skip-permission flags and binds Add sandbox_skip_permission_flags class attr and sandbox_binds() method to HarnessConfigurator ABC; thread sandboxed: bool = False through all build_spawn_config / build_resume_config signatures. Claude and Codex append their respective skip-permission flags when sandboxed=True; Kiro and Generic accept the param but add no flags. Co-Authored-By: Claude Opus 4.8 --- src/hqt/harnesses/base.py | 20 ++++++- src/hqt/harnesses/configurators/claude.py | 21 +++++++- src/hqt/harnesses/configurators/codex.py | 21 +++++++- src/hqt/harnesses/configurators/generic.py | 12 ++++- src/hqt/harnesses/configurators/kiro.py | 16 +++++- tests/test_harnesses.py | 62 ++++++++++++++++++++-- 6 files changed, 137 insertions(+), 15 deletions(-) diff --git a/src/hqt/harnesses/base.py b/src/hqt/harnesses/base.py index ba03a19..6f49ab9 100644 --- a/src/hqt/harnesses/base.py +++ b/src/hqt/harnesses/base.py @@ -2,6 +2,8 @@ from abc import ABC, abstractmethod from dataclasses import dataclass, field from pathlib import Path +from hqt.sandbox import Bind + @dataclass class SpawnConfig: @@ -14,20 +16,34 @@ class HarnessConfigurator(ABC): name: str display_name: str captures_session_id: bool = False + # Flags appended to the harness command when the session is sandboxed. + sandbox_skip_permission_flags: list[str] = [] @abstractmethod def generate_session_id(self, hqt_session_id: int) -> str: ... @abstractmethod def build_spawn_config( - self, project_path: Path, harness_session_id: str, model: str | None = None + self, + project_path: Path, + harness_session_id: str, + model: str | None = None, + sandboxed: bool = False, ) -> SpawnConfig: ... @abstractmethod def build_resume_config( - self, project_path: Path, harness_session_id: str, model: str | None = None + self, + project_path: Path, + harness_session_id: str, + model: str | None = None, + sandboxed: bool = False, ) -> SpawnConfig: ... + def sandbox_binds(self) -> list[Bind]: + """Host paths this harness needs inside the jail (credentials, config).""" + return [] + def capture_session_id(self, project_path: Path, since: float) -> str | None: return None diff --git a/src/hqt/harnesses/configurators/claude.py b/src/hqt/harnesses/configurators/claude.py index 3adbded..d9e08a5 100644 --- a/src/hqt/harnesses/configurators/claude.py +++ b/src/hqt/harnesses/configurators/claude.py @@ -2,6 +2,7 @@ import uuid from pathlib import Path from hqt.harnesses.base import HarnessConfigurator, SpawnConfig +from hqt.sandbox import Bind NAMESPACE = uuid.UUID("a1b2c3d4-e5f6-7890-abcd-ef1234567890") @@ -9,6 +10,7 @@ NAMESPACE = uuid.UUID("a1b2c3d4-e5f6-7890-abcd-ef1234567890") class ClaudeConfigurator(HarnessConfigurator): name = "claude" display_name = "Claude Code" + sandbox_skip_permission_flags = ["--dangerously-skip-permissions"] # Markers for parse_status — conservative, case-insensitive substring checks. # None returned when none match; caller falls back to activity-based status. @@ -21,20 +23,35 @@ class ClaudeConfigurator(HarnessConfigurator): def generate_session_id(self, hqt_session_id: int) -> str: return str(uuid.uuid5(NAMESPACE, str(hqt_session_id))) + def sandbox_binds(self) -> list[Bind]: + return [Bind(Path.home() / ".claude", writable=True)] + def build_spawn_config( - self, project_path: Path, harness_session_id: str, model: str | None = None + self, + project_path: Path, + harness_session_id: str, + model: str | None = None, + sandboxed: bool = False, ) -> SpawnConfig: cmd = ["claude", "--session-id", harness_session_id] if model: cmd += ["--model", model] + if sandboxed: + cmd += self.sandbox_skip_permission_flags return SpawnConfig(command=cmd, cwd=project_path) def build_resume_config( - self, project_path: Path, harness_session_id: str, model: str | None = None + self, + project_path: Path, + harness_session_id: str, + model: str | None = None, + sandboxed: bool = False, ) -> SpawnConfig: cmd = ["claude", "--resume", harness_session_id] if model: cmd += ["--model", model] + if sandboxed: + cmd += self.sandbox_skip_permission_flags return SpawnConfig(command=cmd, cwd=project_path) def parse_status(self, pane_text: str) -> str | None: diff --git a/src/hqt/harnesses/configurators/codex.py b/src/hqt/harnesses/configurators/codex.py index 10107f4..8555fa0 100644 --- a/src/hqt/harnesses/configurators/codex.py +++ b/src/hqt/harnesses/configurators/codex.py @@ -4,6 +4,7 @@ from datetime import datetime from pathlib import Path from hqt.harnesses.base import HarnessConfigurator, SpawnConfig +from hqt.sandbox import Bind log = logging.getLogger(__name__) @@ -12,6 +13,7 @@ class CodexConfigurator(HarnessConfigurator): name = "codex" display_name = "Codex" captures_session_id = True + sandbox_skip_permission_flags = ["--dangerously-bypass-approvals-and-sandbox"] # Markers for parse_status — conservative, case-insensitive substring checks. _WORKING_MARKERS = ("esc to interrupt",) @@ -19,20 +21,35 @@ class CodexConfigurator(HarnessConfigurator): def generate_session_id(self, hqt_session_id: int) -> str: return str(hqt_session_id) + def sandbox_binds(self) -> list[Bind]: + return [Bind(Path.home() / ".codex", writable=True)] + def build_spawn_config( - self, project_path: Path, harness_session_id: str, model: str | None = None + self, + project_path: Path, + harness_session_id: str, + model: str | None = None, + sandboxed: bool = False, ) -> SpawnConfig: cmd = ["codex"] if model: cmd += ["-m", model] + if sandboxed: + cmd += self.sandbox_skip_permission_flags return SpawnConfig(command=cmd, cwd=project_path) def build_resume_config( - self, project_path: Path, harness_session_id: str, model: str | None = None + self, + project_path: Path, + harness_session_id: str, + model: str | None = None, + sandboxed: bool = False, ) -> SpawnConfig: cmd = ["codex", "resume", harness_session_id] if model: cmd += ["-m", model] + if sandboxed: + cmd += self.sandbox_skip_permission_flags return SpawnConfig(command=cmd, cwd=project_path) def parse_status(self, pane_text: str) -> str | None: diff --git a/src/hqt/harnesses/configurators/generic.py b/src/hqt/harnesses/configurators/generic.py index 505ea61..28b48e4 100644 --- a/src/hqt/harnesses/configurators/generic.py +++ b/src/hqt/harnesses/configurators/generic.py @@ -14,11 +14,19 @@ class GenericConfigurator(HarnessConfigurator): return str(hqt_session_id) def build_spawn_config( - self, project_path: Path, harness_session_id: str, model: str | None = None + self, + project_path: Path, + harness_session_id: str, + model: str | None = None, + sandboxed: bool = False, ) -> SpawnConfig: return SpawnConfig(command=self._command, cwd=project_path) def build_resume_config( - self, project_path: Path, harness_session_id: str, model: str | None = None + self, + project_path: Path, + harness_session_id: str, + model: str | None = None, + sandboxed: bool = False, ) -> SpawnConfig: return SpawnConfig(command=self._command, cwd=project_path) diff --git a/src/hqt/harnesses/configurators/kiro.py b/src/hqt/harnesses/configurators/kiro.py index c652aff..2b30a87 100644 --- a/src/hqt/harnesses/configurators/kiro.py +++ b/src/hqt/harnesses/configurators/kiro.py @@ -1,6 +1,7 @@ from pathlib import Path from hqt.harnesses.base import HarnessConfigurator, SpawnConfig +from hqt.sandbox import Bind class KiroConfigurator(HarnessConfigurator): @@ -10,12 +11,23 @@ class KiroConfigurator(HarnessConfigurator): def generate_session_id(self, hqt_session_id: int) -> str: return str(hqt_session_id) + def sandbox_binds(self) -> list[Bind]: + return [Bind(Path.home() / ".kiro", writable=True)] + def build_spawn_config( - self, project_path: Path, harness_session_id: str, model: str | None = None + self, + project_path: Path, + harness_session_id: str, + model: str | None = None, + sandboxed: bool = False, ) -> SpawnConfig: return SpawnConfig(command=["kiro-cli", "chat"], cwd=project_path) def build_resume_config( - self, project_path: Path, harness_session_id: str, model: str | None = None + self, + project_path: Path, + harness_session_id: str, + model: str | None = None, + sandboxed: bool = False, ) -> SpawnConfig: return SpawnConfig(command=["kiro-cli", "chat"], cwd=project_path) diff --git a/tests/test_harnesses.py b/tests/test_harnesses.py index 2e164a6..eae807e 100644 --- a/tests/test_harnesses.py +++ b/tests/test_harnesses.py @@ -4,8 +4,11 @@ import time from pathlib import Path from unittest.mock import patch -from hqt.harnesses.configurators.codex import CodexConfigurator +from hqt.harnesses.base import HarnessConfigurator from hqt.harnesses.configurators.claude import ClaudeConfigurator +from hqt.harnesses.configurators.codex import CodexConfigurator +from hqt.harnesses.configurators.generic import GenericConfigurator +from hqt.harnesses.configurators.kiro import KiroConfigurator from hqt.harnesses.registry import discover_harnesses @@ -129,8 +132,6 @@ def test_codex_captures_session_id_flag(): def test_base_captures_session_id_flag_false(): """HarnessConfigurator base class default must be False.""" - from hqt.harnesses.base import HarnessConfigurator - assert HarnessConfigurator.captures_session_id is False @@ -287,8 +288,6 @@ CODEX_IDLE_SCREEN = """\ def test_base_parse_status_returns_none(): """Base HarnessConfigurator.parse_status always returns None.""" - from hqt.harnesses.configurators.kiro import KiroConfigurator - k = KiroConfigurator() assert k.parse_status("anything") is None assert k.parse_status("") is None @@ -346,3 +345,56 @@ def test_codex_parse_status_none_for_empty(): """CodexConfigurator: empty text → None.""" c = CodexConfigurator() assert c.parse_status("") is None + + +# --------------------------------------------------------------------------- +# Task 3: sandbox skip-permission flags and binds +# --------------------------------------------------------------------------- + + +def test_base_skip_flags_default_empty(): + assert HarnessConfigurator.sandbox_skip_permission_flags == [] + + +def test_base_sandbox_binds_default_empty(): + # GenericConfigurator does not override sandbox_binds; verifies the base default. + assert GenericConfigurator(["mytool"]).sandbox_binds() == [] + + +def test_claude_skip_flag_added_when_sandboxed(): + c = ClaudeConfigurator() + sid = c.generate_session_id(1) + cfg = c.build_spawn_config(Path("/tmp"), sid, sandboxed=True) + assert "--dangerously-skip-permissions" in cfg.command + + +def test_claude_skip_flag_absent_when_not_sandboxed(): + c = ClaudeConfigurator() + sid = c.generate_session_id(1) + cfg = c.build_spawn_config(Path("/tmp"), sid, sandboxed=False) + assert "--dangerously-skip-permissions" not in cfg.command + + +def test_claude_resume_skip_flag_when_sandboxed(): + c = ClaudeConfigurator() + sid = c.generate_session_id(1) + cfg = c.build_resume_config(Path("/tmp"), sid, sandboxed=True) + assert "--dangerously-skip-permissions" in cfg.command + + +def test_claude_sandbox_binds_includes_dot_claude(): + c = ClaudeConfigurator() + binds = c.sandbox_binds() + assert any(b.src == Path.home() / ".claude" and b.writable for b in binds) + + +def test_codex_skip_flag_added_when_sandboxed(): + c = CodexConfigurator() + cfg = c.build_spawn_config(Path("/tmp"), "1", sandboxed=True) + assert "--dangerously-bypass-approvals-and-sandbox" in cfg.command + + +def test_generic_sandboxed_is_noop_flagwise(): + g = GenericConfigurator(["mytool"]) + cfg = g.build_spawn_config(Path("/tmp"), "1", sandboxed=True) + assert cfg.command == ["mytool"] From 080ddd9364b88a0023dd7f039439829418e70544 Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Wed, 10 Jun 2026 21:23:45 -0400 Subject: [PATCH 4/8] feat: add Session.sandbox_json column and migration Adds a nullable sandbox_json TEXT column to the sessions table via a new version-3 migration. Updates LATEST_VERSION to 3 and adjusts existing migration tests to account for the new column and version number. Co-Authored-By: Claude Opus 4.8 --- src/hqt/db/migrations.py | 5 ++++ src/hqt/db/models.py | 1 + tests/test_db.py | 63 ++++++++++++++++++++++++++++++++++------ 3 files changed, 60 insertions(+), 9 deletions(-) diff --git a/src/hqt/db/migrations.py b/src/hqt/db/migrations.py index d234f6c..2f42478 100644 --- a/src/hqt/db/migrations.py +++ b/src/hqt/db/migrations.py @@ -18,8 +18,13 @@ def _migrate_v2(conn: Connection) -> None: conn.exec_driver_sql("ALTER TABLE sessions ADD COLUMN worktree_branch TEXT") +def _migrate_v3(conn: Connection) -> None: + conn.exec_driver_sql("ALTER TABLE sessions ADD COLUMN sandbox_json TEXT") + + MIGRATIONS: list[tuple[int, Callable[[Connection], None]]] = [ (2, _migrate_v2), + (3, _migrate_v3), ] BASELINE_VERSION = 1 diff --git a/src/hqt/db/models.py b/src/hqt/db/models.py index d90f89e..448726e 100644 --- a/src/hqt/db/models.py +++ b/src/hqt/db/models.py @@ -38,6 +38,7 @@ class Session(Base): tmux_session_name: Mapped[str] = mapped_column(unique=True) harness_session_id: Mapped[str | None] = mapped_column(default=None) model: Mapped[str | None] = mapped_column(default=None) + sandbox_json: Mapped[str | None] = mapped_column(default=None) worktree_path: Mapped[str | None] = mapped_column(default=None) worktree_branch: Mapped[str | None] = mapped_column(default=None) created_at: Mapped[datetime] = mapped_column(insert_default=func.now()) diff --git a/tests/test_db.py b/tests/test_db.py index d28d876..349f1c1 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -97,6 +97,7 @@ def test_unversioned_existing_db_stamped_latest(tmp_path): with engine.begin() as conn: conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN worktree_path") conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN worktree_branch") + conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN sandbox_json") migrate(engine) assert _user_version(engine) == migrations.LATEST_VERSION @@ -113,7 +114,7 @@ def test_db_from_newer_hqt_raises(tmp_path): def test_pending_migrations_applied_in_order_once(tmp_path, monkeypatch): settings = _tmp_settings(tmp_path) - ensure_db(settings) # stamps migrations.LATEST_VERSION (currently 2) + ensure_db(settings) # stamps migrations.LATEST_VERSION (currently 3) applied: list[int] = [] # Use versions above current LATEST_VERSION so none have been applied yet. monkeypatch.setattr( @@ -121,17 +122,17 @@ def test_pending_migrations_applied_in_order_once(tmp_path, monkeypatch): "MIGRATIONS", [ *migrations.MIGRATIONS, - (3, lambda conn: applied.append(3)), (4, lambda conn: applied.append(4)), + (5, lambda conn: applied.append(5)), ], ) - monkeypatch.setattr(migrations, "LATEST_VERSION", 4) + monkeypatch.setattr(migrations, "LATEST_VERSION", 5) engine = get_engine(settings) migrate(engine) - assert applied == [3, 4] - assert _user_version(engine) == 4 + assert applied == [4, 5] + assert _user_version(engine) == 5 migrate(engine) # re-run is a no-op - assert applied == [3, 4] + assert applied == [4, 5] def _session_column_names(engine) -> list[str]: @@ -147,11 +148,11 @@ def test_fresh_db_has_worktree_columns(tmp_path): cols = _session_column_names(engine) assert "worktree_path" in cols assert "worktree_branch" in cols - assert _user_version(engine) == 2 + assert _user_version(engine) == 3 def test_v1_db_upgraded_to_v2(tmp_path): - # Build a v1 database: create all tables, drop the two new columns, stamp v1. + # Build a v1 database: create all tables, drop v2+ columns, stamp v1. settings = _tmp_settings(tmp_path) settings.db_path.parent.mkdir(parents=True, exist_ok=True) engine = get_engine(settings) @@ -159,6 +160,7 @@ def test_v1_db_upgraded_to_v2(tmp_path): with engine.begin() as conn: conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN worktree_path") conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN worktree_branch") + conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN sandbox_json") conn.exec_driver_sql("PRAGMA user_version = 1") migrate(engine) @@ -166,7 +168,8 @@ def test_v1_db_upgraded_to_v2(tmp_path): cols = _session_column_names(engine) assert "worktree_path" in cols assert "worktree_branch" in cols - assert _user_version(engine) == 2 + assert "sandbox_json" in cols + assert _user_version(engine) == 3 def test_session_worktree_fields_default_none(tmp_path): @@ -215,3 +218,45 @@ def test_rows_stay_readable_after_session_closes(tmp_path): session.add(p) session.commit() assert p.name == "x" # would raise DetachedInstanceError without the flag + + +def test_latest_version_is_three(): + assert migrations.LATEST_VERSION == 3 + + +def test_migration_adds_sandbox_json_column(tmp_path): + # Simulate a baseline DB whose sessions table predates the column. + settings = _tmp_settings(tmp_path) + settings.db_path.parent.mkdir(parents=True, exist_ok=True) + engine = get_engine(settings) + with engine.begin() as conn: + conn.exec_driver_sql( + "CREATE TABLE sessions (" + "id INTEGER PRIMARY KEY, project_id INTEGER, harness_id INTEGER, " + "tmux_session_name TEXT)" + ) + conn.exec_driver_sql("PRAGMA user_version = 2") + migrate(engine) + cols = [c["name"] for c in inspect(engine).get_columns("sessions")] + assert "sandbox_json" in cols + assert _user_version(engine) == 3 + + +def test_session_round_trips_sandbox_json(tmp_path): + settings = _tmp_settings(tmp_path) + ensure_db(settings) + factory = get_session_factory(get_engine(settings)) + with factory() as session: + p = Project(name="proj", path="/tmp/proj-sandbox") + h = Harness(name="claude-sandbox") + session.add_all([p, h]) + session.commit() + s = Session( + project_id=p.id, + harness_id=h.id, + tmux_session_name="hqt-9", + sandbox_json='{"fs": "rw", "net": true}', + ) + session.add(s) + session.commit() + assert session.get(Session, s.id).sandbox_json == '{"fs": "rw", "net": true}' From e54cbb96b66222e77dd1b38336d55270fc6c6d8b Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Wed, 10 Jun 2026 21:28:22 -0400 Subject: [PATCH 5/8] feat: wrap sandboxed sessions in bwrap on spawn and resume - Import `hqt.sandbox` module and `SandboxPolicy` in service.py - Add `_sandbox_policy` helper to decode persisted JSON policy from a session row - Add `_wrap_command` helper that calls `sandbox.is_available` / `sandbox.wrap` and raises `ServiceError` when bwrap is unavailable for a sandboxed session - Add `sandbox: SandboxPolicy | None = None` parameter to `create_session`; persist as `sandbox_json`, pass `sandboxed=` to `build_spawn_config`, wrap command before spawning - Update `_get_resume_config` to pass `sandboxed=` to `build_resume_config` and wrap the resulting command - Update rung-2 path in `_respawn_with_fallback` to pass `sandboxed=` and wrap the fresh-spawn command - Add three new tests covering: sandboxed create wraps command, unsandboxed create does not wrap, and ServiceError on missing bwrap Co-Authored-By: Claude Opus 4.8 --- src/hqt/sessions/service.py | 59 +++++++++++++++++++++++++++++---- tests/test_sessions.py | 66 ++++++++++++++++++++++++++++++++++++- 2 files changed, 117 insertions(+), 8 deletions(-) diff --git a/src/hqt/sessions/service.py b/src/hqt/sessions/service.py index 7d18a9c..9e9595a 100644 --- a/src/hqt/sessions/service.py +++ b/src/hqt/sessions/service.py @@ -1,5 +1,6 @@ import asyncio import contextlib +import json import logging import time from collections.abc import Mapping @@ -10,11 +11,13 @@ from pathlib import Path from sqlalchemy.orm import Session as DBSession from sqlalchemy.orm import selectinload, sessionmaker +from hqt import sandbox from hqt.db.models import Harness, Project, Session from hqt.errors import ServiceError from hqt.git import worktree from hqt.git.worktree import WorktreeState from hqt.harnesses.base import HarnessConfigurator, SpawnConfig +from hqt.sandbox import SandboxPolicy from hqt.status import window_label from hqt.tmux.manager import SpawnRequest, TmuxManager from hqt.tmux.runner import WindowInfo @@ -159,6 +162,33 @@ class SessionService: return created_at.timestamp() return 0.0 + def _sandbox_policy(self, sess: Session) -> SandboxPolicy | None: + """Decode the persisted sandbox policy, or None if the session is plain.""" + if not sess.sandbox_json: + return None + data = json.loads(sess.sandbox_json) + return SandboxPolicy(fs=data["fs"], net=data["net"]) + + def _wrap_command( + self, + configurator: HarnessConfigurator, + command: list[str], + cwd: Path, + policy: SandboxPolicy | None, + ) -> list[str]: + """Wrap a harness command in bwrap when a policy is set. + + Raises ServiceError if a sandbox is required but bwrap is unavailable — + a backstop; the dialog already disables the toggle in that case. + """ + if policy is None: + return command + if not sandbox.is_available(): + raise ServiceError( + "bubblewrap (bwrap) is not available; cannot start a sandboxed session" + ) + return sandbox.wrap(command, cwd, policy, configurator.sandbox_binds()) + def _maybe_capture_missing_session_id(self, db: DBSession, sess: Session) -> None: configurator = self.harnesses.get(sess.harness.name) if configurator is None or not configurator.captures_session_id: @@ -192,6 +222,7 @@ class SessionService: nickname: str | None = None, model: str | None = None, worktree_branch: str | None = None, + sandbox: SandboxPolicy | None = None, ) -> "CreateSessionResult": """Create a new session row, spawn the harness window, and return a CreateSessionResult with the Session row and spawn outcome. @@ -238,13 +269,21 @@ class SessionService: sess.tmux_session_name = f"hqt-{sess.id}" harness_session_id = configurator.generate_session_id(sess.id) sess.harness_session_id = harness_session_id + sess.sandbox_json = ( + json.dumps({"fs": sandbox.fs, "net": sandbox.net}) + if sandbox is not None + else None + ) spawn_cfg = configurator.build_spawn_config( - run_path, harness_session_id, model + run_path, harness_session_id, model, sandboxed=sandbox is not None + ) + command = self._wrap_command( + configurator, spawn_cfg.command, spawn_cfg.cwd, sandbox ) log.info( "Spawning window %s: cmd=%s cwd=%s", sess.tmux_session_name, - spawn_cfg.command, + command, spawn_cfg.cwd, ) async with self._maybe_capture_lock(configurator.captures_session_id): @@ -252,7 +291,7 @@ class SessionService: spawn_result = await self.tmux.spawn( SpawnRequest( window_name=sess.tmux_session_name, - command=spawn_cfg.command, + command=command, cwd=str(spawn_cfg.cwd), env=spawn_cfg.env, ) @@ -353,13 +392,15 @@ class SessionService: harness_session_id = ( sess.harness_session_id or configurator.generate_session_id(sess.id) ) + policy = self._sandbox_policy(sess) spawn_cfg = configurator.build_spawn_config( - run_path, harness_session_id, sess.model + run_path, harness_session_id, sess.model, sandboxed=policy is not None ) + command = self._wrap_command(configurator, spawn_cfg.command, spawn_cfg.cwd, policy) async with self._maybe_capture_lock(configurator.captures_session_id): since = time.time() result = await self.tmux.respawn_verified( - window_name, spawn_cfg.command, str(spawn_cfg.cwd), env=spawn_cfg.env + window_name, command, str(spawn_cfg.cwd), env=spawn_cfg.env ) if not result.ok: log.error( @@ -552,6 +593,10 @@ class SessionService: harness_session_id = ( sess.harness_session_id or configurator.generate_session_id(sess.id) ) - return configurator.build_resume_config( - self._session_path(db, sess), harness_session_id, sess.model + policy = self._sandbox_policy(sess) + cfg = configurator.build_resume_config( + self._session_path(db, sess), harness_session_id, sess.model, + sandboxed=policy is not None, ) + command = self._wrap_command(configurator, cfg.command, cfg.cwd, policy) + return SpawnConfig(command=command, env=cfg.env, cwd=cfg.cwd) diff --git a/tests/test_sessions.py b/tests/test_sessions.py index 361b0f7..4b24fd9 100644 --- a/tests/test_sessions.py +++ b/tests/test_sessions.py @@ -8,6 +8,7 @@ from sqlalchemy.pool import StaticPool from hqt.db.models import Base, Harness, Project from hqt.errors import ServiceError +from hqt.sandbox import SandboxPolicy from hqt.sessions.service import SessionInfo, SessionService from hqt.tmux.manager import SpawnResult, TmuxManager @@ -816,7 +817,7 @@ async def test_attach_late_capture_updates_id_before_resume( assert ok is True harness.build_resume_config.assert_called_with( - Path("/tmp/myproj"), "late-codex-id", result.session.model + Path("/tmp/myproj"), "late-codex-id", result.session.model, sandboxed=False ) from hqt.db.models import Session as DBSession @@ -1518,3 +1519,66 @@ async def test_create_session_no_lock_for_non_capturing_harness(service, db, tmu # The `service` fixture's harness has captures_session_id = False, so the # spawn must NOT be serialized under the capture lock. assert observed["locked_during_spawn"] is False + + +# --------------------------------------------------------------------------- +# Task 5: sandbox wrapping on spawn and resume +# --------------------------------------------------------------------------- + + +@pytest.fixture +def sandbox_harness(): + h = MagicMock() + h.captures_session_id = False + h.generate_session_id.return_value = "sess-1" + h.build_spawn_config.return_value = MagicMock( + command=["claude", "--dangerously-skip-permissions"], + env={}, + cwd=Path("/tmp/myproj"), + ) + h.sandbox_binds.return_value = [] + h.parse_status = MagicMock(return_value=None) + return {"claude-code": h} + + +@pytest.fixture +def sandbox_service(factory, db, tmux, sandbox_harness): + return SessionService(factory=factory, tmux=tmux, harnesses=sandbox_harness) + + +@pytest.mark.asyncio +async def test_sandboxed_create_wraps_command(sandbox_service, db, tmux, sandbox_harness): + with patch("hqt.sessions.service.sandbox.is_available", return_value=True), patch( + "hqt.sessions.service.sandbox.wrap", return_value=["bwrap", "--", "claude"] + ) as mock_wrap: + result = await sandbox_service.create_session( + project_id=1, + harness_name="claude-code", + sandbox=SandboxPolicy(fs="rw", net=True), + ) + mock_wrap.assert_called_once() + sandbox_harness["claude-code"].build_spawn_config.assert_called_once() + assert sandbox_harness["claude-code"].build_spawn_config.call_args.kwargs[ + "sandboxed" + ] is True + assert tmux.spawn.call_args.args[0].command == ["bwrap", "--", "claude"] + assert result.session.sandbox_json == '{"fs": "rw", "net": true}' + + +@pytest.mark.asyncio +async def test_unsandboxed_create_does_not_wrap(service, db, tmux, harnesses): + with patch("hqt.sessions.service.sandbox.wrap") as mock_wrap: + result = await service.create_session(project_id=1, harness_name="claude-code") + mock_wrap.assert_not_called() + assert result.session.sandbox_json is None + + +@pytest.mark.asyncio +async def test_sandboxed_create_raises_when_bwrap_missing(sandbox_service, db): + with patch("hqt.sessions.service.sandbox.is_available", return_value=False): + with pytest.raises(ServiceError, match="bubblewrap"): + await sandbox_service.create_session( + project_id=1, + harness_name="claude-code", + sandbox=SandboxPolicy(fs="rw", net=True), + ) From f6ed53ce230e5af7453e8f1ae5bfc0f3e2eda27d Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Wed, 10 Jun 2026 21:33:34 -0400 Subject: [PATCH 6/8] feat: sandbox toggles in New Session dialog, gated on bwrap availability Adds sandbox_available param to NewSessionScreen, renders Switch/Select/ Switch widgets for sandbox on/off, filesystem access (rw/ro), and network toggle. When bwrap is unavailable the switch is disabled and a warning label is shown. The _policy_from staticmethod maps widget values to SandboxPolicy. The result tuple gains SandboxPolicy | None as its fifth element. app.py passes sandbox_available=sandbox.is_available() when opening the dialog and unpacks the new element to forward sandbox= to create_session. Co-Authored-By: Claude Opus 4.8 --- src/hqt/tui/app.py | 15 ++++++-- src/hqt/tui/screens/new_session.py | 49 +++++++++++++++++++++++-- tests/test_tui.py | 59 ++++++++++++++++++++++++++---- 3 files changed, 108 insertions(+), 15 deletions(-) diff --git a/src/hqt/tui/app.py b/src/hqt/tui/app.py index 82a1de1..2726734 100644 --- a/src/hqt/tui/app.py +++ b/src/hqt/tui/app.py @@ -10,6 +10,7 @@ from textual.screen import ModalScreen from textual.theme import Theme from textual.widgets import Footer +from hqt import sandbox from hqt.config import get_settings from hqt.git import worktree from hqt.db.engine import ensure_db, get_engine, get_session_factory @@ -255,10 +256,10 @@ class HqtApp(App): project_path = project.path def on_dismiss( - result: tuple[str, str, str | None, str | None] | None, + result: tuple[str, str, str | None, str | None, object] | None, ) -> None: if result: - harness_name, nickname, model, worktree_branch = result + harness_name, nickname, model, worktree_branch, sandbox_policy = result async def _do() -> None: create_result = await self._sessions().create_session( @@ -267,6 +268,7 @@ class HqtApp(App): nickname, model, worktree_branch=worktree_branch, + sandbox=sandbox_policy, ) if not create_result.spawn_ok: first_line = next( @@ -296,7 +298,14 @@ class HqtApp(App): except Exception as exc: # noqa: BLE001 - keep the worker from crashing the TUI log.warning("is_git_repo check failed for %s: %s", project_path, exc) is_git_repo = False - self.push_screen(NewSessionScreen(harness_names, is_git_repo), on_dismiss) + self.push_screen( + NewSessionScreen( + harness_names, + is_git_repo, + sandbox_available=sandbox.is_available(), + ), + on_dismiss, + ) self.run_worker(_open()) diff --git a/src/hqt/tui/screens/new_session.py b/src/hqt/tui/screens/new_session.py index 4933cf9..2b43fd6 100644 --- a/src/hqt/tui/screens/new_session.py +++ b/src/hqt/tui/screens/new_session.py @@ -1,15 +1,24 @@ from textual.app import ComposeResult from textual.containers import Horizontal, Vertical from textual.screen import ModalScreen -from textual.widgets import Button, Checkbox, Input, Label, Select +from textual.widgets import Button, Checkbox, Input, Label, Select, Switch from hqt.git import worktree +from hqt.sandbox import SandboxPolicy -class NewSessionScreen(ModalScreen[tuple[str, str, str | None, str | None] | None]): - def __init__(self, harness_names: list[str], is_git_repo: bool) -> None: +class NewSessionScreen( + ModalScreen[tuple[str, str, str | None, str | None, SandboxPolicy | None] | None] +): + def __init__( + self, + harness_names: list[str], + is_git_repo: bool = False, + sandbox_available: bool = False, + ) -> None: self.harness_names = harness_names self._is_git_repo = is_git_repo + self.sandbox_available = sandbox_available super().__init__() def compose(self) -> ComposeResult: @@ -43,6 +52,26 @@ class NewSessionScreen(ModalScreen[tuple[str, str, str | None, str | None] | Non branch_input = Input(placeholder="branch name", id="branch-input") branch_input.display = False yield branch_input + yield Label("Sandboxed (bubblewrap):") + yield Switch( + id="sandbox-switch", + value=False, + disabled=not self.sandbox_available, + ) + if not self.sandbox_available: + yield Label( + "bubblewrap not found — run `hqt doctor`", + id="sandbox-warning", + ) + yield Label("Filesystem access:") + yield Select( + [("Read-write", "rw"), ("Read-only", "ro")], + id="fs-select", + value="rw", + allow_blank=False, + ) + yield Label("Network:") + yield Switch(id="net-switch", value=True) with Horizontal(classes="dialog-actions"): yield Button("OK", variant="primary", id="ok-btn") yield Button("Cancel", id="cancel-btn") @@ -71,6 +100,13 @@ class NewSessionScreen(ModalScreen[tuple[str, str, str | None, str | None] | Non # can be completed from the keyboard without reaching for the mouse. self._submit() + @staticmethod + def _policy_from(enabled: bool, fs: str, net: bool) -> SandboxPolicy | None: + """Pure mapping from widget values to a policy (None when disabled).""" + if not enabled: + return None + return SandboxPolicy(fs=fs, net=net) + def _submit(self) -> None: harness = self.query_one("#harness-select", Select).value nickname = self.query_one("#nickname-input", Input).value @@ -86,7 +122,12 @@ class NewSessionScreen(ModalScreen[tuple[str, str, str | None, str | None] | Non branch_input.focus() return + enabled = self.query_one("#sandbox-switch", Switch).value + fs = self.query_one("#fs-select", Select).value + net = self.query_one("#net-switch", Switch).value + policy = self._policy_from(bool(enabled), str(fs), bool(net)) + if harness and harness != Select.BLANK: - self.dismiss((str(harness), nickname, model, worktree_branch)) + self.dismiss((str(harness), nickname, model, worktree_branch, policy)) else: self.dismiss(None) diff --git a/tests/test_tui.py b/tests/test_tui.py index d5ad978..fb50dc4 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -292,7 +292,7 @@ async def test_new_session_on_dismiss_sequential_create_then_refresh(): call_order: list[str] = [] async def fake_create( - project_id, harness_name, nickname, model, worktree_branch=None + project_id, harness_name, nickname, model, worktree_branch=None, sandbox=None ): # Brief yield so that, if two separate workers were used, the list # worker would be able to start and record "list" before this @@ -324,7 +324,7 @@ async def test_new_session_on_dismiss_sequential_create_then_refresh(): await pilot.pause() # The NewSessionScreen is now on top; dismiss it with a valid result tuple. - app.screen.dismiss(("claude", "mynick", None, None)) + app.screen.dismiss(("claude", "mynick", None, None, None)) await pilot.pause() # Wait for the worker spawned by on_dismiss to finish deterministically. @@ -396,7 +396,7 @@ async def test_new_session_on_dismiss_notifies_on_spawn_failure(): await app.workers.wait_for_complete() await pilot.pause() - app.screen.dismiss(("claude", "mynick", None, None)) + app.screen.dismiss(("claude", "mynick", None, None, None)) await pilot.pause() await app.workers.wait_for_complete() await pilot.pause() @@ -467,7 +467,7 @@ async def test_new_session_on_dismiss_no_notify_on_success(): await app.workers.wait_for_complete() await pilot.pause() - app.screen.dismiss(("claude", "mynick", None, None)) + app.screen.dismiss(("claude", "mynick", None, None, None)) await pilot.pause() await app.workers.wait_for_complete() await pilot.pause() @@ -582,7 +582,7 @@ async def test_new_session_enter_in_input_submits_dialog(): await pilot.press("enter") await pilot.pause() - assert results == [("claude", "mywork", None, None)], f"got {results}" + assert results == [("claude", "mywork", None, None, None)], f"got {results}" # --------------------------------------------------------------------------- @@ -1291,7 +1291,7 @@ async def test_new_session_worktree_unchecked_returns_none_branch(): app.screen.query_one("#nickname-input", Input).value = "mywork" app.screen.query_one("#ok-btn", Button).press() await pilot.pause() - assert results == [("claude", "mywork", None, None)], f"got {results}" + assert results == [("claude", "mywork", None, None, None)], f"got {results}" @pytest.mark.asyncio @@ -1316,7 +1316,7 @@ async def test_new_session_worktree_checked_uses_typed_branch(): branch.value = "feature-x" app.screen.query_one("#ok-btn", Button).press() await pilot.pause() - assert results == [("claude", "My Work", None, "feature-x")], f"got {results}" + assert results == [("claude", "My Work", None, "feature-x", None)], f"got {results}" @pytest.mark.asyncio @@ -1338,7 +1338,7 @@ async def test_new_session_worktree_checked_blank_branch_slugifies_nickname(): app.screen.query_one("#branch-input", Input).value = "" app.screen.query_one("#ok-btn", Button).press() await pilot.pause() - assert results == [("claude", "Cool Feature!", None, "cool-feature")], ( + assert results == [("claude", "Cool Feature!", None, "cool-feature", None)], ( f"got {results}" ) @@ -1504,3 +1504,46 @@ async def test_delete_non_worktree_session_does_not_push_modal(): verify_db = app._db_factory() assert verify_db.get(Session, sess_id) is None verify_db.close() + + +# --------------------------------------------------------------------------- +# Task 6: Sandbox controls in NewSessionScreen +# --------------------------------------------------------------------------- + + +def test_policy_from_disabled_returns_none(): + from hqt.tui.screens.new_session import NewSessionScreen + + assert NewSessionScreen._policy_from(False, "rw", True) is None + + +def test_policy_from_enabled_builds_policy(): + from hqt.sandbox import SandboxPolicy + from hqt.tui.screens.new_session import NewSessionScreen + + p = NewSessionScreen._policy_from(True, "ro", False) + assert p == SandboxPolicy(fs="ro", net=False) + + +@pytest.mark.asyncio +async def test_sandbox_switch_disabled_when_unavailable(): + from hqt.tui.screens.new_session import NewSessionScreen + from textual.widgets import Switch + + app = HqtApp() + async with app.run_test(size=(120, 40)) as pilot: + app.push_screen(NewSessionScreen(["claude"], is_git_repo=False, sandbox_available=False)) + await pilot.pause() + assert app.screen.query_one("#sandbox-switch", Switch).disabled is True + + +@pytest.mark.asyncio +async def test_sandbox_switch_enabled_when_available(): + from hqt.tui.screens.new_session import NewSessionScreen + from textual.widgets import Switch + + app = HqtApp() + async with app.run_test(size=(120, 40)) as pilot: + app.push_screen(NewSessionScreen(["claude"], is_git_repo=False, sandbox_available=True)) + await pilot.pause() + assert app.screen.query_one("#sandbox-switch", Switch).disabled is False From 7d851dcb7e29e7930c59111433d4f1835701f25b Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Wed, 10 Jun 2026 21:37:30 -0400 Subject: [PATCH 7/8] feat: doctor reports bubblewrap availability Adds a non-fatal bubblewrap check to the doctor command that reports whether the bwrap binary is available. This is necessary for sandboxed sessions and the check is displayed alongside other system requirement checks. Co-Authored-By: Claude Opus 4.8 --- src/hqt/cli.py | 7 ++++++- tests/test_cli.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 1 deletion(-) create mode 100644 tests/test_cli.py diff --git a/src/hqt/cli.py b/src/hqt/cli.py index 7069f7f..e43cef5 100644 --- a/src/hqt/cli.py +++ b/src/hqt/cli.py @@ -1,5 +1,6 @@ import logging import os +import shutil import subprocess import sys import traceback @@ -105,7 +106,6 @@ def launch(inside_tmux: bool = False): @main.command() def doctor(): """Check system requirements.""" - import shutil from hqt.harnesses.registry import discover_harnesses tmux = shutil.which("tmux") @@ -119,6 +119,11 @@ def doctor(): click.echo(f"{name}: ✓") if not harnesses: click.echo("No harnesses found on PATH") + bwrap = shutil.which("bwrap") + if bwrap: + click.echo(f"bubblewrap: ✓ {bwrap}") + else: + click.echo("bubblewrap: ✗ not found — sandboxed sessions unavailable") @main.command(name="list") diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..3be3570 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,30 @@ +from unittest.mock import patch + +from click.testing import CliRunner + +from hqt import cli as cli_module +from hqt.cli import main + + +def _which(found: set[str]): + return lambda name: f"/usr/bin/{name}" if name in found else None + + +def test_doctor_reports_bwrap_present(): + """Test that doctor reports bubblewrap when present.""" + # Patch the single shutil.which used by both cli and registry + with patch.object( + cli_module.shutil, "which", side_effect=_which({"tmux", "bwrap"}) + ): + result = CliRunner().invoke(main, ["doctor"]) + assert result.exit_code == 0 + assert "bubblewrap: ✓" in result.output + + +def test_doctor_reports_bwrap_missing(): + """Test that doctor reports bubblewrap as missing when not present.""" + # Patch the single shutil.which used by both cli and registry + with patch.object(cli_module.shutil, "which", side_effect=_which({"tmux"})): + result = CliRunner().invoke(main, ["doctor"]) + assert result.exit_code == 0 + assert "bubblewrap: ✗" in result.output From 5f2941818657fcb51ae313ce2ecda9bf9ac2bc5a Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Thu, 11 Jun 2026 08:16:48 -0400 Subject: [PATCH 8/8] fix: make sandboxed sessions usable for real git workflows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address three integration gaps found in review: - Bind ~/.gitconfig read-only in the sandbox base layer so `git commit` finds user.name/email inside the jail (spec §3). - Bind a worktree session's repo common .git dir writable: the worktree's real gitdir lives at /.git/worktrees/, outside the bound cwd, so git was broken in sandboxed worktree sessions. - doctor now gates on sandbox.is_available() (the single source of truth shared with the New Session dialog) rather than a raw shutil.which, so it agrees with the service on non-Linux platforms (spec §6). Co-Authored-By: Claude Opus 4.8 --- src/hqt/cli.py | 8 ++++--- src/hqt/sandbox.py | 5 ++++ src/hqt/sessions/service.py | 46 +++++++++++++++++++++++++++++++------ tests/test_cli.py | 15 ++++++++---- tests/test_sandbox.py | 19 +++++++++++++++ tests/test_sessions.py | 44 +++++++++++++++++++++++++++++------ tests/test_tui.py | 19 +++++++++++---- 7 files changed, 130 insertions(+), 26 deletions(-) diff --git a/src/hqt/cli.py b/src/hqt/cli.py index e43cef5..6460e94 100644 --- a/src/hqt/cli.py +++ b/src/hqt/cli.py @@ -106,6 +106,7 @@ def launch(inside_tmux: bool = False): @main.command() def doctor(): """Check system requirements.""" + from hqt import sandbox from hqt.harnesses.registry import discover_harnesses tmux = shutil.which("tmux") @@ -119,9 +120,10 @@ def doctor(): click.echo(f"{name}: ✓") if not harnesses: click.echo("No harnesses found on PATH") - bwrap = shutil.which("bwrap") - if bwrap: - click.echo(f"bubblewrap: ✓ {bwrap}") + # Use the shared helper so doctor agrees with the New Session dialog and the + # spawn backstop (single source of truth: Linux + bwrap on PATH). + if sandbox.is_available(): + click.echo(f"bubblewrap: ✓ {shutil.which('bwrap')}") else: click.echo("bubblewrap: ✗ not found — sandboxed sessions unavailable") diff --git a/src/hqt/sandbox.py b/src/hqt/sandbox.py index 1763d12..0dbca8a 100644 --- a/src/hqt/sandbox.py +++ b/src/hqt/sandbox.py @@ -59,6 +59,11 @@ def wrap( for path in _SYSTEM_RO_PATHS: if Path(path).exists(): args += ["--ro-bind", path, path] + # Git reads user.name/email from ~/.gitconfig; without it `git commit` fails + # inside the jail. Bound read-only since $HOME itself is hidden. + gitconfig = Path.home() / ".gitconfig" + if gitconfig.exists(): + args += ["--ro-bind", str(gitconfig), str(gitconfig)] args += ["--proc", "/proc", "--dev", "/dev", "--tmpfs", "/tmp"] for bind in binds: if not bind.src.exists(): diff --git a/src/hqt/sessions/service.py b/src/hqt/sessions/service.py index 9e9595a..f7ea5ab 100644 --- a/src/hqt/sessions/service.py +++ b/src/hqt/sessions/service.py @@ -3,7 +3,7 @@ import contextlib import json import logging import time -from collections.abc import Mapping +from collections.abc import Mapping, Sequence from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path @@ -17,7 +17,7 @@ from hqt.errors import ServiceError from hqt.git import worktree from hqt.git.worktree import WorktreeState from hqt.harnesses.base import HarnessConfigurator, SpawnConfig -from hqt.sandbox import SandboxPolicy +from hqt.sandbox import Bind, SandboxPolicy from hqt.status import window_label from hqt.tmux.manager import SpawnRequest, TmuxManager from hqt.tmux.runner import WindowInfo @@ -169,12 +169,29 @@ class SessionService: data = json.loads(sess.sandbox_json) return SandboxPolicy(fs=data["fs"], net=data["net"]) + def _worktree_git_binds(self, db: DBSession, sess: Session) -> list[Bind]: + """Extra binds a sandboxed worktree session needs for git to work. + + A linked worktree's `.git` is a file pointing at the real gitdir under + the repo's common `.git` directory (`/.git/worktrees/`), + which lives OUTSIDE the bound cwd. Bind the repo's `.git` writable so + git operations (status, commit) resolve inside the jail. Plain sessions + need nothing extra. + """ + if not sess.worktree_path or sess.worktree_branch is None: + return [] + repo = self._repo_path_for_worktree( + db, sess.project_id, sess.worktree_path, sess.worktree_branch + ) + return [Bind(repo / ".git", writable=True)] + def _wrap_command( self, configurator: HarnessConfigurator, command: list[str], cwd: Path, policy: SandboxPolicy | None, + extra_binds: Sequence[Bind] = (), ) -> list[str]: """Wrap a harness command in bwrap when a policy is set. @@ -187,7 +204,8 @@ class SessionService: raise ServiceError( "bubblewrap (bwrap) is not available; cannot start a sandboxed session" ) - return sandbox.wrap(command, cwd, policy, configurator.sandbox_binds()) + binds = [*configurator.sandbox_binds(), *extra_binds] + return sandbox.wrap(command, cwd, policy, binds) def _maybe_capture_missing_session_id(self, db: DBSession, sess: Session) -> None: configurator = self.harnesses.get(sess.harness.name) @@ -278,7 +296,11 @@ class SessionService: run_path, harness_session_id, model, sandboxed=sandbox is not None ) command = self._wrap_command( - configurator, spawn_cfg.command, spawn_cfg.cwd, sandbox + configurator, + spawn_cfg.command, + spawn_cfg.cwd, + sandbox, + self._worktree_git_binds(db, sess), ) log.info( "Spawning window %s: cmd=%s cwd=%s", @@ -396,7 +418,13 @@ class SessionService: spawn_cfg = configurator.build_spawn_config( run_path, harness_session_id, sess.model, sandboxed=policy is not None ) - command = self._wrap_command(configurator, spawn_cfg.command, spawn_cfg.cwd, policy) + command = self._wrap_command( + configurator, + spawn_cfg.command, + spawn_cfg.cwd, + policy, + self._worktree_git_binds(db, sess), + ) async with self._maybe_capture_lock(configurator.captures_session_id): since = time.time() result = await self.tmux.respawn_verified( @@ -595,8 +623,12 @@ class SessionService: ) policy = self._sandbox_policy(sess) cfg = configurator.build_resume_config( - self._session_path(db, sess), harness_session_id, sess.model, + self._session_path(db, sess), + harness_session_id, + sess.model, sandboxed=policy is not None, ) - command = self._wrap_command(configurator, cfg.command, cfg.cwd, policy) + command = self._wrap_command( + configurator, cfg.command, cfg.cwd, policy, self._worktree_git_binds(db, sess) + ) return SpawnConfig(command=command, env=cfg.env, cwd=cfg.cwd) diff --git a/tests/test_cli.py b/tests/test_cli.py index 3be3570..377c809 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -3,6 +3,7 @@ from unittest.mock import patch from click.testing import CliRunner from hqt import cli as cli_module +from hqt import sandbox from hqt.cli import main @@ -12,9 +13,11 @@ def _which(found: set[str]): def test_doctor_reports_bwrap_present(): """Test that doctor reports bubblewrap when present.""" - # Patch the single shutil.which used by both cli and registry - with patch.object( - cli_module.shutil, "which", side_effect=_which({"tmux", "bwrap"}) + # tmux is discovered via cli.shutil.which; bwrap availability comes from the + # shared sandbox.is_available() helper (the single source of truth). + with ( + patch.object(cli_module.shutil, "which", side_effect=_which({"tmux", "bwrap"})), + patch.object(sandbox, "is_available", return_value=True), ): result = CliRunner().invoke(main, ["doctor"]) assert result.exit_code == 0 @@ -23,8 +26,10 @@ def test_doctor_reports_bwrap_present(): def test_doctor_reports_bwrap_missing(): """Test that doctor reports bubblewrap as missing when not present.""" - # Patch the single shutil.which used by both cli and registry - with patch.object(cli_module.shutil, "which", side_effect=_which({"tmux"})): + with ( + patch.object(cli_module.shutil, "which", side_effect=_which({"tmux"})), + patch.object(sandbox, "is_available", return_value=False), + ): result = CliRunner().invoke(main, ["doctor"]) assert result.exit_code == 0 assert "bubblewrap: ✗" in result.output diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py index b39e650..d3c9143 100644 --- a/tests/test_sandbox.py +++ b/tests/test_sandbox.py @@ -102,3 +102,22 @@ def test_wrap_skips_missing_binds(tmp_path): ["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [Bind(missing)] ) assert str(missing) not in args + + +def test_wrap_binds_gitconfig_readonly_when_present(tmp_path, monkeypatch): + """~/.gitconfig is bound read-only so `git commit` works inside the jail.""" + home = tmp_path / "home" + home.mkdir() + gitconfig = home / ".gitconfig" + gitconfig.write_text("[user]\n\tname = Test\n") + monkeypatch.setattr(sandbox.Path, "home", classmethod(lambda cls: home)) + args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), []) + assert (str(gitconfig), str(gitconfig)) in _split(args, "--ro-bind") + + +def test_wrap_skips_gitconfig_when_absent(tmp_path, monkeypatch): + home = tmp_path / "home" + home.mkdir() + monkeypatch.setattr(sandbox.Path, "home", classmethod(lambda cls: home)) + args = sandbox.wrap(["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), []) + assert str(home / ".gitconfig") not in args diff --git a/tests/test_sessions.py b/tests/test_sessions.py index 4b24fd9..999fd71 100644 --- a/tests/test_sessions.py +++ b/tests/test_sessions.py @@ -1547,10 +1547,15 @@ def sandbox_service(factory, db, tmux, sandbox_harness): @pytest.mark.asyncio -async def test_sandboxed_create_wraps_command(sandbox_service, db, tmux, sandbox_harness): - with patch("hqt.sessions.service.sandbox.is_available", return_value=True), patch( - "hqt.sessions.service.sandbox.wrap", return_value=["bwrap", "--", "claude"] - ) as mock_wrap: +async def test_sandboxed_create_wraps_command( + sandbox_service, db, tmux, sandbox_harness +): + with ( + patch("hqt.sessions.service.sandbox.is_available", return_value=True), + patch( + "hqt.sessions.service.sandbox.wrap", return_value=["bwrap", "--", "claude"] + ) as mock_wrap, + ): result = await sandbox_service.create_session( project_id=1, harness_name="claude-code", @@ -1558,9 +1563,10 @@ async def test_sandboxed_create_wraps_command(sandbox_service, db, tmux, sandbox ) mock_wrap.assert_called_once() sandbox_harness["claude-code"].build_spawn_config.assert_called_once() - assert sandbox_harness["claude-code"].build_spawn_config.call_args.kwargs[ - "sandboxed" - ] is True + assert ( + sandbox_harness["claude-code"].build_spawn_config.call_args.kwargs["sandboxed"] + is True + ) assert tmux.spawn.call_args.args[0].command == ["bwrap", "--", "claude"] assert result.session.sandbox_json == '{"fs": "rw", "net": true}' @@ -1582,3 +1588,27 @@ async def test_sandboxed_create_raises_when_bwrap_missing(sandbox_service, db): harness_name="claude-code", sandbox=SandboxPolicy(fs="rw", net=True), ) + + +@pytest.mark.asyncio +async def test_sandboxed_worktree_binds_repo_git_dir( + sandbox_service, db, tmux, sandbox_harness, fake_worktree +): + """A sandboxed worktree session binds the repo's common .git writable so + git works inside the jail (the worktree's gitdir lives under /.git).""" + from hqt.sandbox import Bind + + with ( + patch("hqt.sessions.service.sandbox.is_available", return_value=True), + patch( + "hqt.sessions.service.sandbox.wrap", return_value=["bwrap", "--", "claude"] + ) as mock_wrap, + ): + await sandbox_service.create_session( + project_id=1, + harness_name="claude-code", + worktree_branch="feature-x", + sandbox=SandboxPolicy(fs="rw", net=True), + ) + binds = mock_wrap.call_args.args[3] + assert Bind(Path("/tmp/myproj/.git"), writable=True) in binds diff --git a/tests/test_tui.py b/tests/test_tui.py index fb50dc4..5ee64a4 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -292,7 +292,12 @@ async def test_new_session_on_dismiss_sequential_create_then_refresh(): call_order: list[str] = [] async def fake_create( - project_id, harness_name, nickname, model, worktree_branch=None, sandbox=None + project_id, + harness_name, + nickname, + model, + worktree_branch=None, + sandbox=None, ): # Brief yield so that, if two separate workers were used, the list # worker would be able to start and record "list" before this @@ -1316,7 +1321,9 @@ async def test_new_session_worktree_checked_uses_typed_branch(): branch.value = "feature-x" app.screen.query_one("#ok-btn", Button).press() await pilot.pause() - assert results == [("claude", "My Work", None, "feature-x", None)], f"got {results}" + assert results == [("claude", "My Work", None, "feature-x", None)], ( + f"got {results}" + ) @pytest.mark.asyncio @@ -1532,7 +1539,9 @@ async def test_sandbox_switch_disabled_when_unavailable(): app = HqtApp() async with app.run_test(size=(120, 40)) as pilot: - app.push_screen(NewSessionScreen(["claude"], is_git_repo=False, sandbox_available=False)) + app.push_screen( + NewSessionScreen(["claude"], is_git_repo=False, sandbox_available=False) + ) await pilot.pause() assert app.screen.query_one("#sandbox-switch", Switch).disabled is True @@ -1544,6 +1553,8 @@ async def test_sandbox_switch_enabled_when_available(): app = HqtApp() async with app.run_test(size=(120, 40)) as pilot: - app.push_screen(NewSessionScreen(["claude"], is_git_repo=False, sandbox_available=True)) + app.push_screen( + NewSessionScreen(["claude"], is_git_repo=False, sandbox_available=True) + ) await pilot.pause() assert app.screen.query_one("#sandbox-switch", Switch).disabled is False