244 lines
9.2 KiB
Python
244 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import unittest
|
|
import uuid
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
|
|
class ExternalCredentialCryptoTests(unittest.TestCase):
|
|
def test_requires_master_key_and_never_falls_back_to_plaintext(self):
|
|
from core.external_systems.crypto import encrypt_secret
|
|
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with self.assertRaisesRegex(RuntimeError, "ZCBOT_CREDENTIAL_MASTER_KEY"):
|
|
encrypt_secret("secret")
|
|
|
|
def test_roundtrip_uses_ciphertext(self):
|
|
from core.external_systems.crypto import decrypt_secret, encrypt_secret
|
|
|
|
with patch.dict(os.environ, {"ZCBOT_CREDENTIAL_MASTER_KEY": "unit-test-key-at-least-32-characters"}, clear=False):
|
|
stored = encrypt_secret("mes-password")
|
|
self.assertTrue(stored.startswith("v1:"))
|
|
self.assertNotIn("mes-password", stored)
|
|
self.assertEqual(decrypt_secret(stored), "mes-password")
|
|
|
|
def test_rejects_short_master_key(self):
|
|
from core.external_systems.crypto import configured, encrypt_secret
|
|
|
|
with patch.dict(os.environ, {"ZCBOT_CREDENTIAL_MASTER_KEY": "too-short"}, clear=False):
|
|
self.assertFalse(configured())
|
|
with self.assertRaisesRegex(RuntimeError, "至少需要 32"):
|
|
encrypt_secret("mes-password")
|
|
|
|
|
|
def _cfg(*, allowed=frozenset()):
|
|
from core.external_systems.factory import FactoryMesConfig
|
|
|
|
return FactoryMesConfig(
|
|
base_url="https://factory.invalid",
|
|
openapi_url="https://factory.invalid/swagger.json",
|
|
login_path="/api/auth/token/",
|
|
allowed_post_operations=frozenset(allowed),
|
|
timeout_seconds=5,
|
|
max_result_bytes=65536,
|
|
verify_tls=True,
|
|
)
|
|
|
|
|
|
_SPEC = {
|
|
"swagger": "2.0",
|
|
"paths": {
|
|
"/api/qm/ftestwork/{batch}/": {
|
|
"get": {
|
|
"operationId": "qm_ftestwork_read",
|
|
"summary": "查询成品检验批次",
|
|
"tags": ["quality"],
|
|
"parameters": [
|
|
{"name": "batch", "in": "path", "required": True, "type": "string"},
|
|
{"name": "page_size", "in": "query", "required": False, "type": "integer"},
|
|
],
|
|
}
|
|
},
|
|
"/api/bi/dataset/{code}/exec/": {
|
|
"post": {
|
|
"operationId": "bi_dataset_exec",
|
|
"summary": "执行只读数据集",
|
|
"parameters": [
|
|
{"name": "code", "in": "path", "required": True, "type": "string"},
|
|
{"name": "payload", "in": "body", "required": True, "schema": {"type": "object"}},
|
|
],
|
|
}
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
class _Response:
|
|
def __init__(self, status_code=200, payload=None):
|
|
self.status_code = status_code
|
|
self._payload = payload
|
|
self.headers = {"content-type": "application/json"}
|
|
self.text = json.dumps(payload, ensure_ascii=False)
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
|
|
class _Http:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def post(self, url, **kwargs):
|
|
self.calls.append(("POST", url, kwargs))
|
|
return _Response(payload={"access": "remote-jwt"})
|
|
|
|
def get(self, url, **kwargs):
|
|
self.calls.append(("GET", url, kwargs))
|
|
return _Response(payload=_SPEC)
|
|
|
|
def request(self, method, url, **kwargs):
|
|
self.calls.append((method, url, kwargs))
|
|
return _Response(payload={"count": 1, "results": [{"batch": "B/1"}]})
|
|
|
|
|
|
class FactoryOpenApiConnectorTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from core.external_systems import factory
|
|
factory._SPEC_CACHE.clear()
|
|
|
|
def test_admin_mapping_builds_bounded_runtime_config(self):
|
|
from core.external_systems.factory import FactoryMesConfig
|
|
|
|
cfg = FactoryMesConfig.from_mapping({
|
|
"base_url": "https://factory.invalid/",
|
|
"openapi_url": "https://factory.invalid/swagger.json",
|
|
"allowed_post_operations": "bi_dataset_exec, report_preview",
|
|
"timeout_seconds": 999,
|
|
"max_result_bytes": 1,
|
|
"verify_tls": True,
|
|
})
|
|
self.assertEqual(cfg.base_url, "https://factory.invalid")
|
|
self.assertEqual(cfg.timeout_seconds, 60)
|
|
self.assertEqual(cfg.max_result_bytes, 4096)
|
|
self.assertEqual(cfg.allowed_post_operations, {"bi_dataset_exec", "report_preview"})
|
|
|
|
def test_admin_mapping_rejects_embedded_url_credentials(self):
|
|
from core.external_systems.factory import FactoryMesConfig, FactoryMesError
|
|
|
|
with self.assertRaisesRegex(FactoryMesError, "不能内嵌凭据"):
|
|
FactoryMesConfig.from_mapping({
|
|
"base_url": "https://user:secret@factory.invalid",
|
|
"openapi_url": "https://factory.invalid/swagger.json",
|
|
})
|
|
|
|
def test_search_discovers_operation_without_exposing_credentials(self):
|
|
from core.external_systems.factory import FactoryMesClient
|
|
|
|
http = _Http()
|
|
client = FactoryMesClient("mes-user", "mes-password", _cfg())
|
|
with patch.object(client, "_client", return_value=http):
|
|
result = client.search("成品检验")
|
|
self.assertEqual(result[0]["operation_id"], "qm_ftestwork_read")
|
|
rendered = json.dumps(result, ensure_ascii=False)
|
|
self.assertNotIn("mes-password", rendered)
|
|
self.assertNotIn("remote-jwt", rendered)
|
|
|
|
def test_get_call_resolves_encoded_path_and_query(self):
|
|
from core.external_systems.factory import FactoryMesClient
|
|
|
|
http = _Http()
|
|
client = FactoryMesClient("mes-user", "mes-password", _cfg())
|
|
with patch.object(client, "_client", return_value=http):
|
|
result = client.call(
|
|
"qm_ftestwork_read",
|
|
arguments={"batch": "B/1", "page_size": 50},
|
|
)
|
|
method, url, kwargs = [call for call in http.calls if call[0] == "GET" and "/api/" in call[1]][0]
|
|
self.assertEqual(method, "GET")
|
|
self.assertIn("B%2F1", url)
|
|
self.assertEqual(kwargs["params"], {"page_size": 50})
|
|
self.assertEqual(result["data"]["count"], 1)
|
|
|
|
def test_post_is_denied_unless_admin_allowlists_operation(self):
|
|
from core.external_systems.factory import FactoryMesClient, FactoryMesError
|
|
|
|
denied = FactoryMesClient("u", "p", _cfg())
|
|
with patch.object(denied, "authenticate", return_value="jwt"), patch.object(
|
|
denied, "_fetch_spec", return_value=_SPEC
|
|
):
|
|
with self.assertRaisesRegex(FactoryMesError, "只读调用范围"):
|
|
denied.call("bi_dataset_exec", arguments={"code": "x", "payload": {}})
|
|
|
|
http = _Http()
|
|
allowed = FactoryMesClient("u", "p", _cfg(allowed={"bi_dataset_exec"}))
|
|
with patch.object(allowed, "_client", return_value=http):
|
|
result = allowed.call(
|
|
"bi_dataset_exec",
|
|
arguments={"code": "yield", "payload": {"query": {"month": "2026-08"}}},
|
|
)
|
|
request = [call for call in http.calls if call[0] == "POST" and "/dataset/" in call[1]][0]
|
|
self.assertEqual(request[2]["json"], {"query": {"month": "2026-08"}})
|
|
self.assertFalse(result["truncated"])
|
|
|
|
def test_allowlisted_post_accepts_separate_body_field(self):
|
|
from core.external_systems.factory import FactoryMesClient
|
|
|
|
http = _Http()
|
|
allowed = FactoryMesClient("u", "p", _cfg(allowed={"bi_dataset_exec"}))
|
|
with patch.object(allowed, "_client", return_value=http):
|
|
result = allowed.call(
|
|
"bi_dataset_exec",
|
|
arguments={"code": "quality"},
|
|
body={"batch": "B-1"},
|
|
)
|
|
request = [call for call in http.calls if call[0] == "POST" and "/dataset/" in call[1]][0]
|
|
self.assertEqual(request[2]["json"], {"batch": "B-1"})
|
|
self.assertFalse(result["truncated"])
|
|
|
|
def test_rejects_unknown_arguments(self):
|
|
from core.external_systems.factory import FactoryMesClient, FactoryMesError
|
|
|
|
client = FactoryMesClient("u", "p", _cfg())
|
|
with patch.object(client, "authenticate", return_value="jwt"), patch.object(
|
|
client, "_fetch_spec", return_value=_SPEC
|
|
):
|
|
with self.assertRaisesRegex(FactoryMesError, "接口定义之外"):
|
|
client.call(
|
|
"qm_ftestwork_read",
|
|
arguments={"batch": "B1", "unexpected": "x"},
|
|
)
|
|
|
|
|
|
class ExternalSystemToolSafetyTests(unittest.TestCase):
|
|
def test_tools_are_scoped_to_constructor_user(self):
|
|
from tools.external_systems import ExternalSystemListTool
|
|
|
|
uid = uuid.uuid4()
|
|
with patch(
|
|
"tools.external_systems.list_external_systems",
|
|
return_value=[{
|
|
"external_system_id": str(uuid.uuid4()),
|
|
"status": "active",
|
|
"username_masked": "me***r",
|
|
}],
|
|
) as listed:
|
|
output = ExternalSystemListTool(uid).execute()
|
|
listed.assert_called_once_with(uid)
|
|
self.assertNotIn("password", output.lower())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|