diff --git a/src/hqt/cli.py b/src/hqt/cli.py index 7069f7f..6460e94 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,7 @@ def launch(inside_tmux: bool = False): @main.command() def doctor(): """Check system requirements.""" - import shutil + from hqt import sandbox from hqt.harnesses.registry import discover_harnesses tmux = shutil.which("tmux") @@ -119,6 +120,12 @@ def doctor(): click.echo(f"{name}: ✓") if not harnesses: click.echo("No harnesses found on PATH") + # 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") @main.command(name="list") diff --git a/src/hqt/db/migrations.py b/src/hqt/db/migrations.py index d8b7f02..c91c164 100644 --- a/src/hqt/db/migrations.py +++ b/src/hqt/db/migrations.py @@ -36,9 +36,14 @@ def _migrate_v3(conn: Connection) -> None: ) +def _migrate_v4(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), + (4, _migrate_v4), ] BASELINE_VERSION = 1 diff --git a/src/hqt/db/models.py b/src/hqt/db/models.py index cc70a68..7dcc6bc 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/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/src/hqt/sandbox.py b/src/hqt/sandbox.py new file mode 100644 index 0000000..0dbca8a --- /dev/null +++ b/src/hqt/sandbox.py @@ -0,0 +1,81 @@ +import os +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 + + +# 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] + # 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(): + 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/src/hqt/sessions/service.py b/src/hqt/sessions/service.py index 1f4579c..83ee26c 100644 --- a/src/hqt/sessions/service.py +++ b/src/hqt/sessions/service.py @@ -1,8 +1,9 @@ import asyncio 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 @@ -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, ScheduledPrompt, 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 Bind, SandboxPolicy from hqt.status import window_label from hqt.tmux.manager import SpawnRequest, TmuxManager from hqt.tmux.runner import WindowInfo @@ -164,6 +167,51 @@ 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 _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. + + 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" + ) + 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) if configurator is None or not configurator.captures_session_id: @@ -197,6 +245,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. @@ -243,13 +292,25 @@ 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, + self._worktree_git_binds(db, sess), ) 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): @@ -257,7 +318,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, ) @@ -358,13 +419,21 @@ 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, + 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( - 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( @@ -643,6 +712,14 @@ 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, self._worktree_git_binds(db, sess) + ) + return SpawnConfig(command=command, env=cfg.env, cwd=cfg.cwd) 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_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..377c809 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,35 @@ +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 + + +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.""" + # 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 + assert "bubblewrap: ✓" in result.output + + +def test_doctor_reports_bwrap_missing(): + """Test that doctor reports bubblewrap as missing when not present.""" + 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_db.py b/tests/test_db.py index 10c1671..cabeb6d 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -98,6 +98,7 @@ def test_unversioned_existing_db_stamped_latest(tmp_path): conn.exec_driver_sql("DROP TABLE scheduled_prompts") 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 @@ -117,22 +118,23 @@ def test_pending_migrations_applied_in_order_once(tmp_path, monkeypatch): ensure_db(settings) # stamps migrations.LATEST_VERSION applied: list[int] = [] # Use versions above current LATEST_VERSION so none have been applied yet. + next_version = migrations.LATEST_VERSION + 1 monkeypatch.setattr( migrations, "MIGRATIONS", [ *migrations.MIGRATIONS, - (4, lambda conn: applied.append(4)), - (5, lambda conn: applied.append(5)), + (next_version, lambda conn: applied.append(next_version)), + (next_version + 1, lambda conn: applied.append(next_version + 1)), ], ) - monkeypatch.setattr(migrations, "LATEST_VERSION", 5) + monkeypatch.setattr(migrations, "LATEST_VERSION", next_version + 1) engine = get_engine(settings) migrate(engine) - assert applied == [4, 5] - assert _user_version(engine) == 5 + assert applied == [next_version, next_version + 1] + assert _user_version(engine) == next_version + 1 migrate(engine) # re-run is a no-op - assert applied == [4, 5] + assert applied == [next_version, next_version + 1] def _session_column_names(engine) -> list[str]: @@ -154,11 +156,12 @@ 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 "sandbox_json" in cols assert _user_version(engine) == migrations.LATEST_VERSION -def test_v1_db_upgraded_to_v2(tmp_path): - # Build a v1 database: create all tables, drop the two new columns, stamp v1. +def test_v1_db_upgraded_to_latest(tmp_path): + # Build a v1 database: create all tables, drop v2+ objects, stamp v1. settings = _tmp_settings(tmp_path) settings.db_path.parent.mkdir(parents=True, exist_ok=True) engine = get_engine(settings) @@ -167,6 +170,7 @@ def test_v1_db_upgraded_to_v2(tmp_path): conn.exec_driver_sql("DROP TABLE scheduled_prompts") 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) @@ -174,6 +178,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 "sandbox_json" in cols + assert "scheduled_prompts" in inspect(engine).get_table_names() assert _user_version(engine) == migrations.LATEST_VERSION @@ -195,22 +201,24 @@ def test_fresh_db_has_scheduled_prompts_table(tmp_path): "created_at", "sent_at", ] - assert _user_version(engine) == 3 + assert _user_version(engine) == migrations.LATEST_VERSION -def test_v2_db_upgraded_to_v3_adds_scheduled_prompts(tmp_path): +def test_v2_db_upgraded_to_latest_adds_scheduled_prompts_and_sandbox(tmp_path): settings = _tmp_settings(tmp_path) settings.db_path.parent.mkdir(parents=True, exist_ok=True) engine = get_engine(settings) Base.metadata.create_all(engine) with engine.begin() as conn: conn.exec_driver_sql("DROP TABLE scheduled_prompts") + conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN sandbox_json") conn.exec_driver_sql("PRAGMA user_version = 2") migrate(engine) assert "scheduled_prompts" in inspect(engine).get_table_names() - assert _user_version(engine) == 3 + assert "sandbox_json" in _session_column_names(engine) + assert _user_version(engine) == migrations.LATEST_VERSION def test_scheduled_prompt_one_row_per_session(tmp_path): @@ -291,3 +299,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_four(): + assert migrations.LATEST_VERSION == 4 + + +def test_migration_adds_sandbox_json_column(tmp_path): + # Simulate a v3 DB whose sessions table predates the sandbox 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 = 3") + migrate(engine) + cols = [c["name"] for c in inspect(engine).get_columns("sessions")] + assert "sandbox_json" in cols + assert _user_version(engine) == migrations.LATEST_VERSION + + +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}' 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"] diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py new file mode 100644 index 0000000..d3c9143 --- /dev/null +++ b/tests/test_sandbox.py @@ -0,0 +1,123 @@ +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 + + +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 + + +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 d645222..cf3544f 100644 --- a/tests/test_sessions.py +++ b/tests/test_sessions.py @@ -9,6 +9,7 @@ from sqlalchemy.pool import StaticPool from hqt.db.models import Base, Harness, Project, ScheduledPrompt, Session from hqt.errors import ServiceError +from hqt.sandbox import SandboxPolicy from hqt.sessions.service import SessionInfo, SessionService from hqt.tmux.manager import SpawnResult, TmuxManager @@ -931,7 +932,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 @@ -1646,3 +1647,96 @@ 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), + ) + + +@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 d5ad978..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 + 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 +329,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 +401,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 +472,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 +587,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 +1296,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 +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")], f"got {results}" + assert results == [("claude", "My Work", None, "feature-x", None)], ( + f"got {results}" + ) @pytest.mark.asyncio @@ -1338,7 +1345,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 +1511,50 @@ 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