diff --git a/README.md b/README.md index ea39078bfe..c9a4cbf23e 100644 --- a/README.md +++ b/README.md @@ -53,8 +53,9 @@ main: ```bash $ tccli configure -TencentCloud API secretId [*afcQ]:AKIDwLw1234MMfPRle2g9nR2OTI787aBCDP -TencentCloud API secretKey [*ArFd]:OxXj7khcV1234dQSSYNABcdCc1LiArFd +# 方括号内为已配置密钥的掩码后四位；请在冒号后输入完整密钥，不要输入尖括号。 +TencentCloud API secretId [*abcd]: +TencentCloud API secretKey [*abcd]: region: ap-guangzhou output[json]: ``` @@ -69,7 +70,7 @@ output： 可选参数，请求回包输出格式，支持[json table text]三 2. 命令行模式，通过命令行模式您可以在自动化脚本中配置您的信息。 ```bash # set子命令可以设置某一配置，也可同时配置多个。 -tccli configure set secretId AKIDwLw1234MMfPRle2g9nR2OTI787aBCDP +tccli configure set secretId tccli configure set region ap-guangzhou output json language zh-CN # set-root-domain命令可以将配置文件中的endpoint的根域名全部设置为同一值。 @@ -77,13 +78,13 @@ tccli configure set-root-domain internal.tencentcloudapi.com # get子命令用于获取配置信息。 tccli configure get secretKey -secretKey = OxXj7khcV1234dQSSYNABcdCc1LiArFd +secretKey = # list子命令打印所有配置信息。 tccli configure list credential: -secretId = AKIDwLw1234MMfPRle2g9nR2OTI787aBCDP -secretKey = OxXj7khcV1234dQSSYNABcdCc1LiArFd +secretId = +secretKey = configure: region = ap-guangzhou output = json @@ -98,8 +99,9 @@ tccli configure remove ```bash 在交互模式中指定账户名test。 $ tccli configure --profile test -TencentCloud API secretId [*BCDP]:AKIDwLw1234MMfPRle2g9nR2OTI787aBCDP -TencentCloud API secretKey [*ArFd]:OxXj7khcV1234dQSSYNABcdCc1LiArFd +# 方括号内为已配置密钥的掩码后四位；请在冒号后输入完整密钥，不要输入尖括号。 +TencentCloud API secretId [*abcd]: +TencentCloud API secretKey [*abcd]: region: ap-guangzhou output[json]: diff --git a/tccli/configure.py b/tccli/configure.py index 41891d11ed..262c391537 100644 --- a/tccli/configure.py +++ b/tccli/configure.py @@ -410,8 +410,8 @@ class ConfigureCommand(BasicConfigure): "\n\n" \ "To update just the region name::\n" \ " $ tccli configure\n" \ - " TencentCloud API secretId [****]:\n" \ - " TencentCloud API secretKey [****]:\n" \ + " TencentCloud API secretId [None]:\n" \ + " TencentCloud API secretKey [None]:\n" \ " Default region name [ap-guangzhou]: ap-beijing\n" \ " Default output format [json]:\n" SUBCOMMANDS = [ @@ -442,16 +442,38 @@ def _run_main(self, parsed_args, parsed_globals): OptionsDefine.Output: "json" } - cred = { - OptionsDefine.SecretId: "None", - OptionsDefine.SecretKey: "None" - } - is_conf_exist, config_path = self._profile_existed(profile_name + ".configure") is_cred_exist, cred_path = self._profile_existed(profile_name + ".credential") conf_data = {} cred_data = {} + old_cred_data = Utils.load_json_msg(cred_path) if is_cred_exist else {} + credential_type = old_cred_data.get("type") + if "type" not in old_cred_data or credential_type == "default": + dynamic_credential = False + elif credential_type in ("sso", "oauth", "cvm-role"): + dynamic_credential = True + else: + raise ConfigurationError("Invalid credential type: %s" % credential_type) + + if dynamic_credential: + display_type = { + "sso": "SSO", + "oauth": "OAuth", + "cvm-role": "CVM Role" + }[credential_type] + print("Current authentication method: %s." % display_type) + print("Continuing will switch to static SecretId/SecretKey credentials.") + print("The existing credential refresh capability will no longer be used.") + if self._compat_input("Continue? [y/N]: ").strip().lower() != "y": + print("Configuration cancelled. No changes were saved.") + return + + cred = { + OptionsDefine.SecretId: old_cred_data.get(OptionsDefine.SecretId, "") or "None", + OptionsDefine.SecretKey: old_cred_data.get(OptionsDefine.SecretKey, "") or "None" + } + if is_conf_exist: conf_data = Utils.load_json_msg(config_path) for c in config: @@ -459,12 +481,6 @@ def _run_main(self, parsed_args, parsed_globals): and c in conf_data[OptionsDefine.SysParam] \ and conf_data[OptionsDefine.SysParam][c]: config[c] = conf_data[OptionsDefine.SysParam][c] - if is_cred_exist: - cred_data = Utils.load_json_msg(cred_path) - for c in cred: - if c in cred_data and cred_data[c]: - cred[c] = cred_data[c] - if OptionsDefine.SysParam not in conf_data: conf_data[OptionsDefine.SysParam] = {} @@ -484,12 +500,35 @@ def _run_main(self, parsed_args, parsed_globals): else: conf_data[OptionsDefine.SysParam][index] = response if response else config[index] else: - response = self._compat_input( - "%s[%s]: " % (prompt_text, "*"+cred[index][-4:] if cred[index] != "None" else cred[index])) - cred_data[index] = response if response else cred[index] + old_value = old_cred_data.get(index, "") + while True: + response = self._compat_input( + "%s[%s]: " % (prompt_text, "*" + cred[index][-4:] if cred[index] != "None" else cred[index])) + if response: + cred_data[index] = response + break + if not dynamic_credential: + cred_data[index] = old_value + break + field_name = "SecretId" if index == OptionsDefine.SecretId else "SecretKey" + print("%s will be replaced with an empty value." % field_name) + if self._compat_input("Continue? [y/N]: ").strip().lower() == "y": + cred_data[index] = "" + break self._init_configure(profile_name + ".configure", conf_data) - self._init_configure(profile_name + ".credential", cred_data) + Utils.dump_json_msg(cred_path, cred_data) + + secret_id = cred_data.get(OptionsDefine.SecretId, "") + secret_key = cred_data.get(OptionsDefine.SecretKey, "") + if dynamic_credential: + print('Profile "%s" has been switched from %s to static credentials.' % ( + profile_name, display_type)) + if not secret_id or not secret_key: + print("Warning: SecretId or SecretKey is empty. Subsequent API requests may fail authentication.") + print("Run `tccli configure --profile %s` to update the credentials." % profile_name) + elif not dynamic_credential: + print("Configure saved successfully.") def init_configures(self): config = {} diff --git a/tccli/oauth.py b/tccli/oauth.py index 34481b8cfd..1e04039991 100644 --- a/tccli/oauth.py +++ b/tccli/oauth.py @@ -5,6 +5,8 @@ import requests import uuid +from tccli.utils import Utils + _API_ENDPOINT = "https://cli.cloud.tencent.com" _CRED_REFRESH_SAFE_DUR = 60 * 5 _ACCESS_REFRESH_SAFE_DUR = 60 * 5 @@ -111,5 +113,4 @@ def save_credential(token, new_cred, profile): "site": token["site"], }, } - with open(cred_path, "w") as cred_file: - json.dump(cred, cred_file, indent=4) + Utils.dump_json_msg(cred_path, cred) diff --git a/tccli/plugins/sso/login.py b/tccli/plugins/sso/login.py index 743abecabc..b0c6efab66 100644 --- a/tccli/plugins/sso/login.py +++ b/tccli/plugins/sso/login.py @@ -14,6 +14,10 @@ from tccli.plugins.sso import texts, terminal, configs from tccli.plugins.sso.texts import get as _ +_DURATION_MIN = 1800 +_DURATION_MAX = 43200 +_DURATION_DEFAULT = 7200 + def print_message(msg): print(msg) @@ -34,6 +38,11 @@ def login_command_entrypoint(args, parsed_globals): def login(args, profile, language): + duration = args.get("duration", _DURATION_DEFAULT) + if not (_DURATION_MIN <= duration <= _DURATION_MAX): + print_message(_("invalid_duration") % duration) + return + cred_path = sso.cred_path_of_profile(profile) auth_url = "" if os.path.exists(cred_path): @@ -106,7 +115,7 @@ def login(args, profile, language): role_arn = "qcs::cam::uin/%s:roleName/TencentCloudSSO-%s" % (account["Uin"], role["RoleConfigurationName"]) principal_arn = "qcs::cam::uin/%s:saml-provider/TencentReservedSSO-%s" % (account["Uin"], token_info["ZoneId"]) cred = sso.assume_role_with_saml( - saml_resp["SAMLResponse"], principal_arn, role_arn, "ses-%s" % uuid.uuid4(), args.get("duration", 7200), site) + saml_resp["SAMLResponse"], principal_arn, role_arn, "ses-%s" % uuid.uuid4(), duration, site) sso_info = { "token": login_token, diff --git a/tccli/plugins/sso/texts.py b/tccli/plugins/sso/texts.py index 9e8842c941..d86feac44b 100644 --- a/tccli/plugins/sso/texts.py +++ b/tccli/plugins/sso/texts.py @@ -8,6 +8,7 @@ "invalid_auth_url": "输入的 url 不合法: %s", "auth_url_not_configured": "尚未配置 sso url, 使用 `tccli sso configure %s--url https://your-login-domain.com` 来进行配置", "configure_succeed": "url 已配置为 '%s', 接下来可以使用 `tccli sso login` 进行登陆", + "invalid_duration": "duration 需在 1800~43200 秒（30min~12h）之间, 当前值: %s", "try_login_with_url": "在浏览器中转到以下链接, 并根据提示完成登录:", "account_select_prompt": "登录成功, 请选择您的用户: ", "role_select_prompt": "请选择您的角色: ", @@ -22,6 +23,7 @@ "invalid_auth_url": "The entered url is invalid: %s", "auth_url_not_configured": "sso url is not configured yet, use `tccli sso configure %s--url https://your-login-domain.com` to configure", "configure_succeed": "The url has been configured as '%s', use `tccli sso login` to log in", + "invalid_duration": "duration must be between 1800 and 43200 seconds (30min~12h), current value: %s", "try_login_with_url": "Go to the following link in your browser, and complete the sign-in prompts:", "account_select_prompt": "Login succeed, choose your account: ", "role_select_prompt": "choose your role: ", diff --git a/tccli/sso.py b/tccli/sso.py index 54714708ec..bdde391152 100644 --- a/tccli/sso.py +++ b/tccli/sso.py @@ -1,3 +1,4 @@ +# -*- coding: utf-8 -*- import json import os import time @@ -5,9 +6,12 @@ import requests +from tccli.utils import Utils + _API_ENDPOINT = "https://cli.cloud.tencent.com" _CRED_REFRESH_SAFE_DUR = 60 * 5 -_SKEY_REFRESH_SAFE_DUR = 3600 * 12 - 300 +_CRED_DEFAULT_DUR = 7200 +_SKEY_REFRESH_SAFE_DUR = 60 * 5 # STS 临时凭证最短有效期，低于此值需重新登录 def maybe_refresh_credential(profile): @@ -32,8 +36,9 @@ def maybe_refresh_credential(profile): sso_info = cred["sso"] site = sso_info["site"] sso_expires = sso_info["expiresAt"] - if sso_expires - now < _SKEY_REFRESH_SAFE_DUR: - # sso can't be refreshed if expired, re-login is required + sso_remaining = int(sso_expires - time.time()) + if sso_remaining < _SKEY_REFRESH_SAFE_DUR: + # sso can't issue a credential with the minimum duration, re-login is required return saml_resp = gen_saml_response( @@ -43,8 +48,9 @@ def maybe_refresh_credential(profile): role_arn = "qcs::cam::uin/%s:roleName/TencentCloudSSO-%s" % (sso_info["uin"], sso_info["roleConfigurationName"]) principal_arn = "qcs::cam::uin/%s:saml-provider/TencentReservedSSO-%s" % (sso_info["uin"], sso_info["zoneId"]) + refresh_dur = min(_CRED_DEFAULT_DUR, sso_remaining) cred = assume_role_with_saml( - saml_resp["SAMLResponse"], principal_arn, role_arn, "ses-%s" % uuid.uuid4(), 7200, site) + saml_resp["SAMLResponse"], principal_arn, role_arn, "ses-%s" % uuid.uuid4(), refresh_dur, site) save_credential(cred, sso_info, profile) except KeyError as e: @@ -206,8 +212,7 @@ def save_credential(cred, sso_info, profile): "expiresAt": sso_info["expiresAt"], }, } - with open(cred_path, "w") as cred_file: - json.dump(cred, cred_file, indent=4) + Utils.dump_json_msg(cred_path, cred) def cred_path_of_profile(profile): diff --git a/tests/test_configure.py b/tests/test_configure.py index aca04ba348..fa9fe1f539 100644 --- a/tests/test_configure.py +++ b/tests/test_configure.py @@ -7,8 +7,19 @@ import six +try: + from unittest import mock +except ImportError: + import mock + import tccli.options_define as OptionsDefine -from tccli.configure import ConfigureGetCommand, ConfigureListCommand, mask_secret +from tccli.configure import ( + BasicConfigure, + ConfigureCommand, + ConfigureGetCommand, + ConfigureListCommand, + mask_secret, +) from tccli.utils import Utils @@ -95,5 +106,70 @@ def test_get_masks_sensitive_fields_only(self): self.assertEqual(Utils.load_json_msg(self.credential_path), self.credential) +class TestConfigureDynamicCredentialMigration(unittest.TestCase): + """交互式 configure 切换动态凭证时的安全行为。""" + + def setUp(self): + self.cli_path = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.cli_path) + + def _create_command(self): + command = ConfigureCommand.__new__(ConfigureCommand) + BasicConfigure.__init__(command) + command.cli_path = self.cli_path + return command + + def _credential_path(self, profile): + return os.path.join(self.cli_path, "%s.credential" % profile) + + def _write_credential(self, profile, credential): + Utils.dump_json_msg(self._credential_path(profile), credential) + + def _read_credential(self, profile): + return Utils.load_json_msg(self._credential_path(profile)) + + def test_dynamic_credential_switch_requires_confirmation(self): + old_credential = { + "type": "sso", + OptionsDefine.SecretId: "OLD_ID", + OptionsDefine.SecretKey: "OLD_KEY", + "sso": {"token": "old-token"}, + } + self._write_credential("default", old_credential) + command = self._create_command() + + with mock.patch.object(command, "_compat_input", return_value="n"): + with mock.patch.object(command, "_init_configure") as init_configure: + command._run_main(mock.Mock(), argparse.Namespace(profile="default")) + + init_configure.assert_not_called() + self.assertEqual(self._read_credential("default"), old_credential) + + def test_confirmed_dynamic_credential_switch_replaces_refresh_data(self): + for credential_type in ("sso", "oauth", "cvm-role"): + profile = credential_type + self._write_credential(profile, { + "type": credential_type, + OptionsDefine.SecretId: "OLD_ID", + OptionsDefine.SecretKey: "OLD_KEY", + "sso": {"token": "old-token"}, + }) + command = self._create_command() + + with mock.patch.object( + command, "_compat_input", + side_effect=["y", "NEW_ID", "NEW_KEY", "", ""]): + with mock.patch.object(command, "_init_configure") as init_configure: + command._run_main(mock.Mock(), argparse.Namespace(profile=profile)) + + init_configure.assert_called_once() + self.assertEqual(self._read_credential(profile), { + OptionsDefine.SecretId: "NEW_ID", + OptionsDefine.SecretKey: "NEW_KEY", + }) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_oauth.py b/tests/test_oauth.py new file mode 100644 index 0000000000..59b6c58d42 --- /dev/null +++ b/tests/test_oauth.py @@ -0,0 +1,92 @@ +# -*- coding: utf-8 -*- +import json +import os +import shutil +import tempfile +import unittest + +try: + from unittest import mock +except ImportError: + import mock + +from tccli import oauth +from tccli.plugins.auth import login as login_module + + +class TestOAuthCredentialReplacement(unittest.TestCase): + """OAuth 登录成功后原子替换旧 credential。""" + + def setUp(self): + self.temp_dir = tempfile.mkdtemp(prefix="tccli_oauth_test_") + self.cred_path = os.path.join(self.temp_dir, "default.credential") + self.path_patcher = mock.patch.object( + oauth, "cred_path_of_profile", return_value=self.cred_path) + self.path_patcher.start() + + def tearDown(self): + self.path_patcher.stop() + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def _write_credential(self, data): + with open(self.cred_path, "w") as cred_file: + json.dump(data, cred_file) + + def _read_credential(self): + with open(self.cred_path, "r") as cred_file: + return json.load(cred_file) + + @staticmethod + def _token(): + return { + "openId": "open-id", + "accessToken": "access-token", + "expiresAt": 2000, + "refreshToken": "refresh-token", + "site": "cn", + } + + @staticmethod + def _credential(): + return { + "secretId": "NEW_ID", + "secretKey": "NEW_KEY", + "token": "NEW_TOKEN", + "expiresAt": 1000, + } + + def test_success_replaces_cvm_role_with_oauth_credential(self): + """OAuth 登录成功后清除 CVM Role 并切换为完整 OAuth credential。""" + self._write_credential({"type": "cvm-role", "secretId": "OLD_ID"}) + oauth.save_credential(self._token(), self._credential(), "default") + data = self._read_credential() + self.assertEqual(data["type"], "oauth") + self.assertEqual(data["secretId"], "NEW_ID") + self.assertEqual(data["secretKey"], "NEW_KEY") + self.assertEqual(data["token"], "NEW_TOKEN") + self.assertEqual(data["oauth"]["refreshToken"], "refresh-token") + + def test_atomic_replace_failure_preserves_old_credential(self): + """原子替换失败时旧 CVM Role credential 保持不变。""" + old_cred = {"type": "cvm-role", "secretId": "OLD_ID"} + self._write_credential(old_cred) + with mock.patch("tccli.utils.os.rename", side_effect=OSError("rename failed")): + with self.assertRaises(Exception): + oauth.save_credential(self._token(), self._credential(), "default") + self.assertEqual(self._read_credential(), old_cred) + + def test_login_failure_preserves_old_credential(self): + """OAuth 登录流程失败时不调用保存，旧 credential 保持不变。""" + old_cred = {"type": "cvm-role", "secretId": "OLD_ID"} + self._write_credential(old_cred) + with mock.patch.object( + login_module, "_get_token", side_effect=RuntimeError("login failed")): + with mock.patch.object(oauth, "save_credential") as save_credential: + with self.assertRaises(RuntimeError): + login_module.login(True, "default", "zh-CN") + save_credential.assert_not_called() + self.assertEqual(self._read_credential(), old_cred) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sso.py b/tests/test_sso.py new file mode 100644 index 0000000000..8512378276 --- /dev/null +++ b/tests/test_sso.py @@ -0,0 +1,301 @@ +# -*- coding: utf-8 -*- +import json +import os +import shutil +import sys +import tempfile +import time +import unittest + +try: + from unittest import mock +except ImportError: + import mock + +_BUILTINS_MODULE = "__builtin__" if sys.version_info[0] == 2 else "builtins" + +from tccli import sso as sso_module +from tccli.plugins.sso import login as login_module + + +def _make_cred_resp(): + """构造 assume_role_with_saml 的模拟返回值。""" + return { + "Credentials": { + "TmpSecretId": "sid", + "TmpSecretKey": "skey", + "Token": "tok", + }, + "ExpiredTime": int(time.time()) + 7200, + } + + +def _make_sso_info(): + """构造 sso_info 基础字段。""" + return { + "token": "t", + "uin": 123, + "roleConfigurationId": "rid", + "roleConfigurationName": "rname", + "zoneId": "z", + "site": "ap", + "authUrl": "https://example.com", + "expiresAt": int(time.time()) + 3600 * 12, + } + + +def _make_refresh_credential(now, sso_remaining): + """构造即将过期、可进入自动刷新流程的 SSO credential。""" + return { + "type": "sso", + "expiresAt": now + 10, + "sso": { + "expiresAt": now + sso_remaining, + "token": "t", + "uin": 123, + "roleConfigurationId": "rid", + "roleConfigurationName": "rname", + "zoneId": "z", + "site": "ap", + "authUrl": "https://example.com", + }, + } + + +# --------------------------------------------------------------------------- +# TestSaveCredential +# --------------------------------------------------------------------------- + +class TestSaveCredential(unittest.TestCase): + """SSO 登录成功后原子替换旧 credential。""" + + def setUp(self): + self.temp_dir = tempfile.mkdtemp(prefix="tccli_sso_test_") + self.cred_path = os.path.join(self.temp_dir, "default.credential") + self.path_patcher = mock.patch.object( + sso_module, "cred_path_of_profile", return_value=self.cred_path) + self.path_patcher.start() + + def tearDown(self): + self.path_patcher.stop() + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def _write_old_credential(self, data): + with open(self.cred_path, "w") as cred_file: + json.dump(data, cred_file) + + def _read_credential(self): + with open(self.cred_path, "r") as cred_file: + return json.load(cred_file) + + def _call_save(self, sso_info=None): + sso_module.save_credential( + _make_cred_resp(), sso_info or _make_sso_info(), "default") + return self._read_credential() + + def test_legacy_default_duration_is_not_persisted(self): + """即使输入包含旧字段，保存的新 credential 也应丢弃它。""" + sso_info = _make_sso_info() + sso_info["defaultDuration"] = 34200 + data = self._call_save(sso_info) + self.assertNotIn("defaultDuration", data["sso"]) + + def test_success_replaces_cvm_role_with_sso_credential(self): + """SSO 登录成功后清除 CVM Role 并切换为完整 SSO credential。""" + self._write_old_credential({"type": "cvm-role", "secretId": "OLD_ID"}) + data = self._call_save() + self.assertEqual(data["type"], "sso") + self.assertEqual(data["secretId"], "sid") + self.assertEqual(data["secretKey"], "skey") + self.assertEqual(data["token"], "tok") + self.assertNotIn("defaultDuration", data["sso"]) + + def test_atomic_replace_failure_preserves_old_credential(self): + """原子替换失败时旧 CVM Role credential 保持不变。""" + old_cred = {"type": "cvm-role", "secretId": "OLD_ID"} + self._write_old_credential(old_cred) + with mock.patch("tccli.utils.os.rename", side_effect=OSError("rename failed")): + with self.assertRaises(Exception): + self._call_save() + self.assertEqual(self._read_credential(), old_cred) + + +# --------------------------------------------------------------------------- +# TestLoginDuration +# --------------------------------------------------------------------------- + +class TestLoginDuration(unittest.TestCase): + """login.py 单次 duration 参数的解析、校验和传递。""" + + def _run_login(self, args, legacy_duration=None): + cred_data = {"sso": {"authUrl": "https://example.com"}} + if legacy_duration is not None: + cred_data["sso"]["defaultDuration"] = legacy_duration + + login_args = {"uin": "123", "rolename": "rname"} + login_args.update(args) + patchers = [ + ("open", mock.patch(_BUILTINS_MODULE + ".open", mock.mock_open(read_data=json.dumps(cred_data)))), + ("exists", mock.patch.object(login_module.os.path, "exists", return_value=True)), + ("cred_path", mock.patch.object( + login_module.sso, "cred_path_of_profile", return_value="/tmp/default.credential")), + ("get_token", mock.patch.object( + login_module, "_get_token", + side_effect=lambda auth_url, state, language: { + "State": state, "Token": "login-token", "Site": "ap" + })), + ("accounts", mock.patch.object( + login_module.sso, "list_accounts_for_access_assignment", + return_value=[{"Uin": 123, "Name": "account"}])), + ("roles", mock.patch.object( + login_module.sso, "list_role_configurations_for_account", + return_value=[{"RoleConfigurationName": "rname", "RoleConfigurationId": "rid"}])), + ("gen_saml", mock.patch.object( + login_module.sso, "gen_saml_response", return_value={"SAMLResponse": "saml"})), + ("verify", mock.patch.object( + login_module.sso, "verify_login_skey", return_value={"ZoneId": "zone"})), + ("assume", mock.patch.object( + login_module.sso, "assume_role_with_saml", return_value=_make_cred_resp())), + ("save", mock.patch.object(login_module.sso, "save_credential")), + ("print", mock.patch.object(login_module, "print_message")), + ] + mocks = {} + try: + for name, patcher in patchers: + mocks[name] = patcher.start() + login_module.login(login_args, "default", "zh-CN") + finally: + for _, patcher in reversed(patchers): + patcher.stop() + return mocks + + def test_cli_duration_is_used_only_for_current_login(self): + """命令行 duration 传给 STS，但不写入 credential 配置。""" + mocks = self._run_login({"duration": 34200}) + self.assertEqual(mocks["assume"].call_args[0][4], 34200) + saved_sso_info = mocks["save"].call_args[0][1] + self.assertNotIn("defaultDuration", saved_sso_info) + + def test_default_duration_ignores_legacy_persisted_duration(self): + """未传参数时使用默认值，不读取历史 defaultDuration。""" + mocks = self._run_login({}, legacy_duration=34200) + self.assertEqual(mocks["assume"].call_args[0][4], login_module._DURATION_DEFAULT) + + def test_duration_boundaries_are_accepted(self): + """1800 和 43200 两个边界值均可传给 STS。""" + for duration in (1800, 43200): + mocks = self._run_login({"duration": duration}) + self.assertEqual(mocks["assume"].call_args[0][4], duration) + + def test_out_of_range_duration_stops_before_login_flow(self): + """越界值应在读取凭证、打开浏览器前被拒绝。""" + for duration in (1799, 43201): + with mock.patch.object(login_module.sso, "cred_path_of_profile") as cred_path: + with mock.patch.object(login_module, "_get_token") as get_token: + with mock.patch.object(login_module, "print_message") as print_message: + login_module.login({"duration": duration}, "default", "zh-CN") + cred_path.assert_not_called() + get_token.assert_not_called() + self.assertIn("duration", print_message.call_args[0][0]) + + +# --------------------------------------------------------------------------- +# TestLoginCredentialReplacement +# --------------------------------------------------------------------------- + +class TestLoginCredentialReplacement(unittest.TestCase): + """SSO 登录失败时保留旧 credential。""" + + def setUp(self): + self.temp_dir = tempfile.mkdtemp(prefix="tccli_sso_login_test_") + self.cred_path = os.path.join(self.temp_dir, "default.credential") + self.old_cred = { + "type": "cvm-role", + "secretId": "OLD_ID", + "sso": {"authUrl": "https://example.com"}, + } + with open(self.cred_path, "w") as cred_file: + json.dump(self.old_cred, cred_file) + + def tearDown(self): + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_login_failure_preserves_old_credential(self): + """SSO 网络登录失败时不保存新凭证，旧 CVM Role 保持不变。""" + with mock.patch.object( + login_module.sso, "cred_path_of_profile", return_value=self.cred_path): + with mock.patch.object( + login_module, "_get_token", side_effect=RuntimeError("login failed")): + with mock.patch.object(login_module.sso, "save_credential") as save_credential: + with self.assertRaises(RuntimeError): + login_module.login({}, "default", "zh-CN") + save_credential.assert_not_called() + with open(self.cred_path, "r") as cred_file: + self.assertEqual(json.load(cred_file), self.old_cred) + + +# --------------------------------------------------------------------------- +# TestAutoRefresh +# --------------------------------------------------------------------------- + +class TestAutoRefresh(unittest.TestCase): + """sso.py 自动刷新逻辑。""" + + def _run_refresh(self, sso_remaining, time_values=None): + now = 100000.0 + cred = _make_refresh_credential(now, sso_remaining) + patchers = [ + ("open", mock.patch(_BUILTINS_MODULE + ".open", mock.mock_open(read_data=json.dumps(cred)))), + ("time", mock.patch.object( + sso_module.time, "time", side_effect=time_values or [now, now])), + ("gen_saml", mock.patch.object( + sso_module, "gen_saml_response", return_value={"SAMLResponse": "saml"})), + ("assume", mock.patch.object( + sso_module, "assume_role_with_saml", return_value=_make_cred_resp())), + ("save", mock.patch.object(sso_module, "save_credential")), + ] + mocks = {} + try: + for name, patcher in patchers: + mocks[name] = patcher.start() + sso_module.maybe_refresh_credential("default") + finally: + for _, patcher in reversed(patchers): + patcher.stop() + return mocks + + def test_refresh_minimum_session_remaining_is_300(self): + """刷新下限与 SSO 自动刷新安全窗口一致。""" + self.assertEqual(sso_module._SKEY_REFRESH_SAFE_DUR, 300) + + def test_no_refresh_below_minimum_session_remaining(self): + """SSO 会话剩余不足 300 秒时不再尝试刷新。""" + mocks = self._run_refresh(299) + mocks["gen_saml"].assert_not_called() + mocks["assume"].assert_not_called() + + def test_refresh_accepts_minimum_session_remaining(self): + """SSO 会话恰好剩余 300 秒时可申请对应有效期凭证。""" + mocks = self._run_refresh(300) + self.assertEqual(mocks["assume"].call_args[0][4], 300) + + def test_refresh_duration_capped_by_session_remaining(self): + """会话剩余不足默认值时，刷新凭证有效期应截断至会话剩余时间。""" + mocks = self._run_refresh(3000) + self.assertEqual(mocks["assume"].call_args[0][4], 3000) + + def test_refresh_duration_uses_default_when_session_sufficient(self): + """会话剩余充足时，刷新凭证有效期使用默认值。""" + mocks = self._run_refresh(36000) + self.assertEqual(mocks["assume"].call_args[0][4], sso_module._CRED_DEFAULT_DUR) + + def test_refresh_checks_session_with_current_time(self): + """计算会话剩余时间时若已不足 300 秒，不生成 SAML。""" + now = 100000.0 + mocks = self._run_refresh(400, time_values=[now, now + 101]) + mocks["gen_saml"].assert_not_called() + mocks["assume"].assert_not_called() + + +if __name__ == "__main__": + unittest.main()