security: refine workspace trust controls

This commit is contained in:
Rohit P
2026-07-24 18:44:39 -07:00
parent 8ee0a0d082
commit 3f5ac872ca
11 changed files with 427 additions and 9 deletions
+3 -1
View File
@@ -37,6 +37,7 @@ from .tools.directories import request_directory_tool
from .tools.plan import propose_plan_tool
from .tools.subagent import explorer_tools
from .web import make_web_fetch_tool, make_web_search_tool
from .workspace_trust import WorkspaceTrustStore
from .tools.shell import LocalExecutor
from .tools.todo import TodoList
@@ -147,7 +148,8 @@ def build_engine(
else:
root_list = []
config = load_config(ws)
workspace_trusted = bool(ws and WorkspaceTrustStore().is_trusted(ws))
config = load_config(ws, workspace_trusted=workspace_trusted)
executor = (
LocalExecutor(cwd=ws) if (agent.needs_workspace and ws is not None) else None
)
+24 -5
View File
@@ -3,8 +3,8 @@
Global: <state-dir>/config.toml (see `secrets.state_dir`; platform-native)
Workspace: <workspace>/.coworker/config.toml (overrides global)
Permission grants are deliberately global-only. A checked-out repository is untrusted
input and must not be able to grant its own shell commands or consequential tools.
Workspace command allowances apply only after the user trusts that exact canonical
workspace path. Other permission grants remain global-only.
"""
from __future__ import annotations
@@ -74,8 +74,9 @@ _FIELDS = {
"cloud_relay_ws_url",
}
# These fields change what consequential actions can run without a prompt. Only the
# user-owned global config may set them; a repository's `.coworker/config.toml` may not.
# These fields change what consequential actions can run without a prompt, so the normal
# workspace override pass never applies them. `allowed_commands` is added separately only
# for a canonically trusted workspace; `auto_allow` remains user-global only.
_GLOBAL_ONLY_FIELDS = {"allowed_commands", "auto_allow"}
_WORKSPACE_FIELDS = _FIELDS - _GLOBAL_ONLY_FIELDS
@@ -92,8 +93,20 @@ def _read(path: Path) -> dict[str, Any]:
return {}
def workspace_allowed_commands(workspace: str | Path) -> list[str]:
"""Command prefixes requested by repository config; advisory until workspace trust."""
path = Path(workspace).expanduser() / ".coworker" / "config.toml"
value = _read(path).get("allowed_commands", [])
if not isinstance(value, list):
return []
return list(dict.fromkeys(v.strip() for v in value if isinstance(v, str) and v.strip()))
def load_config(
workspace: Optional[str | Path] = None, *, global_path: Optional[Path] = None
workspace: Optional[str | Path] = None,
*,
global_path: Optional[Path] = None,
workspace_trusted: bool = False,
) -> Config:
cfg = Config()
@@ -108,4 +121,10 @@ def load_config(
for key, value in _read(w).items():
if key in _WORKSPACE_FIELDS:
setattr(cfg, key, value)
if workspace_trusted:
cfg.allowed_commands = list(
dict.fromkeys(
[*cfg.allowed_commands, *workspace_allowed_commands(workspace)]
)
)
return cfg
+14
View File
@@ -523,6 +523,17 @@ def create_app(manager: SessionManager) -> FastAPI:
body.get("path", ""), create=bool(body.get("create"))
)
@app.get("/v1/workspaces/trusted")
def trusted_workspaces() -> dict[str, Any]:
return {"workspaces": manager.trusted_workspaces()}
@app.post("/v1/workspaces/trust")
def set_workspace_trust(body: dict) -> dict[str, Any]:
return manager.set_workspace_trust(
str((body or {}).get("path", "")),
trusted=bool((body or {}).get("trusted", False)),
)
@app.post("/v1/workspaces/pick")
async def pick_workspace() -> dict[str, Any]:
# Native folder picker opened by the LOCAL sidecar (browser GUIs can't get absolute
@@ -1607,6 +1618,9 @@ def create_app(manager: SessionManager) -> FastAPI:
if getattr(engine, "executor", None)
else None
),
"command_trust": manager.workspace_command_trust(
str(getattr(engine, "audit_context", {}).get("workspace", ""))
),
},
}
)
+66 -1
View File
@@ -35,9 +35,11 @@ from ..subscriptions import ChannelBuffer, SubscriptionStore
from ..unrouted import UnroutedStore
from ..unattended import UnattendedRegistry
from ..audit import AuditStore
from ..config import load_config, workspace_allowed_commands
from ..conversations import ConversationStore, title_from
from ..engine import ApprovalOutcome, Approver, TurnEngine
from ..roots import RootDir
from ..workspace_trust import WorkspaceTrustStore
from ..automation import Schedule, ScheduledTask, Scheduler, TaskRun, TaskStore
from ..connectors import (
Gateway,
@@ -140,6 +142,7 @@ class SessionManager:
self._autotitle_inflight: set[str] = set()
self._autotitle_tasks: set[asyncio.Task] = set()
self._autotitle_attempts: dict[str, int] = {}
self.workspace_trust = WorkspaceTrustStore()
self.secrets = SecretStore()
# No explicit provider injected → route by the model's `provider:` prefix (OpenAI default,
# Ollama, …). Tests inject a provider directly and bypass the router. The same router is
@@ -242,7 +245,69 @@ class SessionManager:
return {"path": str(resolved), "ok": False, "error": str(exc)}
resolved = resolved.resolve()
self.session_store.touch_workspace(str(resolved))
return {"path": str(resolved), "ok": True, "git_branch": _git_branch(resolved)}
return {
"path": str(resolved),
"ok": True,
"git_branch": _git_branch(resolved),
"command_trust": self.workspace_command_trust(resolved),
}
def workspace_command_trust(self, path: str | Path) -> dict[str, Any]:
if not str(path).strip():
return {
"workspace": "",
"requested_commands": [],
"trusted": False,
"required": False,
}
canonical = WorkspaceTrustStore.canonical(path)
commands = (
workspace_allowed_commands(canonical)
if Path(canonical).is_dir()
else []
)
trusted = self.workspace_trust.is_trusted(canonical)
return {
"workspace": canonical,
"requested_commands": commands,
"trusted": trusted,
"required": bool(commands and not trusted),
}
def set_workspace_trust(
self, path: str | Path, *, trusted: bool
) -> dict[str, Any]:
if not str(path).strip():
return {"ok": False, "error": "workspace path is required"}
candidate = Path(path).expanduser()
if trusted and not candidate.is_dir():
return {"ok": False, "error": "workspace is not a directory"}
canonical = self.workspace_trust.set_trusted(candidate, trusted)
effective = load_config(
canonical, workspace_trusted=trusted
).allowed_commands
# Apply trust/revocation immediately to live sessions rooted at this exact path.
for engine in self._engines.values():
engine_workspace = str(
(getattr(engine, "audit_context", {}) or {}).get("workspace", "")
)
if engine_workspace and WorkspaceTrustStore.canonical(
engine_workspace
) == canonical:
engine.permissions.allowed_commands = list(effective)
return {
"ok": True,
**self.workspace_command_trust(canonical),
}
def trusted_workspaces(self) -> list[dict[str, Any]]:
return [
{
**self.workspace_command_trust(path),
"exists": Path(path).is_dir(),
}
for path in self.workspace_trust.list()
]
def recent_workspaces(self) -> list[dict[str, Any]]:
"""Recent real projects for the folder gate. Per-conversation scratch dirs are
+62
View File
@@ -0,0 +1,62 @@
"""User-owned trust decisions for repository-provided command allowances.
A repository may declare command prefixes in `.coworker/config.toml`, but those grants
take effect only after the user trusts that exact canonical workspace root. Trust follows
the path rather than a snapshot of the config: future changes at a trusted path are
accepted until the user revokes trust.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Optional
from .secrets import state_dir
class WorkspaceTrustStore:
def __init__(self, path: Optional[str | Path] = None) -> None:
self.path = (
Path(path) if path is not None else state_dir() / "workspace_trust.json"
)
@staticmethod
def canonical(path: str | Path) -> str:
return str(Path(path).expanduser().resolve())
def _load(self) -> set[str]:
try:
data = json.loads(self.path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return set()
if not isinstance(data, dict):
return set()
values = data.get("trusted_workspaces", [])
if not isinstance(values, list):
return set()
return {str(v) for v in values if isinstance(v, str) and v}
def is_trusted(self, workspace: str | Path) -> bool:
return self.canonical(workspace) in self._load()
def list(self) -> list[str]:
return sorted(self._load())
def set_trusted(self, workspace: str | Path, trusted: bool) -> str:
canonical = self.canonical(workspace)
values = self._load()
if trusted:
values.add(canonical)
else:
values.discard(canonical)
self.path.parent.mkdir(parents=True, exist_ok=True)
tmp = self.path.with_name(f".{self.path.name}.{os.getpid()}.tmp")
tmp.write_text(
json.dumps({"trusted_workspaces": sorted(values)}, indent=2) + "\n",
encoding="utf-8",
)
os.chmod(tmp, 0o600)
tmp.replace(self.path)
return canonical
+11
View File
@@ -26,6 +26,7 @@ import {
type Persona,
type RecentWorkspace,
type SurfaceVisibility,
type WorkspaceCommandTrust,
} from "./api";
import type { ApprovalDecision, Attachment, Item, SessionInfo, TodoItem, WsEvent } from "./types";
import { isProjectScoped } from "./personaScope";
@@ -54,6 +55,7 @@ import { InboxView } from "./components/InboxView";
import { ApprovalCard } from "./components/ApprovalCard";
import { DirectoryRequestCard } from "./components/DirectoryRequestCard";
import { PlanCard } from "./components/PlanCard";
import { WorkspaceTrustPrompt } from "./components/WorkspaceTrustPrompt";
const newId = () =>
(crypto as any).randomUUID ? crypto.randomUUID().slice(0, 12) : Math.random().toString(36).slice(2, 14);
@@ -145,6 +147,8 @@ export function App() {
const [workspace, setWorkspace] = useState<string | null>(null);
const [branch, setBranch] = useState<string | null>(null);
const [showGate, setShowGate] = useState(false);
const [workspaceTrustRequest, setWorkspaceTrustRequest] =
useState<WorkspaceCommandTrust | null>(null);
const [agent, setAgent] = useState("cowork");
const [model, setModel] = useState("gpt-5.6-sol");
const [models, setModels] = useState<string[]>([]);
@@ -558,6 +562,7 @@ export function App() {
setConnected(true);
if (d.model) setModel(d.model);
if (d.mode) setMode(d.mode);
if (d.command_trust?.required) setWorkspaceTrustRequest(d.command_trust);
// Cowork: adopt the server-provisioned scratch dir (only when we don't already have one).
if (d.workspace) setWorkspace((cur) => cur || d.workspace);
break;
@@ -1623,6 +1628,12 @@ export function App() {
}
/>
)}
{workspaceTrustRequest && (
<WorkspaceTrustPrompt
request={workspaceTrustRequest}
onClose={() => setWorkspaceTrustRequest(null)}
/>
)}
</div>
);
}
+32 -2
View File
@@ -24,6 +24,14 @@ export interface RecentWorkspace {
exists: boolean;
}
export interface WorkspaceCommandTrust {
workspace: string;
requested_commands: string[];
trusted: boolean;
required: boolean;
exists?: boolean;
}
export async function getHealth(): Promise<Health> {
const res = await fetch(`${httpBase()}/v1/health`);
return res.json();
@@ -49,7 +57,13 @@ export async function pickFolderViaServer(): Promise<string | null> {
export async function openWorkspace(
path: string,
create = false,
): Promise<{ path: string; ok: boolean; error?: string; git_branch?: string | null }> {
): Promise<{
path: string;
ok: boolean;
error?: string;
git_branch?: string | null;
command_trust?: WorkspaceCommandTrust;
}> {
const res = await fetch(`${httpBase()}/v1/workspaces/open`, {
method: "POST",
headers: { "Content-Type": "application/json" },
@@ -58,6 +72,23 @@ export async function openWorkspace(
return res.json();
}
export async function getTrustedWorkspaces(): Promise<WorkspaceCommandTrust[]> {
const res = await fetch(`${httpBase()}/v1/workspaces/trusted`);
return (await res.json()).workspaces ?? [];
}
export async function setWorkspaceTrusted(
path: string,
trusted: boolean,
): Promise<{ ok: boolean; error?: string } & WorkspaceCommandTrust> {
const res = await fetch(`${httpBase()}/v1/workspaces/trust`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ path, trusted }),
});
return res.json();
}
export async function getSessions(workspace?: string): Promise<SessionInfo[]> {
const q = workspace ? `?workspace=${encodeURIComponent(workspace)}` : "";
const res = await fetch(`${httpBase()}/v1/sessions${q}`);
@@ -1771,4 +1802,3 @@ export class Session {
this.ws.close();
}
}
@@ -1,12 +1,15 @@
import { useEffect, useState } from "react";
import {
getSettings,
getTrustedWorkspaces,
setOnboarded,
setPdfSettings,
setScratchBase,
setSessionsPeek,
setWorkspaceTrusted,
type ModelSettings,
type PdfSettings,
type WorkspaceCommandTrust,
} from "../api";
import {
cancelDictationModelDownload,
@@ -424,6 +427,8 @@ function AppearanceSection() {
<FilesCard />
<TrustedWorkspacesCard />
{desktop && (
<div className={CARD + " p-4"}>
<div className={FIELD_LABEL + " mb-2.5"}>Always-on</div>
@@ -461,6 +466,61 @@ function AppearanceSection() {
);
}
function TrustedWorkspacesCard() {
const [workspaces, setWorkspaces] = useState<WorkspaceCommandTrust[] | null>(null);
const refresh = () =>
getTrustedWorkspaces()
.then(setWorkspaces)
.catch(() => setWorkspaces([]));
useEffect(() => {
refresh();
}, []);
const revoke = async (path: string) => {
if (!window.confirm(`Revoke command trust for ${path}?`)) return;
await setWorkspaceTrusted(path, false);
refresh();
};
return (
<div className={CARD + " p-4 mb-4"} data-testid="trusted-workspaces-card">
<div className={FIELD_LABEL}>Trusted workspaces</div>
<div className={FIELD_HELP}>
Trusted projects may manage their command allowances in .coworker/config.toml.
</div>
{workspaces === null ? (
<div className="text-[12px] text-muted mt-3">Loading</div>
) : workspaces.length === 0 ? (
<div className="text-[12px] text-muted mt-3">No workspaces are trusted.</div>
) : (
<div className="mt-3 divide-y divide-line">
{workspaces.map((workspace) => (
<div key={workspace.workspace} className="py-2.5 flex items-start gap-3">
<div className="min-w-0 flex-1">
<div className="text-[12.5px] text-ink break-all">{workspace.workspace}</div>
<div className="text-[11.5px] text-muted mt-0.5">
{workspace.requested_commands.length
? `${workspace.requested_commands.length} project command allowance${workspace.requested_commands.length === 1 ? "" : "s"}`
: "No project command allowances currently declared"}
{!workspace.exists ? " · Folder unavailable" : ""}
</div>
</div>
<button
className="text-[12px] text-red-600 px-2 py-1"
onClick={() => void revoke(workspace.workspace)}
>
Revoke
</button>
</div>
))}
</div>
)}
</div>
);
}
function UpdateInline() {
const [state, setState] = useState<"idle" | "checking" | "none" | "found" | "installing" | "error">("idle");
const [version, setVersion] = useState("");
@@ -0,0 +1,56 @@
import { useState } from "react";
import { setWorkspaceTrusted, type WorkspaceCommandTrust } from "../api";
export function WorkspaceTrustPrompt({
request,
onClose,
}: {
request: WorkspaceCommandTrust;
onClose: () => void;
}) {
const [saving, setSaving] = useState(false);
const [error, setError] = useState("");
const trust = async () => {
setSaving(true);
setError("");
const result = await setWorkspaceTrusted(request.workspace, true).catch(() => null);
setSaving(false);
if (!result?.ok) {
setError(result?.error || "Could not save workspace trust.");
return;
}
onClose();
};
return (
<div className="gate-overlay" role="dialog" aria-modal="true" aria-labelledby="workspace-trust-title">
<div className="gate max-w-[560px]">
<div className="gate-mark"></div>
<h2 id="workspace-trust-title">Trust this workspace&rsquo;s commands?</h2>
<p className="gate-sub">
This project asks OpenWorker to run the commands below without individual approval.
Trust applies to future configuration changes at this exact folder until you revoke it
in Settings.
</p>
<div className="rounded-lg border border-line bg-paper px-3 py-2.5 max-h-48 overflow-y-auto">
{request.requested_commands.map((command) => (
<code key={command} className="block text-[12.5px] py-1 text-ink">
{command}
</code>
))}
</div>
<div className="text-[11.5px] text-muted mt-2 break-all">{request.workspace}</div>
{error && <div className="gate-error">{error}</div>}
<div className="gate-foot justify-end gap-2">
<button className="btn" onClick={onClose} disabled={saving}>
Keep asking
</button>
<button className="btn primary" onClick={() => void trust()} disabled={saving}>
{saving ? "Saving…" : "Trust workspace"}
</button>
</div>
</div>
</div>
);
}
+39
View File
@@ -47,6 +47,45 @@ def test_workspace_cannot_grant_its_own_permissions(tmp_path):
assert cfg.auto_allow == ["write_file"]
def test_trusted_workspace_adds_its_command_allowances_only(tmp_path):
g = tmp_path / "global.toml"
g.write_text(
'allowed_commands = ["git status"]\nauto_allow = ["write_file"]\n'
)
ws = tmp_path / "ws"
(ws / ".coworker").mkdir(parents=True)
(ws / ".coworker" / "config.toml").write_text(
'allowed_commands = ["pytest", "git status"]\n'
'auto_allow = ["run_shell"]\n'
)
cfg = load_config(ws, global_path=g, workspace_trusted=True)
assert cfg.allowed_commands == ["git status", "pytest"]
assert cfg.auto_allow == ["write_file"]
def test_workspace_trust_is_canonical_and_user_owned(tmp_path):
from coworker.workspace_trust import WorkspaceTrustStore
real = tmp_path / "real"
real.mkdir()
alias = tmp_path / "alias"
alias.symlink_to(real, target_is_directory=True)
store = WorkspaceTrustStore(tmp_path / "state" / "workspace_trust.json")
canonical = store.set_trusted(alias, True)
assert canonical == str(real.resolve())
assert store.is_trusted(real)
assert store.list() == [str(real.resolve())]
assert (store.path.stat().st_mode & 0o777) == 0o600
store.set_trusted(real, False)
assert not store.is_trusted(alias)
store.path.write_text("[]")
assert store.list() == []
def test_build_engine_honors_explicit_empty_command_allowlist(tmp_path):
from coworker.agent import build_code_engine
from coworker.config import global_config_path
+60
View File
@@ -606,6 +606,66 @@ def test_open_and_recent_workspaces(tmp_path):
assert any(w["path"] == str(proj.resolve()) for w in recents)
def test_workspace_command_trust_controls_live_engine(tmp_path):
from urllib.parse import quote
proj = tmp_path / "trusted-project"
(proj / ".coworker").mkdir(parents=True)
(proj / ".coworker" / "config.toml").write_text(
'allowed_commands = ["pytest"]\nauto_allow = ["write_file"]\n'
)
manager = SessionManager(
workspace=None, data_dir=tmp_path / "data", provider=ScriptedProvider([])
)
client = TestClient(create_app(manager))
with client.websocket_connect(
f"/ws/session/trust?workspace={quote(str(proj))}"
) as ws:
ready = ws.receive_json()
policy = ready["data"]["command_trust"]
assert policy["required"] is True
assert policy["requested_commands"] == ["pytest"]
engine = manager._engines["trust"]
before = engine.permissions.evaluate(
"run_shell", {"command": "pytest -q"}, None
)
assert not before.allowed and before.needs_user
# Workspace auto_allow remains ignored even after command trust.
assert "write_file" not in engine.permissions.auto_allow_tools
trusted = client.post(
"/v1/workspaces/trust",
json={"path": str(proj), "trusted": True},
).json()
assert trusted["ok"] and trusted["trusted"]
assert engine.permissions.evaluate(
"run_shell", {"command": "pytest -q"}, None
).allowed
listed = client.get("/v1/workspaces/trusted").json()["workspaces"]
assert [item["workspace"] for item in listed] == [str(proj.resolve())]
revoked = client.post(
"/v1/workspaces/trust",
json={"path": str(proj), "trusted": False},
).json()
assert revoked["ok"] and not revoked["trusted"]
after = engine.permissions.evaluate(
"run_shell", {"command": "pytest -q"}, None
)
assert not after.allowed and after.needs_user
manager.workspace_trust.set_trusted(proj, True)
proj.rename(tmp_path / "moved-project")
assert client.post(
"/v1/workspaces/trust",
json={"path": str(proj), "trusted": False},
).json()["ok"]
assert manager.trusted_workspaces() == []
def test_recent_workspaces_exclude_scratch_dirs(tmp_path):
# Scratch dirs get touched like any workspace, but must never show up as
# "recent projects" in the folder gate (owner call, 2026-07-03).