fix: make sandboxed sessions usable for real git workflows

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 <repo>/.git/worktrees/<branch>, 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 <noreply@anthropic.com>
This commit is contained in:
2026-06-11 08:16:48 -04:00
parent 7d851dcb7e
commit 5f29418186
7 changed files with 130 additions and 26 deletions
+5 -3
View File
@@ -106,6 +106,7 @@ def launch(inside_tmux: bool = False):
@main.command() @main.command()
def doctor(): def doctor():
"""Check system requirements.""" """Check system requirements."""
from hqt import sandbox
from hqt.harnesses.registry import discover_harnesses from hqt.harnesses.registry import discover_harnesses
tmux = shutil.which("tmux") tmux = shutil.which("tmux")
@@ -119,9 +120,10 @@ def doctor():
click.echo(f"{name}: ✓") click.echo(f"{name}: ✓")
if not harnesses: if not harnesses:
click.echo("No harnesses found on PATH") click.echo("No harnesses found on PATH")
bwrap = shutil.which("bwrap") # Use the shared helper so doctor agrees with the New Session dialog and the
if bwrap: # spawn backstop (single source of truth: Linux + bwrap on PATH).
click.echo(f"bubblewrap: ✓ {bwrap}") if sandbox.is_available():
click.echo(f"bubblewrap: ✓ {shutil.which('bwrap')}")
else: else:
click.echo("bubblewrap: ✗ not found — sandboxed sessions unavailable") click.echo("bubblewrap: ✗ not found — sandboxed sessions unavailable")
+5
View File
@@ -59,6 +59,11 @@ def wrap(
for path in _SYSTEM_RO_PATHS: for path in _SYSTEM_RO_PATHS:
if Path(path).exists(): if Path(path).exists():
args += ["--ro-bind", path, path] 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"] args += ["--proc", "/proc", "--dev", "/dev", "--tmpfs", "/tmp"]
for bind in binds: for bind in binds:
if not bind.src.exists(): if not bind.src.exists():
+39 -7
View File
@@ -3,7 +3,7 @@ import contextlib
import json import json
import logging import logging
import time import time
from collections.abc import Mapping from collections.abc import Mapping, Sequence
from dataclasses import dataclass from dataclasses import dataclass
from datetime import UTC, datetime from datetime import UTC, datetime
from pathlib import Path from pathlib import Path
@@ -17,7 +17,7 @@ from hqt.errors import ServiceError
from hqt.git import worktree from hqt.git import worktree
from hqt.git.worktree import WorktreeState from hqt.git.worktree import WorktreeState
from hqt.harnesses.base import HarnessConfigurator, SpawnConfig 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.status import window_label
from hqt.tmux.manager import SpawnRequest, TmuxManager from hqt.tmux.manager import SpawnRequest, TmuxManager
from hqt.tmux.runner import WindowInfo from hqt.tmux.runner import WindowInfo
@@ -169,12 +169,29 @@ class SessionService:
data = json.loads(sess.sandbox_json) data = json.loads(sess.sandbox_json)
return SandboxPolicy(fs=data["fs"], net=data["net"]) 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 (`<repo>/.git/worktrees/<branch>`),
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( def _wrap_command(
self, self,
configurator: HarnessConfigurator, configurator: HarnessConfigurator,
command: list[str], command: list[str],
cwd: Path, cwd: Path,
policy: SandboxPolicy | None, policy: SandboxPolicy | None,
extra_binds: Sequence[Bind] = (),
) -> list[str]: ) -> list[str]:
"""Wrap a harness command in bwrap when a policy is set. """Wrap a harness command in bwrap when a policy is set.
@@ -187,7 +204,8 @@ class SessionService:
raise ServiceError( raise ServiceError(
"bubblewrap (bwrap) is not available; cannot start a sandboxed session" "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: def _maybe_capture_missing_session_id(self, db: DBSession, sess: Session) -> None:
configurator = self.harnesses.get(sess.harness.name) configurator = self.harnesses.get(sess.harness.name)
@@ -278,7 +296,11 @@ class SessionService:
run_path, harness_session_id, model, sandboxed=sandbox is not None run_path, harness_session_id, model, sandboxed=sandbox is not None
) )
command = self._wrap_command( 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( log.info(
"Spawning window %s: cmd=%s cwd=%s", "Spawning window %s: cmd=%s cwd=%s",
@@ -396,7 +418,13 @@ class SessionService:
spawn_cfg = configurator.build_spawn_config( spawn_cfg = configurator.build_spawn_config(
run_path, harness_session_id, sess.model, sandboxed=policy is not None 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): async with self._maybe_capture_lock(configurator.captures_session_id):
since = time.time() since = time.time()
result = await self.tmux.respawn_verified( result = await self.tmux.respawn_verified(
@@ -595,8 +623,12 @@ class SessionService:
) )
policy = self._sandbox_policy(sess) policy = self._sandbox_policy(sess)
cfg = configurator.build_resume_config( 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, 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) return SpawnConfig(command=command, env=cfg.env, cwd=cfg.cwd)
+10 -5
View File
@@ -3,6 +3,7 @@ from unittest.mock import patch
from click.testing import CliRunner from click.testing import CliRunner
from hqt import cli as cli_module from hqt import cli as cli_module
from hqt import sandbox
from hqt.cli import main from hqt.cli import main
@@ -12,9 +13,11 @@ def _which(found: set[str]):
def test_doctor_reports_bwrap_present(): def test_doctor_reports_bwrap_present():
"""Test that doctor reports bubblewrap when present.""" """Test that doctor reports bubblewrap when present."""
# Patch the single shutil.which used by both cli and registry # tmux is discovered via cli.shutil.which; bwrap availability comes from the
with patch.object( # shared sandbox.is_available() helper (the single source of truth).
cli_module.shutil, "which", side_effect=_which({"tmux", "bwrap"}) 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"]) result = CliRunner().invoke(main, ["doctor"])
assert result.exit_code == 0 assert result.exit_code == 0
@@ -23,8 +26,10 @@ def test_doctor_reports_bwrap_present():
def test_doctor_reports_bwrap_missing(): def test_doctor_reports_bwrap_missing():
"""Test that doctor reports bubblewrap as missing when not present.""" """Test that doctor reports bubblewrap as missing when not present."""
# Patch the single shutil.which used by both cli and registry with (
with patch.object(cli_module.shutil, "which", side_effect=_which({"tmux"})): patch.object(cli_module.shutil, "which", side_effect=_which({"tmux"})),
patch.object(sandbox, "is_available", return_value=False),
):
result = CliRunner().invoke(main, ["doctor"]) result = CliRunner().invoke(main, ["doctor"])
assert result.exit_code == 0 assert result.exit_code == 0
assert "bubblewrap: ✗" in result.output assert "bubblewrap: ✗" in result.output
+19
View File
@@ -102,3 +102,22 @@ def test_wrap_skips_missing_binds(tmp_path):
["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [Bind(missing)] ["sh"], tmp_path, SandboxPolicy(fs="rw", net=True), [Bind(missing)]
) )
assert str(missing) not in args 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
+37 -7
View File
@@ -1547,10 +1547,15 @@ def sandbox_service(factory, db, tmux, sandbox_harness):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_sandboxed_create_wraps_command(sandbox_service, db, tmux, sandbox_harness): async def test_sandboxed_create_wraps_command(
with patch("hqt.sessions.service.sandbox.is_available", return_value=True), patch( sandbox_service, db, tmux, sandbox_harness
"hqt.sessions.service.sandbox.wrap", return_value=["bwrap", "--", "claude"] ):
) as mock_wrap: 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( result = await sandbox_service.create_session(
project_id=1, project_id=1,
harness_name="claude-code", 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() mock_wrap.assert_called_once()
sandbox_harness["claude-code"].build_spawn_config.assert_called_once() sandbox_harness["claude-code"].build_spawn_config.assert_called_once()
assert sandbox_harness["claude-code"].build_spawn_config.call_args.kwargs[ assert (
"sandboxed" sandbox_harness["claude-code"].build_spawn_config.call_args.kwargs["sandboxed"]
] is True is True
)
assert tmux.spawn.call_args.args[0].command == ["bwrap", "--", "claude"] assert tmux.spawn.call_args.args[0].command == ["bwrap", "--", "claude"]
assert result.session.sandbox_json == '{"fs": "rw", "net": true}' 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", harness_name="claude-code",
sandbox=SandboxPolicy(fs="rw", net=True), 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 <repo>/.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
+15 -4
View File
@@ -292,7 +292,12 @@ async def test_new_session_on_dismiss_sequential_create_then_refresh():
call_order: list[str] = [] call_order: list[str] = []
async def fake_create( 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 # Brief yield so that, if two separate workers were used, the list
# worker would be able to start and record "list" before this # 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" branch.value = "feature-x"
app.screen.query_one("#ok-btn", Button).press() app.screen.query_one("#ok-btn", Button).press()
await pilot.pause() 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 @pytest.mark.asyncio
@@ -1532,7 +1539,9 @@ async def test_sandbox_switch_disabled_when_unavailable():
app = HqtApp() app = HqtApp()
async with app.run_test(size=(120, 40)) as pilot: 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() await pilot.pause()
assert app.screen.query_one("#sandbox-switch", Switch).disabled is True 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() app = HqtApp()
async with app.run_test(size=(120, 40)) as pilot: 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() await pilot.pause()
assert app.screen.query_one("#sandbox-switch", Switch).disabled is False assert app.screen.query_one("#sandbox-switch", Switch).disabled is False