feat(domain): DTO ToolPolicyGateway — bản đề xuất, gỡ chốt cho N3
N3 (Co4E) cần gọi tool nhưng Team Hoa chưa bắt đầu. Ba đường: N3 ngồi đợi
(trái nguyên tắc không team nào chặn team nào), N3 tự phỏng đoán (không ai
soi, chắc chắn phải sửa), hoặc viết một bản đề xuất để Hoa duyệt. Chọn cái
thứ ba.
Ranh giới giữ đúng sơ đồ phân hệ trong plan.md: domain/security/ là của
Gamma, application/conversations/tool_policy_gateway.py là của Hoa. Nên Gamma
định nghĩa hình dạng, Hoa cài đặt. Không đụng file nào của họ.
Hình dạng bám vào code đang chạy: SecurityVerdict (allowed/reason/layer) và
hộp thoại xin phép ở chat_panel.py:1312. Khác biệt duy nhất là gộp thành một
câu trả lời ba trạng thái ALLOW/DENY/ASK, thay vì bắt chỗ gọi tự nhớ hỏi hai
nơi.
Hai ràng buộc đưa vào có chủ đích, mỗi cái một test:
- DENY và ASK bắt buộc có reason, ném lỗi ngay lúc dựng. Người dùng cần
biết vì sao bị chặn và audit_log cần ghi lại.
- ASK không phải allowed. Đây là bẫy dễ mắc nhất: coi ASK như ALLOW thì
tool chạy trước khi có ai đồng ý.
Kèm FakeToolPolicyGateway lập trình được theo tên tool hoặc theo hàm, có ghi
lại đã hỏi những gì — test khẳng định được "có hỏi cổng không", không chỉ
"kết quả đúng không".
docs/refactor/GammaTeam_decisions.md thêm quyết định 3, kèm nguyên văn tin
nhắn cần gửi Hoa và ô đánh dấu đã gửi / đã xác nhận.
102 test xanh (96 + 6 mới). CASAN Check 1 sạch. domain/ và application/ có 0
import PySide6 — kiểm bằng AST, vì grep đếm ra 4 mà cả 4 là chữ "PySide6"
nằm trong chính docstring cảnh báo. Check 3 của Team Duy nên phân tích cú
pháp chứ đừng grep.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
a164f32bfb
commit
3138856741
@@ -0,0 +1,44 @@
|
||||
"""ToolPolicyGateway giả — để N3 (Co4E) chạy được khi Team Hoa chưa cài đặt.
|
||||
|
||||
Mặc định cho qua hết, vì phần lớn test Co4E quan tâm tới luồng workflow chứ
|
||||
không phải chính sách. Test nào cần kiểm nhánh bị chặn thì lập trình câu trả
|
||||
lời::
|
||||
|
||||
gate = FakeToolPolicyGateway(rules={"run_command": deny("cấm trong Co4E")})
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Callable, Dict
|
||||
|
||||
from cowork_local.domain.security.tool_policy import (
|
||||
PolicyDecision, ToolCallRequest, allow,
|
||||
)
|
||||
|
||||
|
||||
class FakeToolPolicyGateway:
|
||||
"""Cổng chính sách trong bộ nhớ, có ghi lại đã hỏi những gì."""
|
||||
|
||||
def __init__(self, rules: Dict[str, PolicyDecision] | None = None,
|
||||
default: PolicyDecision | None = None,
|
||||
decide: Callable[[ToolCallRequest], PolicyDecision] | None = None):
|
||||
#: {tên tool: quyết định} — tra trước default
|
||||
self.rules = dict(rules or {})
|
||||
self.default = default or allow()
|
||||
#: hàm tự quyết, dùng khi cần logic phức tạp hơn tra bảng
|
||||
self._decide = decide
|
||||
#: mọi lời gọi đã đi qua — để test khẳng định "có hỏi cổng không"
|
||||
self.seen: list[ToolCallRequest] = []
|
||||
|
||||
def check(self, request: ToolCallRequest) -> PolicyDecision:
|
||||
self.seen.append(request)
|
||||
if self._decide is not None:
|
||||
return self._decide(request)
|
||||
return self.rules.get(request.name, self.default)
|
||||
|
||||
# ---- tiện cho test --------------------------------------------------
|
||||
def asked_for(self, name: str) -> bool:
|
||||
return any(r.name == name for r in self.seen)
|
||||
|
||||
@property
|
||||
def call_count(self) -> int:
|
||||
return len(self.seen)
|
||||
@@ -79,3 +79,65 @@ def test_dung_duoc_fake_ma_khong_hề_nap_config_that():
|
||||
capture_output=True, text=True, timeout=60)
|
||||
assert out.returncode == 0, out.stderr
|
||||
assert "OK" in out.stdout
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ToolPolicyGateway — bản đề xuất Gamma viết hộ, chờ Team Hoa xác nhận.
|
||||
# N3 (Co4E) code dựa vào đây từ hôm nay thay vì tự phỏng đoán.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
from cowork_local.domain.security.tool_policy import ( # noqa: E402
|
||||
PolicyOutcome, ToolCallRequest, ToolPolicyGateway, allow, ask, deny,
|
||||
)
|
||||
from cowork_local.tests.fakes.fake_tool_policy import ( # noqa: E402
|
||||
FakeToolPolicyGateway,
|
||||
)
|
||||
|
||||
|
||||
def test_fake_gateway_khop_hop_dong():
|
||||
assert isinstance(FakeToolPolicyGateway(), ToolPolicyGateway)
|
||||
|
||||
|
||||
def test_mac_dinh_cho_qua_va_co_ghi_lai_da_hoi():
|
||||
gate = FakeToolPolicyGateway()
|
||||
d = gate.check(ToolCallRequest(name="read_file", surface="co4e"))
|
||||
assert d.outcome is PolicyOutcome.ALLOW
|
||||
assert d.allowed is True
|
||||
assert gate.asked_for("read_file")
|
||||
assert gate.call_count == 1
|
||||
|
||||
|
||||
def test_chan_theo_ten_tool():
|
||||
gate = FakeToolPolicyGateway(rules={"run_command": deny("cấm trong Co4E")})
|
||||
assert gate.check(ToolCallRequest(name="run_command")).outcome is PolicyOutcome.DENY
|
||||
assert gate.check(ToolCallRequest(name="read_file")).allowed is True
|
||||
|
||||
|
||||
def test_ask_khong_phai_la_duoc_phep():
|
||||
"""Bẫy dễ mắc nhất: coi ASK như ALLOW thì tool chạy mà chưa ai đồng ý."""
|
||||
d = ask("cần người dùng xác nhận")
|
||||
assert d.outcome is PolicyOutcome.ASK
|
||||
assert d.allowed is False
|
||||
|
||||
|
||||
def test_deny_va_ask_bat_buoc_co_ly_do():
|
||||
"""Người dùng phải biết vì sao bị chặn, và audit log cần ghi lại."""
|
||||
import pytest
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
deny("")
|
||||
with pytest.raises(ValueError):
|
||||
ask("")
|
||||
allow() # ALLOW thì không cần lý do
|
||||
|
||||
|
||||
def test_chinh_sach_khac_nhau_theo_man():
|
||||
"""Co4E chạy nền nên không bật được hộp thoại — chặn thẳng thay vì hỏi."""
|
||||
def by_surface(req: ToolCallRequest):
|
||||
if req.surface == "co4e" and req.name == "run_command":
|
||||
return deny("Co4E chạy nền, không hỏi được người dùng")
|
||||
return ask("cần xác nhận") if req.name == "run_command" else allow()
|
||||
|
||||
gate = FakeToolPolicyGateway(decide=by_surface)
|
||||
assert gate.check(ToolCallRequest("run_command", surface="co4e")).outcome is PolicyOutcome.DENY
|
||||
assert gate.check(ToolCallRequest("run_command", surface="cowork")).outcome is PolicyOutcome.ASK
|
||||
|
||||
Reference in New Issue
Block a user