diff --git a/application/network/__init__.py b/application/network/__init__.py
new file mode 100644
index 0000000..d1abf7e
--- /dev/null
+++ b/application/network/__init__.py
@@ -0,0 +1,5 @@
+"""Application services for the "Block network" switch (Sandbox Security Layer)."""
+
+from .network_guard import NetworkBlockedError, bind, ensure_allowed, is_blocked, refusal
+
+__all__ = ["NetworkBlockedError", "bind", "ensure_allowed", "is_blocked", "refusal"]
diff --git a/application/network/network_guard.py b/application/network/network_guard.py
new file mode 100644
index 0000000..0d655f4
--- /dev/null
+++ b/application/network/network_guard.py
@@ -0,0 +1,61 @@
+"""One gate for every outbound connection the app makes on its own.
+
+The "Block network" switch (``agent_security.block_network``) used to be read
+only where agent tools run, so Microsoft 365, Teams, connector test buttons,
+scheduled task scripts, pip auto-installs and HTML previews still reached the
+internet while Monitoring said "Network: blocked". Each of those now asks
+this module first.
+
+The AI provider path (chat, model list, model test) deliberately does NOT go
+through here: with the switch on the user still talks to the model, but the
+app fetches nothing else on the model's or its own behalf.
+
+The module holds a *reader*, not a copy of the flag: the Composition Root
+binds it to the live config once (``presentation/shell/bootstrap.py``), so a
+change saved in Settings takes effect on the very next call. Nothing bound
+(unit tests, helper subprocesses) means "not blocked", the pre-switch default.
+"""
+from __future__ import annotations
+
+import threading
+from typing import Callable, Optional
+
+_lock = threading.Lock()
+_reader: Optional[Callable[[], bool]] = None
+
+
+class NetworkBlockedError(PermissionError):
+ """Raised by :func:`ensure_allowed` while the switch is on."""
+
+
+def bind(reader: Optional[Callable[[], bool]]) -> None:
+ """Install the callable that says whether the network is blocked right now."""
+ global _reader
+ with _lock:
+ _reader = reader
+
+
+def is_blocked() -> bool:
+ """True while "Block network" is on. A failing reader counts as blocked."""
+ reader = _reader
+ if reader is None:
+ return False
+ try:
+ return bool(reader())
+ except Exception: # noqa: BLE001 - fail closed: an unreadable switch must not open the network
+ return True
+
+
+def refusal(purpose: str) -> str:
+ """The message shown (to the user or the model) when ``purpose`` is refused."""
+ return (f"{purpose}: network access is blocked by the Sandbox Security Layer "
+ "(\"Block network\" is on in Settings). Only the AI provider may be reached.")
+
+
+def ensure_allowed(purpose: str) -> None:
+ """Raise :class:`NetworkBlockedError` if ``purpose`` may not go online now."""
+ if is_blocked():
+ raise NetworkBlockedError(refusal(purpose))
+
+
+__all__ = ["NetworkBlockedError", "bind", "ensure_allowed", "is_blocked", "refusal"]
diff --git a/application/workspaces/file_preview_helpers.py b/application/workspaces/file_preview_helpers.py
index 3f1c78f..868db5c 100644
--- a/application/workspaces/file_preview_helpers.py
+++ b/application/workspaces/file_preview_helpers.py
@@ -21,9 +21,14 @@ def pptx_available() -> bool:
try:
from cowork_local.core.deps import ensure_module
- _PPTX_READY = ensure_module("pptx", "python-pptx") is not None
+ ready = ensure_module("pptx", "python-pptx") is not None
except Exception: # noqa: BLE001
- _PPTX_READY = False
+ ready = False
+ from ..network import network_guard
+
+ if ready or not network_guard.is_blocked():
+ _PPTX_READY = ready # a refusal under "Block network" is retried later
+ return ready
return _PPTX_READY
diff --git a/config.py b/config.py
index e40be0c..b05951e 100644
--- a/config.py
+++ b/config.py
@@ -100,11 +100,11 @@ DEFAULT_CONFIG: Dict[str, Any] = {
"resource_limit_cpu_percent": 80, # 0 = unlimited; caps a run_command/install_package process TREE's total CPU%
"resource_limit_memory_mb": 2048, # 0 = unlimited; caps total RSS memory (MB)
"resource_limit_disk_mb": 512, # 0 = unlimited; caps total disk read+write (MB)
- # Cut the agent off the network: proxy env pointed at a black hole for
- # agent-run shell commands, PLUS a flat refusal from every tool tagged
- # ToolCapability.NETWORK (fetch_url, jira_*, install_package) — those
- # reach the net in-process, where the proxy trick has nothing to act on.
- "block_network": True,
+ # "Block network": every outbound connection except the AI provider is
+ # refused (application/network/network_guard.py), and agent shell
+ # commands / task scripts run in a network-less AppContainer.
+ # OFF on first launch — the user turns it on in Settings.
+ "block_network": False,
# Allow the agent's fetch_url tool to read web pages / online documents /
# SharePoint-OneDrive share links. Its own toggle — reading a URL for info
# is safe and useful, so this defaults ON — but block_network outranks it:
diff --git a/core/chat_agent.py b/core/chat_agent.py
index 1157059..4cdb452 100644
--- a/core/chat_agent.py
+++ b/core/chat_agent.py
@@ -527,7 +527,7 @@ def run_cowork(
preview = {"kind": "info", "title": name, "text": str(args)}
emit({"type": "tool_proposed", "id": tc_id, "name": name, "args": args,
"preview": preview})
- if ctx.block_network:
+ if ctx.block_network and not name.startswith("ms365_local__"):
result = {"ok": False, "output": (
f"{name}: network access is blocked by the Sandbox Security Layer "
'("Block network for agent-run commands" is on in Settings).')}
diff --git a/core/code_agent.py b/core/code_agent.py
index a31cdc3..f65259d 100644
--- a/core/code_agent.py
+++ b/core/code_agent.py
@@ -327,7 +327,7 @@ def run_code(
else:
emit({"type": "tool_start", "id": tc_id, "name": name})
if is_extra and extra_executor is not None:
- if ctx.block_network:
+ if ctx.block_network and not name.startswith("ms365_local__"):
result = {"ok": False, "output": (
f"{name}: network access is blocked by the Sandbox Security Layer "
'("Block network for agent-run commands" is on in Settings).')}
diff --git a/core/deps.py b/core/deps.py
index 8ec5d10..dfc99c7 100644
--- a/core/deps.py
+++ b/core/deps.py
@@ -91,6 +91,7 @@ def run_cancellable(
on_output: Optional[Callable[[str], None]] = None,
env: Optional[Dict[str, str]] = None,
limits: Optional[Dict[str, float]] = None,
+ isolate_network: bool = False,
) -> Tuple[Optional[int], str, bool, bool, bool]:
"""Run a subprocess so the Stop button can actually interrupt it.
@@ -117,17 +118,27 @@ def run_cancellable(
a failure to create/assign the job just means the existing taskkill
fallback is used, same as before this was added.
+ ``isolate_network`` runs ``args`` as a shell command that the OS keeps
+ off the network (see ``infrastructure/sandbox/network_isolation.py``);
+ if that isolation cannot be set up the command is NOT run.
+
Returns ``(returncode, combined_output, cancelled, timed_out,
resource_exceeded)``; on a failure to even launch the process,
``returncode`` is ``None`` and the output holds the launch error."""
cancel = cancel or (lambda: False)
popen_kwargs = {} if sys.platform == "win32" else {"start_new_session": True}
try:
- proc = subprocess.Popen(
- args, shell=shell, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
- text=True, bufsize=1, env=env, **popen_kwargs,
- )
- except OSError as exc:
+ if isolate_network:
+ from ..infrastructure.sandbox.network_isolation import spawn_without_network
+
+ command = args if isinstance(args, str) else subprocess.list2cmdline(args)
+ proc = spawn_without_network(command, cwd, env)
+ else:
+ proc = subprocess.Popen(
+ args, shell=shell, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+ text=True, bufsize=1, env=env, **popen_kwargs,
+ )
+ except (OSError, RuntimeError) as exc: # RuntimeError: NetworkIsolationUnavailable
return None, str(exc), False, False, False
with _active_pids_lock:
@@ -266,6 +277,10 @@ def ensure_module(module: str, package: str | None = None):
pkg = package or module
if pkg in _FAILED or not _can_pip():
return None
+ from ..application.network import network_guard
+
+ if network_guard.is_blocked():
+ return None # not cached in _FAILED: retried once the network is back
ok, _ = pip_install(pkg)
if not ok:
_FAILED.add(pkg)
@@ -339,6 +354,10 @@ def pip_install(package: str, cancel: Optional[CancelFn] = None,
name) is NOT retried, since repeating it would just waste time."""
if not _can_pip():
return False, "This packaged build can't install packages at runtime."
+ from ..application.network import network_guard
+
+ if network_guard.is_blocked():
+ return False, network_guard.refusal(f"pip install {package}")
exe = python or sys.executable
attempt = 0
while True:
diff --git a/core/ext_connectors.py b/core/ext_connectors.py
index ea2cd27..5d4890e 100644
--- a/core/ext_connectors.py
+++ b/core/ext_connectors.py
@@ -132,8 +132,11 @@ class RestApiConnector:
def call(self, args: Dict[str, Any]) -> Dict[str, Any]:
"""Gọi API theo tham số model đưa ra, đi qua lớp TLS có ghim chứng chỉ nội bộ."""
+ from ..application.network import network_guard
from .tls_trust import request_any_method as tls_request
+ if network_guard.is_blocked():
+ return {"ok": False, "output": network_guard.refusal(self.display_name)}
method = str(args.get("method", "GET")).upper()
path = str(args.get("path", "")).lstrip("/")
url = urljoin(self.base_url, path)
@@ -164,10 +167,13 @@ class RestApiConnector:
def test_connection(self) -> Tuple[bool, str]:
"""Thử kết nối tới endpoint; trả về (thành công, thông điệp)."""
+ from ..application.network import network_guard
from .tls_trust import request as tls_request
if not self.base_url.strip("/"):
return False, "No base URL configured."
+ if network_guard.is_blocked():
+ return False, network_guard.refusal(self.display_name)
headers = {}
if self.api_key:
headers[self.auth_header] = (
diff --git a/core/jira_tool.py b/core/jira_tool.py
index ef5db00..167724c 100644
--- a/core/jira_tool.py
+++ b/core/jira_tool.py
@@ -85,8 +85,10 @@ def get_issue_by_url(config: Dict[str, Any] | None, url: str) -> str:
def _get(config: Dict[str, Any], path: str, params: dict = None):
"""Gọi Jira REST API bằng xác thực cơ bản, qua lớp TLS có ghim chứng chỉ nội bộ."""
+ from ..application.network import network_guard
from . import tls_trust
+ network_guard.ensure_allowed("Jira")
c = _conf(config)
url = c["base_url"].rstrip("/") + path
# Same TLS auto-recovery the LLM provider calls get (core/tls_trust.py) —
diff --git a/core/link_fetch.py b/core/link_fetch.py
index 455e5b4..0344e31 100644
--- a/core/link_fetch.py
+++ b/core/link_fetch.py
@@ -147,6 +147,12 @@ def fetch_link_preview(url: str) -> str:
return ""
if not re.match(r"^https?://", url, re.IGNORECASE):
return f"[Link: {url}] (not a fetchable http(s) URL — referenced by address only)"
+ # Every caller (fetch_url, task link attachments, ...) passes through here,
+ # so this one check covers the paths that never saw a ToolContext.
+ from ..application.network import network_guard
+
+ if network_guard.is_blocked():
+ return f"[Link: {url}] (not fetched — {network_guard.refusal('link fetch')})"
# SharePoint / OneDrive share links are rewritten to their direct-download
# form so the shared FILE itself is fetched and parsed (like an attachment),
# not the share page's HTML shell.
diff --git a/core/mcp_client.py b/core/mcp_client.py
index aa4c9de..37bf990 100644
--- a/core/mcp_client.py
+++ b/core/mcp_client.py
@@ -88,7 +88,14 @@ class McpServerConnection:
def start(self, timeout: float = 15.0) -> None:
"""Spawn the server subprocess and complete the MCP handshake.
Raises :class:`McpServerError` on failure (bad command, the server
- crashed on startup, the handshake timed out, ...)."""
+ crashed on startup, the handshake timed out, ...).
+
+ Refused while "Block network" is on: a server process is free to open
+ any socket it likes, so the only safe server is one never started."""
+ from ..application.network import network_guard
+
+ if network_guard.is_blocked():
+ raise McpServerError(network_guard.refusal(f"MCP server '{self.name}'"))
self._thread = threading.Thread(target=self._run_loop, daemon=True)
self._thread.start()
if not self._ready.wait(timeout):
@@ -174,6 +181,10 @@ class McpServerConnection:
def call_tool(self, qualified_name: str, args: Dict[str, Any]) -> Dict[str, Any]:
"""``extra_executor``-shaped result: ``{"ok": bool, "output": str}``."""
+ from ..application.network import network_guard
+
+ if network_guard.is_blocked():
+ return {"ok": False, "output": network_guard.refusal(f"MCP server '{self.name}'")}
tool_name = qualified_name.split(_SEP, 1)[1] if _SEP in qualified_name else qualified_name
try:
result = self._run_coro(self._session.call_tool(tool_name, args or {}))
diff --git a/core/ms365_auth.py b/core/ms365_auth.py
index 693640e..0a7fd88 100644
--- a/core/ms365_auth.py
+++ b/core/ms365_auth.py
@@ -137,8 +137,34 @@ def _app(tenant_id: str, client_id: str):
return app, cache
+def _cached_account_offline() -> Optional[dict]:
+ """First account in the saved token cache, read without any MSAL network setup."""
+ try:
+ import msal
+
+ accounts = _load_cache().find(msal.TokenCache.CredentialType.ACCOUNT)
+ except Exception: # noqa: BLE001 - no msal / unreadable cache = not signed in
+ return None
+ return accounts[0] if accounts else None
+
+
+def _ensure_network(action: str) -> None:
+ """Turn a "Block network" refusal into the error type callers already handle."""
+ from ..application.network import network_guard
+
+ if network_guard.is_blocked():
+ raise Ms365AuthError(network_guard.refusal(action))
+
+
def signed_in_account(tenant_id: str, client_id: str) -> Optional[dict]:
"""The cached account, if any — a local cache lookup, no network call."""
+ from ..application.network import network_guard
+
+ if network_guard.is_blocked():
+ # Building the MSAL app fetches the tenant's OpenID configuration, so
+ # read the token cache directly instead: the UI still sees who is
+ # signed in without the app reaching login.microsoftonline.com.
+ return _cached_account_offline()
try:
app, _cache = _app(tenant_id, client_id)
except Ms365AuthError:
@@ -156,6 +182,7 @@ def sign_in_device_code(tenant_id: str, client_id: str, on_code: Callable[[dict]
``verification_uri_complete`` (URL with the code pre-filled, when the tenant
returns it) and ``message`` (the full human-readable instruction). Returns
the MSAL token result dict; raises Ms365AuthError on failure/timeout."""
+ _ensure_network("Microsoft 365 sign-in")
app, cache = _app(tenant_id, client_id)
flow = app.initiate_device_flow(scopes=SCOPES)
if "user_code" not in flow:
@@ -183,6 +210,7 @@ def get_access_token(tenant_id: str, client_id: str) -> str:
"""Silently reuse the cached sign-in. Raises Ms365AuthError when there is
no valid session — the caller (a Graph call) should surface that as a
normal tool failure telling the user to sign in again from Settings."""
+ _ensure_network("Microsoft 365")
app, cache = _app(tenant_id, client_id)
accounts = app.get_accounts()
if not accounts:
@@ -228,6 +256,7 @@ def sign_out_default(config=None) -> None:
def sign_out(tenant_id: str, client_id: str) -> None:
"""Đăng xuất và xoá token của một tenant/client khỏi kho."""
try:
+ _ensure_network("Microsoft 365 sign-out") # chặn mạng: chỉ xoá kho token bên dưới
app, cache = _app(tenant_id, client_id)
for acc in app.get_accounts():
app.remove_account(acc)
diff --git a/core/ms365_graph.py b/core/ms365_graph.py
index 402ae16..1b00012 100644
--- a/core/ms365_graph.py
+++ b/core/ms365_graph.py
@@ -43,6 +43,10 @@ def _request(method: str, url: str, token: str, **kwargs) -> requests.Response:
"""Gọi Graph API, tự ghép ``GRAPH_BASE`` cho đường dẫn tương đối và đổi lỗi HTTP
thành :class:`Ms365GraphError` kèm thông điệp đọc được.
"""
+ from ..application.network import network_guard
+
+ if network_guard.is_blocked():
+ raise Ms365GraphError(network_guard.refusal("Microsoft 365 (Graph)"))
if not url.startswith("http"):
url = f"{GRAPH_BASE}{url}"
headers = _headers(token, kwargs.pop("headers", None))
diff --git a/core/sandbox_manager.py b/core/sandbox_manager.py
index 6264106..0a4c665 100644
--- a/core/sandbox_manager.py
+++ b/core/sandbox_manager.py
@@ -218,8 +218,13 @@ class SandboxManager:
timeout_sec: int,
cancel: Optional[Callable[[], bool]] = None,
) -> Dict[str, Any]:
- """Dispatch execution to the selected backend."""
- if backend == "direct":
+ """Dispatch execution to the selected backend.
+
+ With the network blocked every backend is replaced by the same OS-level
+ isolation: the backends below only ever set proxy env vars, which
+ anything that ignores proxies (raw sockets, ping, .NET WebClient...)
+ walked straight past."""
+ if block_network or backend == "direct":
return self._run_direct(command, workdir, block_network, timeout_sec, cancel)
if backend == "integrity_job_wfp":
@@ -274,7 +279,8 @@ class SandboxManager:
env = os.environ.copy()
if block_network:
from .deps import network_blocked_env
- env = network_blocked_env(env)
+ env = network_blocked_env(env) # belt and braces on top of the OS block
+ return self._run_network_isolated(command, workdir, env, timeout_sec, cancel)
if cancel is not None:
from .deps import run_cancellable
@@ -334,4 +340,42 @@ class SandboxManager:
"stderr": str(exc),
"returncode": -1,
"sandbox": "direct",
- }
\ No newline at end of file
+ }
+
+ @staticmethod
+ def _run_network_isolated(
+ command: str,
+ workdir: str,
+ env: Dict[str, str],
+ timeout_sec: int,
+ cancel: Optional[Callable[[], bool]] = None,
+ ) -> Dict[str, Any]:
+ """Run ``command`` in a process the OS keeps off the network.
+
+ Fail-closed: when the isolation cannot be set up the command is
+ refused (``sandbox == "blocked"``), never run with the network open."""
+ from .deps import run_cancellable
+
+ rc, output, cancelled, timed_out, exceeded = run_cancellable(
+ command, cwd=workdir or None, timeout=timeout_sec, cancel=cancel,
+ shell=True, env=env, isolate_network=True,
+ )
+ if rc is None and not (cancelled or timed_out or exceeded):
+ return {"ok": False, "stdout": "", "returncode": -1, "sandbox": "blocked",
+ "stderr": ("Command refused: network is blocked and the command could "
+ f"not be isolated from the network ({output.strip()}).")}
+ if cancelled:
+ stderr = "Cancelled by user."
+ elif timed_out:
+ stderr = f"Timeout after {timeout_sec}s"
+ elif exceeded:
+ stderr = "Resource limit exceeded."
+ else:
+ stderr = ""
+ return {
+ "ok": rc == 0 and not (cancelled or timed_out or exceeded),
+ "stdout": output,
+ "stderr": stderr,
+ "returncode": rc if rc is not None else -1,
+ "sandbox": "network_isolated",
+ }
diff --git a/core/task_executors.py b/core/task_executors.py
index 57de9a9..1147f10 100644
--- a/core/task_executors.py
+++ b/core/task_executors.py
@@ -19,7 +19,6 @@ in Waiting Input), so executors here run with an auto gate.
"""
from __future__ import annotations
-import subprocess
import time
import uuid
from datetime import datetime
@@ -30,6 +29,7 @@ from . import agent_roles
from . import agent_security
from . import projects
from .permissions import PermissionGate
+from .task_script import run_script as _run_script
from .tasks import ARTIFACTS_DIR, resolve_input_text
from .tools import ToolContext
@@ -358,18 +358,6 @@ def _run_agent(ctx, task_type: str, prompt: str, out_dir: Path,
return _last_assistant_text(messages), timed_out(), incomplete
-def _run_script(command: str, out_dir: Path, timeout_sec: int) -> str:
- """Chạy một task kiểu script bằng shell trong thư mục kết quả, có hạn giờ."""
- if not command.strip():
- raise RuntimeError("Script task has no command configured.")
- proc = subprocess.run(command, shell=True, cwd=str(out_dir),
- capture_output=True, text=True, timeout=max(1, timeout_sec))
- output = (proc.stdout or "") + (("\n[stderr]\n" + proc.stderr) if proc.stderr else "")
- if proc.returncode != 0:
- raise RuntimeError(f"Script exited with code {proc.returncode}:\n{output[-2000:]}")
- return output
-
-
def execute_task(ctx, task: Dict[str, Any], run_id: str,
emit: Optional[EmitFn] = None, cancel: Optional[CancelFn] = None,
tasks_dir: Path = None) -> Dict[str, Any]:
diff --git a/core/task_script.py b/core/task_script.py
new file mode 100644
index 0000000..213ac59
--- /dev/null
+++ b/core/task_script.py
@@ -0,0 +1,47 @@
+"""Run a scheduled task of type ``script`` (tách khỏi ``task_executors.py``).
+
+Khi công tắc "Chặn mạng" đang bật, lệnh của task chạy trong tiến trình bị hệ
+điều hành cắt mạng — giống ``run_command`` của agent. Trước đây task script
+chạy thẳng bằng ``subprocess.run``, không sandbox, nên lên mạng tự do.
+"""
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+
+
+def run_script(command: str, out_dir: Path, timeout_sec: int) -> str:
+ """Chạy một task kiểu script bằng shell trong thư mục kết quả, có hạn giờ."""
+ if not command.strip():
+ raise RuntimeError("Script task has no command configured.")
+ from ..application.network import network_guard
+
+ if network_guard.is_blocked():
+ return _run_script_without_network(command, out_dir, timeout_sec)
+ proc = subprocess.run(command, shell=True, cwd=str(out_dir),
+ capture_output=True, text=True, timeout=max(1, timeout_sec))
+ output = (proc.stdout or "") + (("\n[stderr]\n" + proc.stderr) if proc.stderr else "")
+ if proc.returncode != 0:
+ raise RuntimeError(f"Script exited with code {proc.returncode}:\n{output[-2000:]}")
+ return output
+
+
+def _run_script_without_network(command: str, out_dir: Path, timeout_sec: int) -> str:
+ """Như :func:`run_script`, nhưng tiến trình không có mạng; không cô lập được thì không chạy."""
+ from .deps import network_blocked_env, run_cancellable
+
+ rc, output, _cancelled, timed_out, _exceeded = run_cancellable(
+ command, cwd=str(out_dir), timeout=max(1, timeout_sec), shell=True,
+ env=network_blocked_env(), isolate_network=True,
+ )
+ if timed_out:
+ raise subprocess.TimeoutExpired(command, timeout_sec)
+ if rc is None:
+ raise RuntimeError("Script not run: network is blocked and the script could not be "
+ f"isolated from the network ({output.strip()}).")
+ if rc != 0:
+ raise RuntimeError(f"Script exited with code {rc} (network blocked):\n{output[-2000:]}")
+ return output
+
+
+__all__ = ["run_script"]
diff --git a/core/teams.py b/core/teams.py
index fd3fe92..578cc0e 100644
--- a/core/teams.py
+++ b/core/teams.py
@@ -43,6 +43,10 @@ class TeamsNotifier:
"""Post a notification. Returns ``(ok, detail)``."""
if not self.configured:
return False, "Teams webhook URL is not configured."
+ from ..application.network import network_guard
+
+ if network_guard.is_blocked():
+ return False, network_guard.refusal("Teams notification")
# Workflows webhooks expect an Adaptive Card; classic connectors expect a
# MessageCard. Try both, then a plain-text fallback.
diff --git a/i18n/settings_dialog.py b/i18n/settings_dialog.py
index 2544101..6ffb9d3 100644
--- a/i18n/settings_dialog.py
+++ b/i18n/settings_dialog.py
@@ -10,22 +10,22 @@ from typing import Dict
STRINGS: Dict[str, Dict[str, str]] = {
"settings.sandbox_block_network": {
- "en": "Block network for agent-run commands",
- "ja": "エージェントが実行するコマンドのネットワークをブロック",
- "vi": "Chặn mạng cho lệnh do agent chạy"},
+ "en": "Block network (AI provider still allowed)",
+ "ja": "ネットワークをブロック（AIプロバイダーのみ許可）",
+ "vi": "Chặn mạng (vẫn cho gọi nhà cung cấp AI)"},
"settings.allow_url_fetch": {
"en": "Allow the agent to fetch URLs (web pages, SharePoint / OneDrive links)",
"ja": "エージェントによるURL取得を許可（Webページ、SharePoint / OneDriveリンク）",
"vi": "Cho phép agent lấy dữ liệu từ URL (trang web, link SharePoint / OneDrive)"},
"settings.allow_url_fetch_tooltip": {
"en": ("Lets the agent's fetch_url tool read web pages, online documents and "
- "SharePoint/OneDrive share links to search & process them. Separate from "
- "'Block network' (which only sandboxes shell commands). Default: on."),
+ "SharePoint/OneDrive share links to search & process them. 'Block network' "
+ "overrides this switch. Default: on."),
"ja": "エージェントのfetch_urlツールがWebページ・オンライン文書・SharePoint/OneDrive共有リンクを"
- "読み取れるようにします。「ネットワークをブロック」（シェルコマンド用）とは別です。既定: オン。",
+ "読み取れるようにします。「ネットワークをブロック」がオンの場合はそちらが優先されます。既定: オン。",
"vi": ("Cho phép tool fetch_url của agent đọc trang web, tài liệu online và link chia sẻ "
- "SharePoint/OneDrive để tìm kiếm & xử lý. Tách biệt với 'Chặn mạng' (chỉ áp cho lệnh "
- "shell). Mặc định: bật.")},
+ "SharePoint/OneDrive để tìm kiếm & xử lý. 'Chặn mạng' được ưu tiên hơn công tắc "
+ "này. Mặc định: bật.")},
"settings.test_internet": {
"en": "Test Internet", "ja": "インターネット接続テスト", "vi": "Kiểm tra Internet"},
"settings.test_internet_tooltip": {
@@ -39,12 +39,18 @@ STRINGS: Dict[str, Dict[str, str]] = {
"en": "Testing internet access…", "ja": "インターネット接続をテスト中…",
"vi": "Đang kiểm tra truy cập internet…"},
"settings.sandbox_block_network_tooltip": {
- "en": ("Policy-level control (proxy env vars point at a black hole) — not a kernel "
- "firewall. Combine with the command whitelist above for defense in depth."),
- "ja": "ポリシーレベルの制御です（プロキシ環境変数をブラックホールに向ける）— カーネルレベルの"
- "ファイアウォールではありません。上のコマンドホワイトリストと併用してください。",
- "vi": "Kiểm soát ở tầng chính sách (trỏ biến môi trường proxy vào hố đen) — không phải "
- "firewall tầng kernel. Kết hợp với whitelist lệnh ở trên để phòng thủ nhiều lớp."},
+ "en": ("Only the AI provider (chat, model list, model test) may reach the network. "
+ "Agent shell commands and task scripts run in a Windows AppContainer with no "
+ "network access; fetch_url, Jira, connectors/MCP, Microsoft 365, Teams, "
+ "connector tests, pip auto-install and remote content in HTML previews are refused."),
+ "ja": "ネットワークに接続できるのはAIプロバイダー（チャット・モデル一覧・モデルテスト）のみです。"
+ "エージェントのシェルコマンドとタスクスクリプトはネットワークなしのWindows AppContainerで実行され、"
+ "fetch_url・Jira・コネクタ/MCP・Microsoft 365・Teams・接続テスト・pip自動インストール・"
+ "HTMLプレビューの外部リソースは拒否されます。",
+ "vi": "Chỉ nhà cung cấp AI (chat, tải danh sách model, thử model) được ra mạng. Lệnh shell "
+ "của agent và task script chạy trong Windows AppContainer không có mạng; fetch_url, "
+ "Jira, connector/MCP, Microsoft 365, Teams, nút Test, tự cài thư viện và tài nguyên "
+ "web trong xem trước HTML đều bị từ chối."},
"settings.sandbox_unlimited": {"en": "Unlimited", "ja": "無制限", "vi": "Không giới hạn"},
"settings.sandbox_cpu_label": {"en": "CPU limit", "ja": "CPU 制限", "vi": "Giới hạn CPU"},
"settings.sandbox_memory_label": {"en": "Memory limit", "ja": "メモリ制限", "vi": "Giới hạn bộ nhớ"},
diff --git a/infrastructure/filesystem/command_tools.py b/infrastructure/filesystem/command_tools.py
index 8d88e40..06f6a39 100644
--- a/infrastructure/filesystem/command_tools.py
+++ b/infrastructure/filesystem/command_tools.py
@@ -76,11 +76,11 @@ def run_command(ctx: ToolContext, args: Dict[str, Any],
denial = "Command blocked by security policy: " + "; ".join(risk.reasons)
return {"ok": False, "output": denial}
- # Every sandbox backend's network block is a proxy-env-var trick (see
- # core/deps.py::network_blocked_env) — it does nothing against a tool
- # that reaches the network without an HTTP proxy (ping/ICMP, nslookup/
- # direct DNS, ssh/ftp/raw TCP...). Deny those BY NAME here instead, so
- # "Chặn mạng cho lệnh do agent chạy" actually blocks them too.
+ # With the network blocked, SandboxManager runs the command in an OS-level
+ # network-less process (AppContainer on Windows — see
+ # infrastructure/sandbox/network_isolation.py). Tools that exist only to
+ # reach the network (ping, nslookup, ssh...) are still denied BY NAME
+ # first: the model gets a clear reason instead of a cryptic socket error.
if ctx.block_network:
bypass_tool = command_bypasses_network_proxy(command)
if bypass_tool:
diff --git a/infrastructure/sandbox/appcontainer_process.py b/infrastructure/sandbox/appcontainer_process.py
new file mode 100644
index 0000000..04a8e61
--- /dev/null
+++ b/infrastructure/sandbox/appcontainer_process.py
@@ -0,0 +1,392 @@
+"""Spawn a shell command inside a Windows AppContainer with NO network capability.
+
+Why this exists
+---------------
+The old "block network" for agent shell commands only pointed the proxy env
+vars at a dead port (``core/deps.py::network_blocked_env``). Anything that
+ignores proxies (``Invoke-WebRequest -NoProxy``, raw sockets, ``certutil``,
+.NET ``WebClient``...) still reached the internet. An AppContainer token that
+is granted no ``internetClient``/``privateNetworkClientServer`` capability is
+refused by the kernel firewall for every outbound connection, loopback
+included, no matter which tool makes it. No admin rights are needed.
+
+An AppContainer can only open files whose ACL admits its SID (or ALL
+APPLICATION PACKAGES). System32 and Program Files already do; the workdir and
+the app's own Python install do not, so :func:`spawn` grants the profile SID
+access to those folders first (an extra ACE, nothing is removed).
+
+:class:`AppContainerProcess` quacks like ``subprocess.Popen`` for the subset
+``core/deps.py::_run_cancellable_body`` uses (``pid``, ``stdout``/``stderr``
+text streams, ``poll``/``wait``/``kill``/``returncode``), so the Stop button,
+timeouts, Job Objects and resource limits keep working unchanged.
+"""
+from __future__ import annotations
+
+import io
+import locale
+import os
+import subprocess
+import sys
+import threading
+from typing import Dict, Iterable, Optional
+
+PROFILE_NAME = "cowork_local.agent_netblock"
+_IS_WINDOWS = sys.platform == "win32"
+
+if _IS_WINDOWS:
+ import ctypes
+ import msvcrt
+ from ctypes import wintypes
+
+ _k32 = ctypes.WinDLL("kernel32", use_last_error=True)
+ _adv = ctypes.WinDLL("advapi32", use_last_error=True)
+ _uenv = ctypes.WinDLL("userenv", use_last_error=True)
+
+ class _SECURITY_CAPABILITIES(ctypes.Structure):
+ _fields_ = [("AppContainerSid", ctypes.c_void_p), ("Capabilities", ctypes.c_void_p),
+ ("CapabilityCount", wintypes.DWORD), ("Reserved", wintypes.DWORD)]
+
+ class _STARTUPINFOW(ctypes.Structure):
+ _fields_ = [("cb", wintypes.DWORD), ("lpReserved", wintypes.LPWSTR),
+ ("lpDesktop", wintypes.LPWSTR), ("lpTitle", wintypes.LPWSTR),
+ ("dwX", wintypes.DWORD), ("dwY", wintypes.DWORD),
+ ("dwXSize", wintypes.DWORD), ("dwYSize", wintypes.DWORD),
+ ("dwXCountChars", wintypes.DWORD), ("dwYCountChars", wintypes.DWORD),
+ ("dwFillAttribute", wintypes.DWORD), ("dwFlags", wintypes.DWORD),
+ ("wShowWindow", wintypes.WORD), ("cbReserved2", wintypes.WORD),
+ ("lpReserved2", ctypes.c_void_p), ("hStdInput", wintypes.HANDLE),
+ ("hStdOutput", wintypes.HANDLE), ("hStdError", wintypes.HANDLE)]
+
+ class _STARTUPINFOEXW(ctypes.Structure):
+ _fields_ = [("StartupInfo", _STARTUPINFOW), ("lpAttributeList", ctypes.c_void_p)]
+
+ class _PROCESS_INFORMATION(ctypes.Structure):
+ _fields_ = [("hProcess", wintypes.HANDLE), ("hThread", wintypes.HANDLE),
+ ("dwProcessId", wintypes.DWORD), ("dwThreadId", wintypes.DWORD)]
+
+ class _SECURITY_ATTRIBUTES(ctypes.Structure):
+ _fields_ = [("nLength", wintypes.DWORD), ("lpSecurityDescriptor", ctypes.c_void_p),
+ ("bInheritHandle", wintypes.BOOL)]
+
+ _uenv.CreateAppContainerProfile.restype = ctypes.c_long
+ _uenv.CreateAppContainerProfile.argtypes = [
+ wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.LPCWSTR, ctypes.c_void_p,
+ wintypes.DWORD, ctypes.POINTER(ctypes.c_void_p)]
+ _uenv.DeriveAppContainerSidFromAppContainerName.restype = ctypes.c_long
+ _uenv.DeriveAppContainerSidFromAppContainerName.argtypes = [
+ wintypes.LPCWSTR, ctypes.POINTER(ctypes.c_void_p)]
+ _uenv.GetAppContainerFolderPath.restype = ctypes.c_long
+ _uenv.GetAppContainerFolderPath.argtypes = [
+ wintypes.LPCWSTR, ctypes.POINTER(ctypes.c_wchar_p)]
+ _adv.ConvertSidToStringSidW.restype = wintypes.BOOL
+ _adv.ConvertSidToStringSidW.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_wchar_p)]
+ _k32.InitializeProcThreadAttributeList.restype = wintypes.BOOL
+ _k32.InitializeProcThreadAttributeList.argtypes = [
+ ctypes.c_void_p, wintypes.DWORD, wintypes.DWORD, ctypes.POINTER(ctypes.c_size_t)]
+ _k32.UpdateProcThreadAttribute.restype = wintypes.BOOL
+ _k32.UpdateProcThreadAttribute.argtypes = [
+ ctypes.c_void_p, wintypes.DWORD, ctypes.c_size_t, ctypes.c_void_p,
+ ctypes.c_size_t, ctypes.c_void_p, ctypes.c_void_p]
+ _k32.DeleteProcThreadAttributeList.argtypes = [ctypes.c_void_p]
+ _k32.CreatePipe.restype = wintypes.BOOL
+ _k32.CreatePipe.argtypes = [ctypes.POINTER(wintypes.HANDLE), ctypes.POINTER(wintypes.HANDLE),
+ ctypes.POINTER(_SECURITY_ATTRIBUTES), wintypes.DWORD]
+ _k32.SetHandleInformation.argtypes = [wintypes.HANDLE, wintypes.DWORD, wintypes.DWORD]
+ _k32.CreateProcessW.restype = wintypes.BOOL
+ _k32.CreateProcessW.argtypes = [
+ wintypes.LPCWSTR, wintypes.LPWSTR, ctypes.c_void_p, ctypes.c_void_p, wintypes.BOOL,
+ wintypes.DWORD, ctypes.c_void_p, wintypes.LPCWSTR, ctypes.POINTER(_STARTUPINFOEXW),
+ ctypes.POINTER(_PROCESS_INFORMATION)]
+ _k32.ResumeThread.argtypes = [wintypes.HANDLE]
+ _k32.WaitForSingleObject.restype = wintypes.DWORD
+ _k32.WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD]
+ _k32.GetExitCodeProcess.argtypes = [wintypes.HANDLE, ctypes.POINTER(wintypes.DWORD)]
+ _k32.TerminateProcess.argtypes = [wintypes.HANDLE, wintypes.UINT]
+ _k32.CloseHandle.argtypes = [wintypes.HANDLE]
+ _k32.GetStdHandle.restype = wintypes.HANDLE
+ _k32.OpenProcess.restype = wintypes.HANDLE
+
+ _ALREADY_EXISTS = ctypes.c_long(0x800700B7).value
+ _PROC_THREAD_ATTRIBUTE_HANDLE_LIST = 0x00020002
+ _PROC_THREAD_ATTRIBUTE_SECURITY_CAPABILITIES = 0x00020009
+ _EXTENDED_STARTUPINFO_PRESENT = 0x00080000
+ _CREATE_UNICODE_ENVIRONMENT = 0x00000400
+ _CREATE_NO_WINDOW = 0x08000000
+ _CREATE_SUSPENDED = 0x00000004
+ _STARTF_USESTDHANDLES = 0x00000100
+ _HANDLE_FLAG_INHERIT = 0x00000001
+ _STILL_ACTIVE = 259
+
+_profile_lock = threading.Lock()
+_profile: Dict[str, object] = {}
+_granted: set = set()
+
+
+class NetworkIsolationUnavailable(RuntimeError):
+ """This machine cannot start a network-less process — callers must refuse to run."""
+
+
+def is_supported() -> bool:
+ """True on Windows builds that ship the AppContainer API (Windows 8+)."""
+ if not _IS_WINDOWS:
+ return False
+ try:
+ return bool(_uenv.CreateAppContainerProfile)
+ except AttributeError:
+ return False
+
+
+def _profile_sid():
+ """``(sid pointer, sid string, temp folder)`` of the shared no-network profile."""
+ with _profile_lock:
+ if _profile:
+ return _profile["sid"], _profile["sid_str"], _profile["temp"]
+ sid = ctypes.c_void_p()
+ hr = _uenv.CreateAppContainerProfile(PROFILE_NAME, "Cowork Local agent (no network)",
+ "Agent shell commands with the network blocked",
+ None, 0, ctypes.byref(sid))
+ if hr == _ALREADY_EXISTS:
+ hr = _uenv.DeriveAppContainerSidFromAppContainerName(PROFILE_NAME, ctypes.byref(sid))
+ if hr != 0 or not sid.value:
+ raise NetworkIsolationUnavailable(f"AppContainer profile error 0x{hr & 0xFFFFFFFF:08X}")
+ text = ctypes.c_wchar_p()
+ if not _adv.ConvertSidToStringSidW(sid, ctypes.byref(text)):
+ raise NetworkIsolationUnavailable("Could not read the AppContainer SID")
+ sid_str = text.value
+ folder = ctypes.c_wchar_p()
+ temp = ""
+ if _uenv.GetAppContainerFolderPath(sid_str, ctypes.byref(folder)) == 0 and folder.value:
+ temp = os.path.join(folder.value, "Temp")
+ os.makedirs(temp, exist_ok=True)
+ _profile.update(sid=sid, sid_str=sid_str, temp=temp)
+ return sid, sid_str, temp
+
+
+_PERMS = {"write": "(OI)(CI)(M)", "read": "(OI)(CI)(RX)", "read_here": "(OI)(NP)(RX)"}
+
+
+def _qt_package_dir() -> str:
+ """Folder of the installed PySide6/Qt binaries ('' if PySide6 is absent)."""
+ try:
+ import importlib.util
+
+ spec = importlib.util.find_spec("PySide6")
+ except (ImportError, ValueError):
+ return ""
+ return os.path.dirname(spec.origin) if spec and spec.origin else ""
+
+
+def _covers(folder: str, target: str) -> bool:
+ """True if ``target`` is ``folder`` itself or lies somewhere below it."""
+ if not folder or not target:
+ return False
+ folder, target = os.path.normcase(folder), os.path.normcase(target)
+ return target == folder or target.startswith(folder.rstrip("\\/") + os.sep)
+
+
+def _icacls(*args: str) -> subprocess.CompletedProcess:
+ return subprocess.run(["icacls", *args], capture_output=True, text=True,
+ creationflags=_CREATE_NO_WINDOW)
+
+
+def grant_access(path: str, sid_str: str, mode: str) -> None:
+ """Let the AppContainer SID open ``path``.
+
+ ``mode`` is ``"write"``/``"read"`` (inherited by everything below) or
+ ``"read_here"`` (this folder and the files directly in it, no deeper).
+
+ An inherited ACE must never reach the Qt WebEngine binaries: Chromium's
+ sandboxed render process then fails to load Qt6WebEngineCore.dll
+ (STATUS_DLL_NOT_FOUND) and every web view in the app goes blank. A folder
+ that contains the PySide6 install is therefore refused.
+ """
+ path = os.path.abspath(path)
+ key = (os.path.normcase(path), mode)
+ if key in _granted or not os.path.exists(path):
+ return
+ if mode != "read_here" and _covers(path, _qt_package_dir()):
+ raise NetworkIsolationUnavailable(
+ f"Refusing to sandbox a folder that contains the app's Qt runtime: {path}")
+ perm = _PERMS[mode]
+ listing = (_icacls(path).stdout or "").lower()
+ if f"{sid_str}:{perm}".lower() not in listing:
+ done = _icacls(path, "/grant", f"*{sid_str}:{perm}", "/Q", "/C")
+ if done.returncode != 0:
+ raise NetworkIsolationUnavailable(
+ f"Could not grant the sandbox access to {path}: {(done.stderr or done.stdout).strip()}")
+ _granted.add(key)
+
+
+def _repair_inherited_grant(path: str, sid_str: str) -> None:
+ """Drop an inherited grant that an earlier build put on the app's venv."""
+ key = (os.path.normcase(os.path.abspath(path)), "repaired")
+ if key in _granted or not os.path.isdir(path):
+ return
+ listing = (_icacls(path).stdout or "").lower()
+ if f"{sid_str}:(oi)(ci)".lower() in listing:
+ _icacls(path, "/remove:g", f"*{sid_str}", "/Q", "/C")
+ _granted.add(key)
+
+
+def _interpreter_grants():
+ """``(folder, mode)`` pairs that let the sandbox run the app's Python.
+
+ The base install is read-only and holds no Qt; a venv only needs its root
+ (``pyvenv.cfg``) and ``Scripts`` — never ``Lib/site-packages``.
+ """
+ qt_dir = _qt_package_dir()
+ if _covers(sys.base_prefix, qt_dir):
+ # PySide6 lives in the base install itself: expose only the executable
+ # and the compiled stdlib modules, never the tree holding Qt.
+ grants = [(sys.base_prefix, "read_here"), (os.path.join(sys.base_prefix, "DLLs"), "read")]
+ else:
+ grants = [(sys.base_prefix, "read")]
+ if os.path.normcase(sys.prefix) != os.path.normcase(sys.base_prefix):
+ grants += [(sys.prefix, "read_here"), (os.path.join(sys.prefix, "Scripts"), "read")]
+ return [(folder, mode) for folder, mode in grants
+ if mode == "read_here" or not _covers(folder, qt_dir)]
+
+
+def _env_block(env: Dict[str, str]) -> ctypes.Array:
+ """Sorted, double-NUL-terminated UTF-16 environment block for CreateProcessW."""
+ items = sorted(env.items(), key=lambda kv: kv[0].upper())
+ text = "".join(f"{k}={v}\0" for k, v in items if k and "=" not in k) + "\0"
+ return ctypes.create_unicode_buffer(text, len(text))
+
+
+def _pipe():
+ """Anonymous pipe; only the child's (write) end is inheritable."""
+ sa = _SECURITY_ATTRIBUTES(ctypes.sizeof(_SECURITY_ATTRIBUTES), None, True)
+ read, write = wintypes.HANDLE(), wintypes.HANDLE()
+ if not _k32.CreatePipe(ctypes.byref(read), ctypes.byref(write), ctypes.byref(sa), 0):
+ raise ctypes.WinError(ctypes.get_last_error())
+ _k32.SetHandleInformation(read, _HANDLE_FLAG_INHERIT, 0)
+ return read, write
+
+
+class AppContainerProcess:
+ """The ``subprocess.Popen`` subset that ``deps._run_cancellable_body`` relies on."""
+
+ def __init__(self, handle, pid: int, stdout: io.TextIOBase, stderr: io.TextIOBase):
+ """Wrap an already-started process handle and its two output streams."""
+ self._handle = handle
+ self.pid = pid
+ self.stdout = stdout
+ self.stderr = stderr
+ self.returncode: Optional[int] = None
+
+ def poll(self) -> Optional[int]:
+ """Exit code if the process has finished, else None."""
+ if self.returncode is None and self._handle:
+ code = wintypes.DWORD()
+ if _k32.GetExitCodeProcess(self._handle, ctypes.byref(code)) and code.value != _STILL_ACTIVE:
+ self.returncode = ctypes.c_int32(code.value).value
+ _k32.CloseHandle(self._handle)
+ self._handle = None
+ return self.returncode
+
+ def wait(self, timeout: Optional[float] = None) -> int:
+ """Block until exit; raise ``subprocess.TimeoutExpired`` like Popen does."""
+ if self.returncode is None and self._handle:
+ ms = 0xFFFFFFFF if timeout is None else int(timeout * 1000)
+ if _k32.WaitForSingleObject(self._handle, ms) != 0:
+ raise subprocess.TimeoutExpired("appcontainer", timeout)
+ return self.poll()
+
+ def kill(self) -> None:
+ """Terminate the process (the Job Object in deps kills its children)."""
+ if self.returncode is None and self._handle:
+ _k32.TerminateProcess(self._handle, 1)
+
+ def communicate(self, timeout: Optional[float] = None):
+ """Read both streams to the end and wait; returns ``(stdout, stderr)``."""
+ chunks: Dict[str, str] = {}
+
+ def _drain(name, stream):
+ chunks[name] = stream.read()
+
+ readers = [threading.Thread(target=_drain, args=(n, s), daemon=True)
+ for n, s in (("out", self.stdout), ("err", self.stderr))]
+ for t in readers:
+ t.start()
+ for t in readers:
+ t.join(timeout)
+ self.wait(timeout)
+ return chunks.get("out", ""), chunks.get("err", "")
+
+
+def spawn(command: str, cwd: Optional[str], env: Optional[Dict[str, str]],
+ readable_dirs: Iterable[str] = ()) -> AppContainerProcess:
+ """Start ``cmd.exe /c command`` in the no-network AppContainer.
+
+ Raises :class:`NetworkIsolationUnavailable` when that cannot be done —
+ callers must then refuse the command rather than run it with the network on.
+ """
+ if not is_supported():
+ raise NetworkIsolationUnavailable("AppContainer is only available on Windows")
+ sid, sid_str, temp = _profile_sid()
+ cwd = os.path.abspath(cwd or os.getcwd())
+ _repair_inherited_grant(sys.prefix, sid_str)
+ grant_access(cwd, sid_str, "write")
+ for folder, mode in [*_interpreter_grants(), *((d, "read") for d in readable_dirs if d)]:
+ grant_access(folder, sid_str, mode)
+
+ child_env = dict(os.environ if env is None else env)
+ if temp:
+ child_env["TEMP"] = child_env["TMP"] = temp
+ child_env.setdefault("PYTHONIOENCODING", "utf-8")
+ env_block = _env_block(child_env)
+
+ out_r, out_w = _pipe()
+ err_r, err_w = _pipe()
+ handles = (wintypes.HANDLE * 2)(out_w, err_w)
+ caps = _SECURITY_CAPABILITIES(sid, None, 0, 0)
+ size = ctypes.c_size_t()
+ _k32.InitializeProcThreadAttributeList(None, 2, 0, ctypes.byref(size))
+ attr = ctypes.create_string_buffer(size.value)
+ pi = _PROCESS_INFORMATION()
+ try:
+ if not (_k32.InitializeProcThreadAttributeList(attr, 2, 0, ctypes.byref(size))
+ and _k32.UpdateProcThreadAttribute(
+ attr, 0, _PROC_THREAD_ATTRIBUTE_SECURITY_CAPABILITIES, ctypes.byref(caps),
+ ctypes.sizeof(caps), None, None)
+ and _k32.UpdateProcThreadAttribute(
+ attr, 0, _PROC_THREAD_ATTRIBUTE_HANDLE_LIST, handles,
+ ctypes.sizeof(handles), None, None)):
+ raise NetworkIsolationUnavailable(str(ctypes.WinError(ctypes.get_last_error())))
+ si = _STARTUPINFOEXW()
+ si.StartupInfo.cb = ctypes.sizeof(si)
+ si.StartupInfo.dwFlags = _STARTF_USESTDHANDLES
+ si.StartupInfo.hStdOutput = out_w
+ si.StartupInfo.hStdError = err_w
+ si.lpAttributeList = ctypes.addressof(attr)
+ comspec = os.environ.get("COMSPEC") or r"C:\Windows\System32\cmd.exe"
+ cmdline = ctypes.create_unicode_buffer(f'"{comspec}" /d /s /c "{command}"')
+ flags = (_EXTENDED_STARTUPINFO_PRESENT | _CREATE_UNICODE_ENVIRONMENT
+ | _CREATE_NO_WINDOW | _CREATE_SUSPENDED)
+ if not _k32.CreateProcessW(None, cmdline, None, None, True, flags,
+ ctypes.addressof(env_block), cwd,
+ ctypes.byref(si), ctypes.byref(pi)):
+ raise NetworkIsolationUnavailable(
+ f"Could not start the sandboxed command: {ctypes.WinError(ctypes.get_last_error())}")
+ _k32.ResumeThread(pi.hThread)
+ _k32.CloseHandle(pi.hThread)
+ except BaseException:
+ for h in (out_r, err_r):
+ _k32.CloseHandle(h)
+ raise
+ finally:
+ _k32.DeleteProcThreadAttributeList(attr)
+ _k32.CloseHandle(out_w)
+ _k32.CloseHandle(err_w)
+
+ def _stream(handle) -> io.TextIOBase:
+ fd = msvcrt.open_osfhandle(handle.value, os.O_RDONLY)
+ return io.TextIOWrapper(io.FileIO(fd, "rb"), encoding=locale.getpreferredencoding(False),
+ errors="replace")
+
+ return AppContainerProcess(pi.hProcess, pi.dwProcessId, _stream(out_r), _stream(err_r))
+
+
+__all__ = ["AppContainerProcess", "NetworkIsolationUnavailable", "PROFILE_NAME",
+ "grant_access", "is_supported", "spawn"]
diff --git a/infrastructure/sandbox/network_isolation.py b/infrastructure/sandbox/network_isolation.py
new file mode 100644
index 0000000..210e7d5
--- /dev/null
+++ b/infrastructure/sandbox/network_isolation.py
@@ -0,0 +1,46 @@
+"""Start a shell command that the operating system keeps off the network.
+
+Used whenever "Block network" is on for agent ``run_command`` calls and for
+scheduled script tasks. The contract is fail-closed: if isolation cannot be
+set up, :class:`NetworkIsolationUnavailable` is raised and the caller refuses
+the command instead of running it with the network open.
+
+* Windows: an AppContainer with no network capability
+ (:mod:`.appcontainer_process`).
+* macOS: ``sandbox-exec`` with a profile that denies every network operation.
+* Linux: ``unshare --net`` in a new user namespace (an empty network namespace
+ has only a downed loopback). If unprivileged namespaces are disabled,
+ ``unshare`` itself fails and the command never runs.
+"""
+from __future__ import annotations
+
+import shutil
+import subprocess
+import sys
+from typing import Dict, Optional
+
+from .appcontainer_process import NetworkIsolationUnavailable
+
+_MACOS_PROFILE = "(version 1)(allow default)(deny network*)"
+
+
+def spawn_without_network(command: str, cwd: Optional[str], env: Optional[Dict[str, str]]):
+ """A ``Popen``-like process running ``command`` through the shell, with no network."""
+ if sys.platform == "win32":
+ from . import appcontainer_process
+
+ return appcontainer_process.spawn(command, cwd, env)
+ if sys.platform == "darwin" and shutil.which("sandbox-exec"):
+ argv = ["sandbox-exec", "-p", _MACOS_PROFILE, "/bin/sh", "-c", command]
+ elif sys.platform.startswith("linux") and shutil.which("unshare"):
+ argv = ["unshare", "--user", "--map-root-user", "--net", "/bin/sh", "-c", command]
+ else:
+ raise NetworkIsolationUnavailable(
+ "No network isolation is available on this system (needs AppContainer, "
+ "sandbox-exec or unshare).")
+ return subprocess.Popen(argv, cwd=cwd, env=env, stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE, text=True, bufsize=1,
+ start_new_session=True)
+
+
+__all__ = ["NetworkIsolationUnavailable", "spawn_without_network"]
diff --git a/presentation/folder/office_document_renderer.py b/presentation/folder/office_document_renderer.py
index ea6985d..3aef705 100644
--- a/presentation/folder/office_document_renderer.py
+++ b/presentation/folder/office_document_renderer.py
@@ -110,7 +110,10 @@ class OfficeDocumentRenderer:
if self._engine is None:
try:
from PySide6.QtWebEngineWidgets import QWebEngineView
+
+ from .offline_web_page import install_offline_page
self._engine = QWebEngineView()
+ install_offline_page(self._engine)
self._owner.stack.addWidget(self._engine)
except Exception: # noqa: BLE001
self._engine = None
diff --git a/presentation/folder/offline_web_page.py b/presentation/folder/offline_web_page.py
new file mode 100644
index 0000000..44c96ee
--- /dev/null
+++ b/presentation/folder/offline_web_page.py
@@ -0,0 +1,39 @@
+"""HTML preview page that loads nothing from the web while "Block network" is on.
+
+``QWebEngineView.setHtml`` happily fetches every ``
``,
+``