Harden SSH and profile resolution boundaries

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a0233b-178d-7162-b92f-31a31ea8ca9b

tests/test_gateway.py

@@ -161,6 +161,45 @@ def test_run_execution_refuses_unknown_profile_before_sandbox() -> None:
     manager.create.assert_not_called()
 
 
+def test_run_execution_refuses_disabled_profile_before_sandbox() -> None:
+    manager = MagicMock()
+    catalog = ProfileCatalog()
+    profile, _ = catalog.resolve(PROFILE)
+    catalog.profiles()[(profile.id, profile.version)] = profile.model_copy(
+        update={"status": "disabled"}
+    )
+
+    result = run_execution(_request(), catalog=catalog, rein=_FakeRein(), manager=manager)
+
+    assert result.ok is False
+    assert result.evidence.outcome == "refused"
+    assert result.evidence.failure_stage == "resolution"
+    assert "profile disabled" in (result.evidence.error or "")
+    manager.create.assert_not_called()
+
+
+def test_run_execution_refuses_ambiguous_profile_before_sandbox() -> None:
+    manager = MagicMock()
+    catalog = ProfileCatalog()
+    profile, _ = catalog.resolve(PROFILE)
+    catalog.profiles()[(profile.id, "2.0.0")] = profile.model_copy(
+        update={"version": "2.0.0"}
+    )
+
+    result = run_execution(
+        _request(profile=profile.id),
+        catalog=catalog,
+        rein=_FakeRein(),
+        manager=manager,
+    )
+
+    assert result.ok is False
+    assert result.evidence.outcome == "refused"
+    assert result.evidence.failure_stage == "resolution"
+    assert "pin one of" in (result.evidence.error or "")
+    manager.create.assert_not_called()
+
+
 def test_run_execution_refuses_blocked_profile_before_sandbox() -> None:
     manager = MagicMock()
 

tests/test_profiles.py

@@ -22,6 +22,7 @@ def _profile(
     profile_id: str = "harness.test",
     version: str = "1.0.0",
     contract: str = "1.0",
+    status: str = "enabled",
     rein: str = "rein-test",
     required: str = "session_style: unattended",
     extra: str = "",
@@ -30,7 +31,7 @@ def _profile(
 id: {profile_id}
 version: \"{version}\"
 contract_version: \"{contract}\"
-status: enabled
+status: {status}
 rein:
   id: {rein}
   required_capabilities:
@@ -150,6 +151,26 @@ def test_unpinned_multi_version_profile_is_ambiguous(tmp_path) -> None:
         ProfileCatalog(profiles, reins).resolve("harness.test")
 
 
+def test_disabled_revision_still_requires_pinned_selection(tmp_path) -> None:
+    profiles = tmp_path / "profiles"
+    reins = tmp_path / "reins"
+    profiles.mkdir()
+    reins.mkdir()
+    _write(profiles, "enabled.yaml", _profile(version="1.0.0"))
+    _write(
+        profiles,
+        "disabled.yaml",
+        _profile(version="2.0.0", status="disabled"),
+    )
+    _write(reins, "rein.yaml", _rein())
+    catalog = ProfileCatalog(profiles, reins)
+
+    with pytest.raises(AmbiguousProfileError, match="pin one of"):
+        catalog.resolve("harness.test")
+    with pytest.raises(IncompatibleProfileError, match="profile disabled"):
+        catalog.resolve("harness.test@2.0.0")
+
+
 @pytest.mark.parametrize(
     ("profile_text", "match"),
     [

tests/test_transport.py

@@ -49,11 +49,34 @@ def test_remote_transport_wraps_command_without_local_shell() -> None:
     assert transport.kind == "ssh"
     assert transport.command(["git", "-C", "/tmp/sbx", "status"]) == [
         "ssh",
+        "--",
         "agent@sandboxer01",
         "sh -c 'cd \"$1\" && shift && exec \"$@\"' sh /tmp/sbx git -C /tmp/sbx status",
     ]
 
 
+@pytest.mark.parametrize(
+    "ssh_target",
+    ["-v", "-oProxyCommand", "agent@-host", "-agent@host", "agent@host extra"],
+)
+def test_remote_transport_rejects_ssh_option_and_invalid_targets(ssh_target) -> None:
+    sandbox = SandboxHandle(
+        sandbox_id="sbx",
+        host="sandboxer01",
+        reachability={"ssh": ssh_target, "remote_dir": "/tmp/sbx"},
+    )
+
+    with pytest.raises(TransportError, match="single non-option host"):
+        transport_from_sandbox(sandbox)
+
+
+def test_direct_remote_transport_cannot_bypass_ssh_target_validation() -> None:
+    transport = ExecutionTransport(kind="ssh", workspace="/tmp/sbx", ssh_target="-v")
+
+    with pytest.raises(TransportError, match="single non-option host"):
+        transport.command(["true"])
+
+
 @pytest.mark.parametrize(
     "reachability",
     [
@@ -103,10 +126,10 @@ def test_remote_task_file_uses_ssh_stdin_and_cleanup() -> None:
         transport.remove_file(task_path)
 
     create = run.call_args_list[0]
-    assert create.args[0][:2] == ["ssh", "agent@sandboxer01"]
+    assert create.args[0][:3] == ["ssh", "--", "agent@sandboxer01"]
     assert create.kwargs["input"] == '{"title": "bounded"}'
-    assert "cat >" in create.args[0][2]
-    assert "rm -f" in run.call_args_list[1].args[0][2]
+    assert "cat >" in create.args[0][3]
+    assert "rm -f" in run.call_args_list[1].args[0][3]
 
 
 def test_resolve_executable_fails_inside_transport() -> None: