[腾讯云]STS（Security Token Service）简单使用

1、什么是STS

腾讯云安全凭证服务（Security Token Service，以下又称 STS）是为腾讯云账号（或 CAM 用户）提供短期访问权限管理的云服务。通过 STS，可以为其他用户颁发一个自定义时效和访问权限的访问凭证。其他用户可以使用 STS 临时访问凭证直接调用腾讯云服务 API。

2、使用STS

一、创建角色

登录腾讯云“控制台 ” – “访问管理" - “角色管理”，点击“新建角色”，角色载体选择“腾讯云账户”

选择“当前主账号”，如果需要为其他主账号中的角色创建临时安全凭证，则选择“其他主账号”

角色授权，授予该角色的权限

输入角色的名称及描述，点击“完成”，即可完成创建

二、申请临时安全凭证

执行脚本，调用STS服务的API接口，传入接口参数后，即可返回申请成功的临时安全凭证，包括 sessionToken ，tmpsecretId , tmpsecretKey，以及过期时间 ExpireTime(指定临时证书的有效期，单位：秒，默认 1800 秒，最长可设定有效期为 7200 秒)

示例代码

[s][p]

# -*- coding:utf-8 -*-
import hashlib
import requests
import hmac
import random
import time
import base64
from urllib import  parse

class STS(object):
    def __init__(self, SecretId, SecretKey, RoleName, ExpireTime):
        self.SecretId = SecretId
        self.SecretKey = SecretKey
        self.RoleName = RoleName
        self.ExpireTime = ExpireTime
        self.requestHost = 'sts.api.qcloud.com'
        self.requestUri = '/v2/index.php?'

    def sts_param(self):
        keydict = {
            'Action' : 'AssumeRole',
            'roleArn' : self.RoleName,
            'roleSessionName' : 'cosrole',
            'durationSecond' : str(self.ExpireTime),
            'Region': 'ap-shanghai',
            'Timestamp': str(int(time.time())),
            'Nonce': str(int(random.random() * 1000)),
            'SecretId': self.SecretId,
        }
        sortlist = sorted(zip(keydict.keys(), keydict.values()))
        return sortlist

    def sts_str_sign(self):
        sortlist = self.sts_param()
        sts_str_init = ''
        for value in sortlist:
            sts_str_init += str(value[0]) + '=' + str(value[1]) + '&'
        sts_str_init = sts_str_init[:-1]
        sign_str = 'GET' + self.requestHost + self.requestUri  + sts_str_init
        return sign_str, sts_str_init

    def get_result_url(self):
        sign_str, sts_str_init = self.sts_str_sign()
        secretkey = self.SecretKey
        signature = bytes(sign_str, encoding='utf-8')
        secretkey = bytes(secretkey, encoding='utf-8')
        my_sign = hmac.new(secretkey, signature, hashlib.sha1).digest()
        my_sign = base64.b64encode(my_sign)
        result_sign = parse.quote(my_sign)
        result_url = 'https://' + self.requestHost + self.requestUri + sts_str_init + '&Signature=' + result_sign
        return result_url

if __name__ == '__main__':
    SecretId = 'AKID54rSwEDDtOjuy4nQkvt'
    SecretKey = 'qUI1zjFWZVGpCPRH'
    RoleName = 'qcs::cam::uin/353:roleName/listcvm'
    ExpireTime = 3600
    STSoper = STS(SecretId, SecretKey, RoleName, ExpireTime)
    url = STSoper.get_result_url()
    try:
        response = requests.get(url)
        if response.status_code == 200:
            #print(response.text)
            data = response.json()
            tmpSecretId = data['data']['credentials']['tmpSecretId']
            tmpSecretKey = data['data']['credentials']['tmpSecretKey']
            sessionToken = data['data']['credentials']['sessionToken']
            expiredTime = data['data']['expiration']
            print('Get sessionToken: %s' % sessionToken)
            print('Get tmpSecretId: %s' % tmpSecretId)
            print('Get tmpSecretKey: %s' % tmpSecretKey)
            print('Expiration: %s' % expiredTime)
    except Exception as e:
        print(e)

