|
|
@@ -0,0 +1,83 @@
|
|
|
+"""SSL bootstrap:企业自签 CA 场景的进程级修复。默认 no-op,env 驱动。"""
|
|
|
+import os
|
|
|
+import ssl
|
|
|
+import unittest
|
|
|
+
|
|
|
+from lambdagent import _ssl_bootstrap as sb
|
|
|
+
|
|
|
+
|
|
|
+class TestSSLBootstrap(unittest.TestCase):
|
|
|
+ def setUp(self):
|
|
|
+ # 隔离:存档并清理相关 env + 默认 context 工厂 + 幂等标志
|
|
|
+ self._env = {k: os.environ.get(k) for k in
|
|
|
+ ("AGENTPAAS_CA_BUNDLE", "SSL_CERT_FILE", "REQUESTS_CA_BUNDLE",
|
|
|
+ "AGENTPAAS_INSECURE_SSL")}
|
|
|
+ for k in self._env:
|
|
|
+ os.environ.pop(k, None)
|
|
|
+ self._orig_factory = ssl._create_default_https_context
|
|
|
+ sb._configured = False
|
|
|
+
|
|
|
+ def tearDown(self):
|
|
|
+ for k, v in self._env.items():
|
|
|
+ if v is None:
|
|
|
+ os.environ.pop(k, None)
|
|
|
+ else:
|
|
|
+ os.environ[k] = v
|
|
|
+ ssl._create_default_https_context = self._orig_factory
|
|
|
+ sb._configured = False
|
|
|
+
|
|
|
+ def test_default_is_noop(self):
|
|
|
+ mode = sb.configure_ssl(force=True)
|
|
|
+ self.assertEqual(mode, "default")
|
|
|
+ self.assertIs(ssl._create_default_https_context, self._orig_factory) # 未改默认
|
|
|
+
|
|
|
+ def test_insecure_disables_verification(self):
|
|
|
+ os.environ["AGENTPAAS_INSECURE_SSL"] = "1"
|
|
|
+ import warnings
|
|
|
+ with warnings.catch_warnings():
|
|
|
+ warnings.simplefilter("ignore")
|
|
|
+ mode = sb.configure_ssl(force=True)
|
|
|
+ self.assertEqual(mode, "insecure")
|
|
|
+ ctx = ssl._create_default_https_context()
|
|
|
+ self.assertEqual(ctx.verify_mode, ssl.CERT_NONE) # 验证已关
|
|
|
+ self.assertFalse(ctx.check_hostname)
|
|
|
+
|
|
|
+ def test_insecure_truthy_variants(self):
|
|
|
+ for val in ("true", "YES", "on", "1"):
|
|
|
+ sb._configured = False
|
|
|
+ os.environ["AGENTPAAS_INSECURE_SSL"] = val
|
|
|
+ self.assertEqual(sb.configure_ssl(force=True), "insecure")
|
|
|
+
|
|
|
+ def test_ca_bundle_trusted_and_verification_on(self):
|
|
|
+ # 用 certifi 的真实 CA 文件当"公司 CA"占位,验证仍开启
|
|
|
+ import certifi
|
|
|
+ os.environ["AGENTPAAS_CA_BUNDLE"] = certifi.where()
|
|
|
+ mode = sb.configure_ssl(force=True)
|
|
|
+ self.assertEqual(mode, "custom-ca:" + certifi.where())
|
|
|
+ ctx = ssl._create_default_https_context()
|
|
|
+ self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED) # 验证保持开启 → 安全
|
|
|
+
|
|
|
+ def test_ca_bundle_takes_precedence_over_insecure(self):
|
|
|
+ import certifi
|
|
|
+ os.environ["AGENTPAAS_CA_BUNDLE"] = certifi.where()
|
|
|
+ os.environ["AGENTPAAS_INSECURE_SSL"] = "1"
|
|
|
+ self.assertEqual(sb.configure_ssl(force=True), "custom-ca:" + certifi.where())
|
|
|
+
|
|
|
+ def test_idempotent(self):
|
|
|
+ os.environ["AGENTPAAS_INSECURE_SSL"] = "1"
|
|
|
+ import warnings
|
|
|
+ with warnings.catch_warnings():
|
|
|
+ warnings.simplefilter("ignore")
|
|
|
+ sb.configure_ssl(force=True)
|
|
|
+ ctx_factory = ssl._create_default_https_context
|
|
|
+ sb.configure_ssl() # 第二次无 force → 不再改动
|
|
|
+ self.assertIs(ssl._create_default_https_context, ctx_factory)
|
|
|
+
|
|
|
+ def test_nonexistent_ca_file_ignored(self):
|
|
|
+ os.environ["AGENTPAAS_CA_BUNDLE"] = "/no/such/ca.pem"
|
|
|
+ # 文件不存在 → 当作未配置 → 回退默认(不崩)
|
|
|
+ self.assertEqual(sb.configure_ssl(force=True), "default")
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ unittest.main()
|