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