# -*- coding:utf-8 -*-

import hashlib

import requests

import hmac

import random

import time

import base64

from urllib import parse

class STS(object):

def __init__(self, SecretId, SecretKey, RoleName, ExpireTime):

self.SecretId = SecretId

self.SecretKey = SecretKey

self.RoleName = RoleName

self.ExpireTime = ExpireTime

self.requestHost = 'sts.api.qcloud.com'

self.requestUri = '/v2/index.php?'

def sts_param(self):

keydict = {

'Action' : 'AssumeRole',

'roleArn' : self.RoleName,

'roleSessionName' : 'cosrole',

'durationSecond' : str(self.ExpireTime),

'Region': 'ap-shanghai',

'Timestamp': str(int(time.time())),

'Nonce': str(int(random.random() * 1000)),

'SecretId': self.SecretId,

}

sortlist = sorted(zip(keydict.keys(), keydict.values()))

return sortlist

def sts_str_sign(self):

sortlist = self.sts_param()

sts_str_init = ''

for value in sortlist:

sts_str_init += str(value[0]) + '=' + str(value[1]) + '&'

sts_str_init = sts_str_init[:-1]

sign_str = 'GET' + self.requestHost + self.requestUri + sts_str_init

return sign_str, sts_str_init

def get_result_url(self):

sign_str, sts_str_init = self.sts_str_sign()

secretkey = self.SecretKey

signature = bytes(sign_str, encoding='utf-8')

secretkey = bytes(secretkey, encoding='utf-8')

my_sign = hmac.new(secretkey, signature, hashlib.sha1).digest()

my_sign = base64.b64encode(my_sign)

result_sign = parse.quote(my_sign)

result_url = 'https://' + self.requestHost + self.requestUri + sts_str_init + '&Signature=' + result_sign

return result_url

if __name__ == '__main__':

SecretId = 'AKID54rSwEDDtOjuy4nQkvt'

SecretKey = 'qUI1zjFWZVGpCPRH'

RoleName = 'qcs::cam::uin/353:roleName/listcvm'

ExpireTime = 3600

STSoper = STS(SecretId, SecretKey, RoleName, ExpireTime)

url = STSoper.get_result_url()

try:

response = requests.get(url)

if response.status_code == 200:

#print(response.text)

data = response.json()

tmpSecretId = data['data']['credentials']['tmpSecretId']

tmpSecretKey = data['data']['credentials']['tmpSecretKey']

sessionToken = data['data']['credentials']['sessionToken']

expiredTime = data['data']['expiration']

print('Get sessionToken: %s' % sessionToken)

print('Get tmpSecretId: %s' % tmpSecretId)

print('Get tmpSecretKey: %s' % tmpSecretKey)

print('Expiration: %s' % expiredTime)

except Exception as e:

print(e)

[/p]

三、临时安全凭证的使用

以列出 CVM 为例，编写脚本调用CVM的API，且将其中的 SecretId 和 SecretKey 更换为上一步中获取到的临时安全凭证 tmpsecretId 和 tmpsecretKey，执行结果：

可以看到已经成功获取（使用的主账号在腾讯云没有服务器，所以没有数据信息）

示例代码

[s][p]

