Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 10 additions & 8 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -53,8 +53,9 @@ main:

```bash
$ tccli configure
TencentCloud API secretId [*afcQ]:AKIDwLw1234MMfPRle2g9nR2OTI787aBCDP
TencentCloud API secretKey [*ArFd]:OxXj7khcV1234dQSSYNABcdCc1LiArFd
# 方括号内为已配置密钥的掩码后四位;请在冒号后输入完整密钥,不要输入尖括号。
TencentCloud API secretId [*abcd]:<YOUR_SECRET_ID>
TencentCloud API secretKey [*abcd]:<YOUR_SECRET_KEY>
region: ap-guangzhou
output[json]:
```
Expand All @@ -69,21 +70,21 @@ output: 可选参数,请求回包输出格式,支持[json table text]三
2. 命令行模式,通过命令行模式您可以在自动化脚本中配置您的信息。
```bash
# set子命令可以设置某一配置,也可同时配置多个。
tccli configure set secretId AKIDwLw1234MMfPRle2g9nR2OTI787aBCDP
tccli configure set secretId <YOUR_SECRET_ID>
tccli configure set region ap-guangzhou output json language zh-CN

# set-root-domain命令可以将配置文件中的endpoint的根域名全部设置为同一值。
tccli configure set-root-domain internal.tencentcloudapi.com

# get子命令用于获取配置信息。
tccli configure get secretKey
secretKey = OxXj7khcV1234dQSSYNABcdCc1LiArFd
secretKey = <YOUR_SECRET_KEY>

# list子命令打印所有配置信息。
tccli configure list
credential:
secretId = AKIDwLw1234MMfPRle2g9nR2OTI787aBCDP
secretKey = OxXj7khcV1234dQSSYNABcdCc1LiArFd
secretId = <YOUR_SECRET_ID>
secretKey = <YOUR_SECRET_KEY>
configure:
region = ap-guangzhou
output = json
Expand All @@ -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]:<YOUR_SECRET_ID>
TencentCloud API secretKey [*abcd]:<YOUR_SECRET_KEY>
region: ap-guangzhou
output[json]:

Expand Down
73 changes: 56 additions & 17 deletions tccli/configure.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = [
Expand Down Expand Up @@ -442,29 +442,45 @@ 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:
if OptionsDefine.SysParam in conf_data \
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] = {}

Expand All @@ -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 = {}
Expand Down
5 changes: 3 additions & 2 deletions tccli/oauth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
11 changes: 10 additions & 1 deletion tccli/plugins/sso/login.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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):
Expand Down Expand Up @@ -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,
Expand Down
2 changes: 2 additions & 0 deletions tccli/plugins/sso/texts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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": "请选择您的角色: ",
Expand All @@ -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: ",
Expand Down
17 changes: 11 additions & 6 deletions tccli/sso.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,17 @@
# -*- coding: utf-8 -*-
import json
import os
import time
import uuid

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):
Expand All @@ -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(
Expand All @@ -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:
Expand Down Expand Up @@ -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):
Expand Down
78 changes: 77 additions & 1 deletion tests/test_configure.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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()
Loading