fix(sandbox): "Chặn mạng" chặn mọi đường ra mạng, trừ nhà cung cấp AI

Trước đây công tắc chỉ chặn tool mạng của agent; lệnh shell chỉ bị proxy
giả, còn M365, Teams, nút Test, MCP đang chạy, task script, link đính kèm
task, pip tự cài và tài nguyên web trong xem trước HTML vẫn ra mạng tự do.

- Cổng chung application/network/network_guard.py, nối vào cấu hình sống
  ở Composition Root; nhà cung cấp AI (chat, danh sách model, thử model)
  không đi qua cổng này.
- Lệnh shell của agent và task script chạy trong Windows AppContainer
  không có quyền mạng (macOS: sandbox-exec, Linux: unshare --net);
  không cô lập được thì từ chối chạy.
- Không cấp quyền kế thừa của AppContainer lên thư mục chứa PySide6:
  Chromium không nạp được Qt6WebEngineCore.dll và tab Graph bị hỏng.
- Bật chặn thì dừng MCP đang chạy; tool OneDrive đồng bộ trên máy vẫn dùng.
- Mặc định tắt khi mở app lần đầu; nhãn và tooltip 3 ngôn ngữ cập nhật.
- Test: tests/test_network_guard_lanes.py (có bài AppContainer thật).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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"]

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"]

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
 
 

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:

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).')}

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).')}

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:
+        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 as exc:
+    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:

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] = (

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) —

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.

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 {}))

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)

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))

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
@@ -335,3 +341,41 @@ class SandboxManager:
                 "returncode": -1,
                 "sandbox": "direct",
             }
+
+    @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",
+        }

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]:

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"]

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.

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ớ"},

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:

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"]

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"]

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

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 ``<img src="https://...">``,
+``<script src>`` and stylesheet the previewed file references — an outbound
+connection the user never asked for, made by the app itself. The preview gets
+its own off-the-record profile (so the interceptor below touches no other web
+view, e.g. the GraphRAG renderer) and every remote request is refused while
+the switch is on. Local files, ``data:`` and ``qrc:`` URLs still load.
+"""
+from __future__ import annotations
+
+from PySide6.QtWebEngineCore import (
+    QWebEnginePage, QWebEngineProfile, QWebEngineUrlRequestInterceptor,
+)
+
+from cowork_local.application.network import network_guard
+
+_REMOTE_SCHEMES = frozenset({"http", "https", "ws", "wss", "ftp"})
+
+
+class RemoteRequestBlocker(QWebEngineUrlRequestInterceptor):
+    """Refuses remote URLs whenever the network guard says the network is blocked."""
+
+    def interceptRequest(self, info) -> None:  # noqa: N802 - Qt override
+        """Called by WebEngine for every request the page makes."""
+        if info.requestUrl().scheme().lower() in _REMOTE_SCHEMES and network_guard.is_blocked():
+            info.block(True)
+
+
+def install_offline_page(view) -> None:
+    """Give ``view`` a private profile whose remote requests obey the guard."""
+    profile = QWebEngineProfile(view)          # no storage name = off the record
+    blocker = RemoteRequestBlocker(profile)
+    profile.setUrlRequestInterceptor(blocker)
+    view.setPage(QWebEnginePage(profile, view))
+    view._remote_blocker = blocker             # keep the interceptor alive with the view
+
+
+__all__ = ["RemoteRequestBlocker", "install_offline_page"]

presentation/shell/bootstrap.py

@@ -42,5 +42,14 @@ def build_config(path: Path | None = None) -> JsonConfigRepository:
 
 
 def build_context(path: Path | None = None) -> AppContext:
-    """Dựng AppContext hoàn chỉnh — điểm vào cho ``app.run()`` và cho checker."""
-    return AppContext(build_config(path))
+    """Dựng AppContext hoàn chỉnh — điểm vào cho ``app.run()`` và cho checker.
+
+    Nối luôn cổng mạng chung vào cấu hình SỐNG của context: đổi công tắc
+    "Chặn mạng" trong Settings là có hiệu lực ngay ở lời gọi mạng kế tiếp.
+    """
+    from ...application.network import network_guard
+    from ...core.agent_security import sandbox_settings
+
+    ctx = AppContext(build_config(path))
+    network_guard.bind(lambda: sandbox_settings(ctx.config)[1])
+    return ctx

state.py

@@ -234,14 +234,19 @@ class AppContext:
         be slow and wasteful). A server/connector that fails to connect is
         skipped, not a hard failure for the turn."""
         # Sandbox Security Layer blocks agent-owned network connectors before