# -*- coding:utf-8 -*-
__author__ = 'wx'
import hashlib
import requests
import hmac
import random
import time
import base64,json
from urllib import  parse
class CVM(object):
    def __init__(self, SecretId, SecretKey, token):
        self.SecretId = SecretId
        self.SecretKey = SecretKey
        self.requestHost = 'cvm.tencentcloudapi.com/'
        self.token = token
    def ketdict(self):
        keydict = {
            'Action' : 'DescribeInstances',
            'Region' : 'ap-shanghai',
            'Timestamp': str(int(time.time())),
            'Nonce': str(int(random.random() * 1000)),
            'SecretId': self.SecretId,
            'Version' : '2017-03-12',
            'Token' : self.token
        }
        sortlist = sorted(zip(keydict.keys(), keydict.values()))
        return sortlist
    def cvm_str_sign(self):
        sortlist = self.ketdict()
        cvm_str_init = ''
        for value in sortlist:
            cvm_str_init += str(value[0]) + '=' + str(value[1]) + '&'
        cvm_str_init = cvm_str_init[:-1]
        sign_str = 'GET' + self.requestHost + '?' + cvm_str_init
        return sign_str, cvm_str_init
    def get_result_url(self):
        sign_str, cvm_str_init = self.cvm_str_sign()
        secretkey = self.SecretKey
        signature = bytes(sign_str, encoding='utf-8')
        secretkey = bytes(secretkey, encoding='utf-8')
        my_sign = hmac.new(secretkey, signature, hashlib.sha1).digest()
        my_sign = base64.b64encode(my_sign)
        result_sign = parse.quote(my_sign)
        result_url = 'https://' + self.requestHost + '?' + cvm_str_init + '&Signature=' + result_sign
        return result_url
if __name__ == '__main__':
    accessid = 'AKIDYreeIOJUUmyhd8IGJ2mr1'
    accesskey = '252s06AGGP70Pb1sRVCAhVh'
    sessionToken = '225013a032f32a12b9455912775dcd6b96'
    CVMoper = CVM(accessid, accesskey, sessionToken)
    url = CVMoper.get_result_url()
    try:
        response = requests.get(url)
        if response.status_code == 200:
            print(response.text)
    except Exception as e:
        print(e)

# -*- coding:utf-8 -*-

__author__ = 'wx'

import hashlib

import requests

import hmac

import random

import time

import base64,json

from urllib import parse

class CVM(object):

def __init__(self, SecretId, SecretKey, token):

self.SecretId = SecretId

self.SecretKey = SecretKey

self.requestHost = 'cvm.tencentcloudapi.com/'

self.token = token

def ketdict(self):

keydict = {

'Action' : 'DescribeInstances',

'Region' : 'ap-shanghai',

'Timestamp': str(int(time.time())),

'Nonce': str(int(random.random() * 1000)),

'SecretId': self.SecretId,

'Version' : '2017-03-12',

'Token' : self.token

}

sortlist = sorted(zip(keydict.keys(), keydict.values()))

return sortlist

def cvm_str_sign(self):

sortlist = self.ketdict()

cvm_str_init = ''

for value in sortlist:

cvm_str_init += str(value[0]) + '=' + str(value[1]) + '&'

cvm_str_init = cvm_str_init[:-1]

sign_str = 'GET' + self.requestHost + '?' + cvm_str_init

return sign_str, cvm_str_init

def get_result_url(self):

sign_str, cvm_str_init = self.cvm_str_sign()

secretkey = self.SecretKey

signature = bytes(sign_str, encoding='utf-8')

secretkey = bytes(secretkey, encoding='utf-8')

my_sign = hmac.new(secretkey, signature, hashlib.sha1).digest()

my_sign = base64.b64encode(my_sign)

result_sign = parse.quote(my_sign)

result_url = 'https://' + self.requestHost + '?' + cvm_str_init + '&Signature=' + result_sign

return result_url

if __name__ == '__main__':

accessid = 'AKIDYreeIOJUUmyhd8IGJ2mr1'

accesskey = '252s06AGGP70Pb1sRVCAhVh'

sessionToken = '225013a032f32a12b9455912775dcd6b96'

CVMoper = CVM(accessid, accesskey, sessionToken)

url = CVMoper.get_result_url()

try:

response = requests.get(url)

if response.status_code == 200:

print(response.text)

except Exception as e:

print(e)

[/p]

四、删除授权

在过期时间之前，如需临时取消授权，在控制台直接删除该角色即可

删除后执行获取 CVM 代码：

发表于：Python

2018-10-12

# python # STS # 腾讯云

复制链接