-        # they can spawn a server or issue a REST request.
-        if self.config.agent_security.get("block_network", False):
-            return [], None
+        # they can spawn a server or issue a REST request — and stops the ones
+        # already running, which could otherwise keep talking to the network.
+        blocked = bool(self.config.agent_security.get("block_network", False))
+        if blocked:
+            self.stop_mcp_connections()
         # Master switch (Monitoring → Tools → Connector): when the admin turns
         # "Connect to external" off, the agent connects to NO external
         # connectors/MCP at all — no subprocesses spawned, no REST calls.
         if not self.config.connect_external:
             return [], None
+        from .core.ms365_local import build_ms365_local_tools
+        if blocked:   # the locally-synced OneDrive folders need no network
+            return build_ms365_local_tools(self.config)
         from .core.ext_connectors import build_ext_connector_tools
         from .core.mcp_client import build_mcp_tools as _merge_mcp_tools
         from .core.tools import combine_tool_sources
@@ -276,7 +281,6 @@ class AppContext:
 
         # Locally-synced OneDrive/SharePoint (no sign-in) — reads/writes the
         # OneDrive-desktop-synced folders directly, gated on ms365.connectors.
-        from .core.ms365_local import build_ms365_local_tools
         local_tools, local_executor = build_ms365_local_tools(self.config)
 
         return combine_tool_sources((mcp_tools, mcp_executor), (ext_tools, ext_executor),

tests/conftest.py

@@ -73,3 +73,17 @@ def _bind_checkout_as_package() -> None:
 
 
 _bind_checkout_as_package()
+
+
+import pytest  # noqa: E402 - after the package binding above
+
+
+@pytest.fixture(autouse=True)
+def _reset_network_guard():
+    """``build_context()`` binds the process-wide "Block network" gate to that
+    context's config (default: blocked). Unbind after every test so one test
+    that built a real context cannot silently block the network for the rest."""
+    yield
+    from cowork_local.application.network import network_guard
+
+    network_guard.bind(None)

tests/test_network_guard_lanes.py

@@ -0,0 +1,389 @@
+"""Công tắc "Chặn mạng" phải chặn mọi đường ra mạng của app, TRỪ nhà cung cấp AI.
+
+Bản đồ ``infura/network-map.html`` liệt kê các làn trước đây "không kiểm soát"
+hoặc chỉ chặn một phần: M365, Teams và nút Test, connector/MCP đang chạy,
+task script và link đính kèm task, tự cài thư viện, xem trước HTML, lệnh shell
+chỉ bị proxy giả. Mỗi làn có ít nhất một bài ở đây:
+
+* **bật** — làn từ chối TRƯỚC khi chạm mạng (mọi lời gọi HTTP thật đều nổ);
+* **tắt** — đường cũ giữ nguyên, vì chặn một chiều là hỏng tính năng.
+"""
+from __future__ import annotations
+
+import socket
+import sys
+import threading
+from pathlib import Path
+
+import pytest
+
+from cowork_local.application.network import network_guard
+
+
+def _no_network(*args, **kwargs):
+    raise AssertionError("đã chạm mạng dù 'Chặn mạng' đang bật")
+
+
+@pytest.fixture
+def chan_mang(monkeypatch):
+    """Bật công tắc qua cổng chung và làm nổ mọi lời gọi HTTP thật."""
+    import requests
+
+    from cowork_local.core import tls_trust
+
+    for name in ("request", "get", "post", "put", "patch", "delete"):
+        monkeypatch.setattr(requests, name, _no_network)
+    monkeypatch.setattr(tls_trust, "request", _no_network)
+    monkeypatch.setattr(tls_trust, "request_any_method", _no_network)
+    network_guard.bind(lambda: True)
+    yield
+    network_guard.bind(None)
+
+
+@pytest.fixture(autouse=True)
+def _go_cong_sau_moi_bai():
+    yield
+    network_guard.bind(None)
+
+
+# ---- cổng chung ------------------------------------------------------------
+
+def test_chua_noi_day_thi_khong_chan():
+    network_guard.bind(None)
+    assert network_guard.is_blocked() is False
+    network_guard.ensure_allowed("x")
+
+
+def test_doc_cong_tac_loi_thi_coi_nhu_dang_chan():
+    """Không đọc được công tắc thì đóng cửa, không mở toang."""
+    def hong():
+        raise KeyError("agent_security")
+
+    network_guard.bind(hong)
+    with pytest.raises(network_guard.NetworkBlockedError):
+        network_guard.ensure_allowed("Teams")
+
+
+def test_cong_doc_cau_hinh_song_khong_chup_lai(tmp_path):
+    """Đổi công tắc trong Settings có hiệu lực ngay, không cần mở lại app."""
+    from cowork_local.presentation.shell.bootstrap import build_context
+
+    ctx = build_context(tmp_path / "config.json")
+    ctx.config.agent_security["block_network"] = True
+    assert network_guard.is_blocked() is True
+    ctx.config.agent_security["block_network"] = False
+    assert network_guard.is_blocked() is False
+
+
+# ---- Microsoft 365 -----------------------------------------------------------
+
+def test_graph_tu_choi_bang_loi_ma_noi_goi_da_bat(chan_mang):
+    from cowork_local.core import ms365_graph
+
+    with pytest.raises(ms365_graph.Ms365GraphError, match="Sandbox Security Layer"):
+        ms365_graph.list_onedrive_files("token")
+
+
+def test_dang_nhap_va_lay_token_tu_choi(chan_mang, monkeypatch):
+    from cowork_local.core import ms365_auth
+
+    monkeypatch.setattr(ms365_auth, "_app", _no_network)
+    with pytest.raises(ms365_auth.Ms365AuthError, match="blocked"):
+        ms365_auth.get_access_token("", "")
+    with pytest.raises(ms365_auth.Ms365AuthError, match="blocked"):
+        ms365_auth.sign_in_device_code("", "", lambda flow: None)
+
+
+def test_kiem_tra_da_dang_nhap_khong_dung_msal_khi_chan(chan_mang, monkeypatch):
+    """Dựng app MSAL là tải cấu hình OpenID của tenant — phải đọc kho token trực tiếp."""
+    from cowork_local.core import ms365_auth
+
+    monkeypatch.setattr(ms365_auth, "_app", _no_network)
+    monkeypatch.setattr(ms365_auth, "_cached_account_offline",
+                        lambda: {"username": "a@b.c"})
+    assert ms365_auth.signed_in_account("", "") == {"username": "a@b.c"}
+
+
+def test_mail_canh_bao_khong_gui_khi_chan(chan_mang, monkeypatch):
+    from types import SimpleNamespace
+
+    from cowork_local.core import agent_security_alert, ms365_auth
+    from cowork_local.core.agent_security_types import SecurityVerdict
+
+    monkeypatch.setattr(ms365_auth, "_app", _no_network)
+    config = SimpleNamespace(data={"agent_security": {"admin_email": "admin@x.y"}}, ms365={})
+    verdict = SecurityVerdict(allowed=False, layer="command", reason="test")
+    sent, note = agent_security_alert.notify_admin(config, verdict)
+    assert sent is False
+    assert "blocked" in note
+
+
+# ---- Teams, connector REST, Jira, nút Test ---------------------------------
+
+def test_teams_tu_choi(chan_mang):
+    from cowork_local.core.teams import TeamsNotifier
+
+    ok, detail = TeamsNotifier("https://x.webhook.office.com/hook").send("t", "b")
+    assert ok is False
+    assert "blocked" in detail
+
+
+def test_connector_rest_va_nut_test_tu_choi(chan_mang):
+    from cowork_local.core.ext_connectors import RestApiConnector
+
+    rc = RestApiConnector({"id": "erp", "name": "ERP", "base_url": "https://erp.example"})
+    assert rc.call({"method": "GET", "path": "items"})["ok"] is False
+    ok, message = rc.test_connection()
+    assert ok is False and "blocked" in message
+
+
+def test_nut_test_jira_tu_choi(chan_mang):
+    from cowork_local.core import jira_tool
+
+    cfg = {"base_url": "https://jira.example", "email": "a@b.c", "api_token": "t"}
+    out = jira_tool.search(cfg, "order by created DESC", 1)
+    assert out.startswith("Jira search failed") and "blocked" in out
+
+
+# ---- MCP -------------------------------------------------------------------
+
+def test_mcp_khong_khoi_dong_va_khong_goi_khi_chan(chan_mang):
+    from cowork_local.core.mcp_client import McpServerConnection, McpServerError
+
+    conn = McpServerConnection("srv", "definitely-not-run")
+    with pytest.raises(McpServerError, match="blocked"):
+        conn.start(timeout=1)
+    assert conn.call_tool("srv__search", {})["ok"] is False
+
+
+def _context(tmp_path, block: bool):
+    from cowork_local.config import AppConfig
+    from cowork_local.state import AppContext
+
+    ctx = AppContext(AppConfig.load(tmp_path / "config.json"))
+    ctx.config.agent_security["block_network"] = block
+    return ctx
+
+
+def test_chuan_bi_connector_dung_mcp_dang_chay_va_giu_onedrive_cuc_bo(tmp_path, monkeypatch):
+    from cowork_local.core import ms365_local
+
+    ctx = _context(tmp_path, block=True)
+    stopped = []
+    monkeypatch.setattr(ctx, "stop_mcp_connections", lambda: stopped.append(True))
+    local = (["ms365_local__onedrive_list"], object())
+    monkeypatch.setattr(ms365_local, "build_ms365_local_tools", lambda config: local)
+    monkeypatch.setattr(ctx._mcp_manager, "ensure", _no_network)
+
+    assert ctx.build_mcp_tools() == local
+    assert stopped == [True]
+
+
+# ---- task lập lịch -----------------------------------------------------------
+
+def test_link_dinh_kem_task_khong_tai(chan_mang):
+    from cowork_local.core.tasks import resolve_input_text
+
+    text = resolve_input_text({"input": {"mode": "empty", "links": ["https://example.com/a"]}})
+    assert "not fetched" in text
+
+
+def test_task_script_chay_trong_tien_trinh_khong_mang(chan_mang, monkeypatch, tmp_path):
+    from cowork_local.core import deps, task_script
+
+    calls = []
+
+    def fake_run(command, **kwargs):
+        calls.append(kwargs)
+        return 0, "done", False, False, False
+
+    monkeypatch.setattr(deps, "run_cancellable", fake_run)
+    assert task_script.run_script("echo hi", tmp_path, 30) == "done"
+    assert calls and calls[0]["isolate_network"] is True
+
+
+def test_task_script_khong_co_lap_duoc_thi_khong_chay(chan_mang, monkeypatch, tmp_path):
+    from cowork_local.core import deps, task_script
+
+    monkeypatch.setattr(deps, "run_cancellable",
+                        lambda command, **kw: (None, "no AppContainer", False, False, False))
+    with pytest.raises(RuntimeError, match="could not be isolated"):
+        task_script.run_script("echo hi", tmp_path, 30)
+
+
+def test_tat_chan_mang_thi_task_script_chay_nhu_cu(tmp_path):
+    from cowork_local.core import task_script
+
+    assert "hi" in task_script.run_script("echo hi", tmp_path, 30)
+
+
+# ---- tự cài thư viện ---------------------------------------------------------
+
+def test_khong_tu_cai_thu_vien_khi_chan(chan_mang, monkeypatch):
+    from cowork_local.core import deps
+
+    monkeypatch.setattr(deps, "run_cancellable", _no_network)
+    assert deps.ensure_module("khong_ton_tai_xyz_123") is None
+    assert "khong_ton_tai_xyz_123" not in deps._FAILED   # thử lại khi mở mạng
+    ok, detail = deps.pip_install("requests")
+    assert ok is False and "blocked" in detail
+
+
+def test_pptx_khong_bi_nho_la_thieu_khi_chi_do_chan_mang(chan_mang, monkeypatch):
+    from cowork_local.application.workspaces import file_preview_helpers as fph
+    from cowork_local.core import deps
+
+    monkeypatch.setattr(fph, "_PPTX_READY", None)
+    monkeypatch.setattr(deps, "ensure_module", lambda *a, **k: None)
+    assert fph.pptx_available() is False
+    assert fph._PPTX_READY is None
+
+
+# ---- lệnh shell của agent ----------------------------------------------------
+
+@pytest.mark.parametrize("backend", ["direct", "integrity_job_wfp", "appcontainer", "windows_sandbox"])
+def test_moi_backend_deu_chay_co_lap_mang(monkeypatch, tmp_path, backend):
+    """Các backend cũ chỉ đặt biến proxy — khi chặn mạng, tất cả đi qua lớp cô lập của OS."""
+    from cowork_local.core import deps
+    from cowork_local.core.sandbox_manager import SandboxManager
+
+    calls = []
+
+    def fake_run(command, **kwargs):
+        calls.append(kwargs)
+        return 0, "ok", False, False, False
+
+    monkeypatch.setattr(deps, "run_cancellable", fake_run)
+    result = SandboxManager()._execute_with_backend(backend, "echo hi", str(tmp_path), True, 30)
+    assert calls and calls[0]["isolate_network"] is True
+    assert result["ok"] is True and result["sandbox"] == "network_isolated"
+
+
+def test_lenh_khong_co_lap_duoc_thi_bi_tu_choi(monkeypatch, tmp_path):
+    from cowork_local.core import deps
+    from cowork_local.core.sandbox_manager import SandboxManager
+
+    monkeypatch.setattr(deps, "run_cancellable",
+                        lambda command, **kw: (None, "no isolation", False, False, False))
+    result = SandboxManager()._execute_with_backend("direct", "echo hi", str(tmp_path), True, 30)
+    assert result["ok"] is False and result["sandbox"] == "blocked"
+
+
+def test_he_dieu_hanh_khong_ho_tro_thi_bao_loi_khong_chay(monkeypatch):
+    from cowork_local.infrastructure.sandbox import network_isolation
+
+    monkeypatch.setattr(network_isolation.sys, "platform", "sunos5")
+    with pytest.raises(network_isolation.NetworkIsolationUnavailable):
+        network_isolation.spawn_without_network("echo hi", None, None)
+
+
+def _listening_socket():
+    server = socket.socket()
+    server.bind(("127.0.0.1", 0))
+    server.listen(4)
+    def serve():
+        try:
+            for _ in range(4):
+                server.accept()
+        except OSError:
+            pass   # socket closed at the end of the test
+
+    threading.Thread(target=serve, daemon=True).start()
+    return server
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="AppContainer chỉ có trên Windows")
+def test_appcontainer_that_su_cat_mang_nhung_van_ghi_duoc_thu_muc(tmp_path):
+    """Bài thật, không giả lập: cùng một lệnh Python nối tới một cổng đang nghe,
+    ngoài sandbox thì được, trong sandbox thì bị kernel từ chối."""
+    from cowork_local.core.deps import run_cancellable
+
+    server = _listening_socket()
+    port = server.getsockname()[1]
+    probe = (f'"{sys.executable}" -c "import socket;'
+             "print('PYTHON'+'_RAN');"
+             f"socket.create_connection(('127.0.0.1',{port}),5);print('CONN'+'ECTED')\"")
+    try:
+        rc, out, *_ = run_cancellable(probe, cwd=str(tmp_path), timeout=60, shell=True)
+        assert rc == 0 and "CONNECTED" in out
+
+        rc, out, *_ = run_cancellable(probe + " & echo written> marker.txt",
+                                      cwd=str(tmp_path), timeout=60, isolate_network=True)
+        assert "PYTHON_RAN" in out, out          # Python itself starts in the sandbox
+        assert "CONNECTED" not in out, out
+        assert (Path(tmp_path) / "marker.txt").read_text().strip() == "written"
+    finally:
+        server.close()
+
+
+# ---- xem trước HTML ------------------------------------------------------------
+
+class _FakeUrl:
+    def __init__(self, scheme):
+        self._scheme = scheme
+
+    def scheme(self):
+        return self._scheme
+
+
+class _FakeRequest:
+    def __init__(self, scheme):
+        self._url = _FakeUrl(scheme)
+        self.blocked = False
+
+    def requestUrl(self):  # noqa: N802 - Qt name
+        return self._url
+
+    def block(self, flag):
+        self.blocked = flag
+
+
+@pytest.mark.parametrize("scheme,expected", [
+    ("https", True), ("http", True), ("wss", True), ("file", False), ("data", False),
+])
+def test_xem_truoc_html_chan_tai_nguyen_web(chan_mang, scheme, expected):
+    pytest.importorskip("PySide6.QtWebEngineCore")
+    from cowork_local.presentation.folder.offline_web_page import RemoteRequestBlocker
+
+    req = _FakeRequest(scheme)
+    RemoteRequestBlocker().interceptRequest(req)
+    assert req.blocked is expected
+
+
+def test_xem_truoc_html_tai_binh_thuong_khi_tat():
+    pytest.importorskip("PySide6.QtWebEngineCore")
+    from cowork_local.presentation.folder.offline_web_page import RemoteRequestBlocker
+
+    req = _FakeRequest("https")
+    RemoteRequestBlocker().interceptRequest(req)
+    assert req.blocked is False
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="AppContainer chỉ có trên Windows")
+def test_khong_cap_quyen_ke_thua_len_thu_muc_chua_qt(tmp_path, monkeypatch):
+    """Quyền AppContainer kế thừa xuống Qt6WebEngineCore.dll làm tiến trình render
+    của WebEngine không nạp được DLL — tab Graph và xem trước HTML trắng trơn."""
+    from cowork_local.infrastructure.sandbox import appcontainer_process as ac
+
+    qt_dir = tmp_path / "venv" / "Lib" / "site-packages" / "PySide6"
+    qt_dir.mkdir(parents=True)
+    monkeypatch.setattr(ac, "_qt_package_dir", lambda: str(qt_dir))
+    monkeypatch.setattr(ac, "_icacls", _no_network)
+
+    with pytest.raises(ac.NetworkIsolationUnavailable, match="Qt runtime"):
+        ac.grant_access(str(tmp_path / "venv"), "S-1-15-2-1", "read")
+    with pytest.raises(ac.NetworkIsolationUnavailable, match="Qt runtime"):
+        ac.grant_access(str(tmp_path), "S-1-15-2-1", "write")
+
+
+def test_quyen_cho_python_khong_bao_gio_phu_len_thu_vien_qt(monkeypatch, tmp_path):
+    from cowork_local.infrastructure.sandbox import appcontainer_process as ac
+
+    venv, base = tmp_path / "venv", tmp_path / "base"
+    monkeypatch.setattr(ac.sys, "prefix", str(venv))
+    monkeypatch.setattr(ac.sys, "base_prefix", str(base))
+    for qt_dir in (venv / "Lib" / "site-packages" / "PySide6",
+                   base / "Lib" / "site-packages" / "PySide6"):
+        monkeypatch.setattr(ac, "_qt_package_dir", lambda d=qt_dir: str(d))
+        for folder, mode in ac._interpreter_grants():
+            assert mode == "read_here" or not ac._covers(folder, str(qt_dir)), (folder, mode)

ui/settings_dialog.py

@@ -96,7 +96,7 @@ class SettingsDialog(QDialog):
         sbl.addWidget(self.sandbox_confirm)
 
         self.sandbox_block_network = ToggleSwitch(tr("settings.sandbox_block_network"))
-        self.sandbox_block_network.setChecked(bool(sec.get("block_network", True)))
+        self.sandbox_block_network.setChecked(bool(sec.get("block_network", False)))
         self.sandbox_block_network.setToolTip(tr("settings.sandbox_block_network_tooltip"))
         sbl.addWidget(self.sandbox_block_network)
 
@@ -277,4 +277,11 @@ class SettingsDialog(QDialog):
         self.ctx.config._data = None  # invalidate cache
         self.ctx.config._agent_security = None
 
+        stop_mcp = getattr(self.ctx, "stop_mcp_connections", None)
+        if self.sandbox_block_network.isChecked() and stop_mcp is not None:
+            # A running MCP server is its own process and may keep using the
+            # network — stop them now, off the UI thread (each stop may wait).
+            import threading
+            threading.Thread(target=stop_mcp, daemon=True).start()
+
         self.accept()