STS Python_SDK授权临时用户读写OSS资源
名词解释 RAM （Resource Access Management）和STS（Security Token Service）是阿里云提供的权限管理系统。RAM主要的作用是控制账号系统的权限。您可以使用RAM在主账号的权限范围内创建子用户，给不同的子用户分配不同的权限从而达到授权管理的目的。STS是一个安全凭证（Token）的管理系统。您可以使用STS来完成对于临时用户的访问授权。 子账号（RAM account）：从阿里云的主账号中创建出来的子账号，在创建的时候可以分配独立的密码和权限，每个子账号拥有自己AccessKey，可以和阿里云主账号一样正常的完成有权限的操作。一般来说，这里的子账号可以理解为具有某种权限的用户，可以被认为是一个具有某些权限的操作发起者。 角色（Role）：表示某种操作权限的虚拟概念，但是没有独立的登录密码和AccessKey。说明子账号可以扮演角色，扮演角色时候的权限是该角色自身的权限。 授权策略（Policy）：用来定义权限的规则，比如允许用户读取或写入某些资源。 资源（Resource）：代表用户可访问的云资源，比如OSS所有的Bucket、OSS的某个Bucket，或OSS的某个Bucket下面的某个Object等。 扮演角色（Assume role）扮演角色是实体用户获取角色身份的安全令牌的方法。一个实体用户调用STS APIAssumeRole可以获得角色的安全令牌，使用安全令牌可以访问云服务API。 这里将手动定义授权策略（Policy），将授权策略授权给角色 ,然后子账号（RAM account）通过扮演角色_方法 _获取 角色 的 安全令牌对 资源 进行操作.RAM 用户可以使用 API 扮演 RAM 角色。当 RAM 用户被授予AliyunSTSAssumeRoleAccess权限策略之后，可以使用其访问密钥调用 STS APIAssumeRole接口，以获取某个角色的安全令牌，从而使用安全令牌访问资源。 新建用户，角色，授权策略 新建RAM用户 登录RAM 控制台。 在左侧导航栏的人员管理菜单下，单击用户。 单击新建用户，输入登录名称和显示名称。说明单击添加用户，可一次性创建多个 RAM 用户。 在访问方式区域下，选择控制台密码登录或编程访问。 控制台密码登录：可以完成对登录安全的基本设置，包括自动生成或自定义登录密码、是否要求下次登录时重置密码以及是否要求开启多因素认证。 编程访问：将会自动为 RAM 用户创建访问密钥（AccessKey）。RAM 用户可以通过 API 或其他开发工具访问阿里云。 说明为了保障账号安全，建议仅为 RAM 用户选择一种登录方式。避免 RAM 用户离开组织后仍可以通过访问密钥访问阿里云资源。 单击确认。 授权RAM用户AssumeRole接口权限 登录RAM 控制台。 在左侧导航栏的人员管理菜单下，单击用户。 在用户列表中找到刚才创建的用户，单击列表【操作】栏下的【添加权限】 在弹出对话框中的【系统权限策略】搜索并添加_AliyunSTSAssumeRoleAccess_ 单击确认。 新建自定义权限策略test_policy 登录RAM 控制台。 在左侧导航栏的权限管理菜单下，单击权限策略管理。 单击新建权限策略。 填写策略名称和备注。 配置模式选择可视化配置或脚本配置。 若选择可视化配置：单击添加授权语句，根据界面提示，对权限效力、操作名称和资源等进行配置。 若选择脚本配置，请参考语法结构编辑策略内容。 单击确认。 下面为OSS所有权限语法示例 { "Statement": [ { "Action": "oss:*", "Effect": "Allow", "Resource": "*" } ], "Version": "1" } 新建角色testRole 注:这里角色有三种类型,创建可信实体为阿里云账号的RAM角色 云账号登录RAM控制台。 在左侧导航栏，单击RAM角色管理。 单击新建RAM角色。 选择可信实体类型为阿里云账号，单击下一步。 输入角色名称和备注。 选择云账号后，单击完成。 将自定义权限策略授权给角色 登录RAM 控制台。 在左侧导航栏的权限管理菜单下，单击授权。 单击新增授权。 在被授权主体区域下，输入 RAM 角色名称后，单击需要授权的 RAM 角色。 在左侧权限策略名称列表下，单击需要授予 RAM 角色的权限策略。说明在右侧区域框，选择某条策略并单击 ×，可撤销该策略。 单击确定。 完成授权后可在角色管理中查看授权详情 注:这里策略主体类型有两种,权限策略也可直接授权给用户 通过STS完成临时授权 STS基本概念 阿里云临时安全令牌（Security Token Service，STS）是阿里云提供的一种临时访问权限管理服务。 RAM角色（RAM role）一种虚拟的RAM用户。RAM角色的全局资源描述符（Role ARN）Role ARN是角色的全局资源描述符（Aliyun Resource Name，简称ARN），用来指定具体角色。每个角色都有一个唯一的全局资源描述符。格式：acs:ram::$accountID:role/$roleName。 可信实体（Trusted entity）角色的可信实体是指可以扮演角色的实体用户身份。创建角色时必须指定可信实体，角色只能被受信的主体扮演。可信实体可以是受信的阿里云账号、受信的阿里云服务或身份提供商。 扮演角色（Assume role）扮演角色是实体用户获取角色身份的安全令牌的方法。一个实体用户调用STS APIAssumeRole可以获得角色的安全令牌，使用安全令牌可以访问云服务API。 注：这里需要区分下RAM角色与扮演角色的区别，可信实体通过扮演角色_方法___获取 RAM角色的安全令牌 扮演角色的API接口概览 STS提供API调用接口每个请求都需要指定如下信息： 要执行的操作：Action参数。 每个操作接口都需要包含的公共请求参数。 操作接口所特有的请求参数。 调用AssumeRole接口获取一个临时身份。参考API RoleArn表示的是需要扮演的角色ID，角色的ID可以在角色管理>角色详情中找到。 RoleSessionName是一个用来标示临时凭证的名称，一般来说建议使用不同的应用程序用户来区分。 Policy表示的是在扮演角色的时候额外加上的一个权限限制。此参数可以限制生成的STS token的权限，若不指定则返回的token拥有指定角色的所有权限。 DurationSeconds指的是临时凭证的有效期，单位是s，最小为900，最大为3600。 id和secret表示的是需要扮演角色的子账号的AccessKey。 AssumeRole接口请求示例 https://sts.aliyuncs.com?Action=AssumeRole &RoleArn=acs:ram::123456789012****:role/adminrole &RoleSessionName=alice &DurationSeconds=3600 &Policy=<url_encoded_policy> &<公共请求参数> AssumeRole接口返回格式（json） { "Credentials": { "AccessKeyId": "STS.L4aBSCSJVMuKg5U1****", "AccessKeySecret": "wyLTSmsyPGP1ohvvw8xYgB29dlGI8KMiH2pK****", "Expiration": "2015-04-09T11:52:19Z", "SecurityToken": "********" }, "AssumedRoleUser": { "arn": "acs:sts::123456789012****:assumed-role/AdminRole/alice", "AssumedRoleUserId":"34458433936495****:alice" }, "RequestId": "6894B13B-6D71-4EF5-88FA-F32781734A7F" } 通过STS_SDK完成角色扮演 这里将通过官方提供的SDK模块进行具体的角色扮演 安装相关SDK包 pip install oss2pip install aliyun-python-sdk-corepip install aliyun-python-sdk-stssts sdkgithub 地址 获取临时身份信息 # 在控制台将 AliyunSTSAssumeRoleAccess 权限授权给子用户testRole，testRole操作AssumeRole接口，获取临时身份 def fetch_sts_info(access_key_id, access_key_secret, sts_role_arn): """子用户角色扮演获取临时身份的信息 :param access_key_id: 子用户的 access key id :param access_key_secret: 子用户的 access key secret :param sts_role_arn: STS角色的Arn :return StsInfo 返回临时身份信息对象 """ # 配置要访问的STS endpoint REGIONID = 'cn-hongkong' ENDPOINT = 'sts.cn-hongkong.aliyuncs.com' region_provider.add_endpoint('Sts', REGIONID, ENDPOINT) clt = client.AcsClient(access_key_id, access_key_secret, 'cn-hongkong') request = AssumeRoleRequest.AssumeRoleRequest() #request.set_accept_format('json') #指定角色ARN request.set_RoleArn(sts_role_arn) #设置会话名称，审计服务使用此名称区分调用者 request.set_RoleSessionName('oss-python-sdk-example') #发起请求，并得到response response = clt.do_action_with_exception(request) #格式化输出返回结果，将字符串结果转化为字典类型 i = json.loads(oss2.to_unicode(response)) #实例化StsInfo类，并将通过sts获取的临时身份信息存入 global StsInfo StsInfo = StsInfo() StsInfo.access_key_id = i['Credentials']['AccessKeyId'] StsInfo.access_key_secret = i['Credentials']['AccessKeySecret'] StsInfo.security_token = i['Credentials']['SecurityToken'] StsInfo.request_id = i['RequestId'] StsInfo.expiration = oss2.utils.to_unixtime(i['Credentials']['Expiration'], '%Y-%m-%dT%H:%M:%SZ') #返回StsInfo对象 return StsInfo 将临时身份信息存入json文件 根据需求，可将临时身份信息存储到json文件中，等到临时身份过期后再重新请求，避免重复请求，造成临时身份泛滥，引发资源匮乏与安全隐患 def save_info(): #存储临时身份信息 with open('StsInfo.json','w',encoding='utf-8') as f: data = {'sts_key_id':StsInfo.access_key_id,'sts_key_secret':StsInfo.access_key_secret,'sts_secrity_token':StsInfo.security_token,'sts_expire_date':StsInfo.expiration,'sts_reques_id':StsInfo.request_id} json.dump(data,f,ensure_ascii=False) f.close def open_info(): #读取临时身份信息 with open('./StsInfo.json','r',encoding='utf-8') as f: global STSINFO STSINFO = json.load(f) return STSINFO 通过临时身份操作Bucket资源 def buck_put_object(): #打印验证临时身份信息 print(StsInfo) print('key id:',StsInfo.access_key_id) print("key_secret:",StsInfo.access_key_secret) print("secrity_token:",StsInfo.security_token) print("request_id:",StsInfo.request_id) print("expiration:",StsInfo.expiration) #实例化Bucket对象，并上传字符串 auth = oss2.StsAuth(StsInfo.access_key_id, StsInfo.access_key_secret, StsInfo.security_token) bucket = oss2.Bucket(auth,endpoint,'fralychen') result = bucket.put_object('fralychen','good good study day day up') 全部代码 # -*- conding:utf-8 -*- import json import os import oss2 from aliyunsdkcore import client from aliyunsdkcore.profile import region_provider from aliyunsdksts.request.v20150401 import AssumeRoleRequest ##定义一些变量 access_key_id = 'LTAI4FoMe6umpCSFQdEC9neg' access_key_secret = 'jgEvtFOqIAGqKZve7zMvg8dJhSZv9J' bucket_name = 'fralychen' endpoint = 'oss-cn-hongkong.aliyuncs.com' sts_role_arn = 'acs:ram::1149877324567510:role/testrole' # 确认上面的参数都填写正确了 for param in (access_key_id, access_key_secret, bucket_name, endpoint, sts_role_arn): assert '<' not in param, '请设置参数：' + param #创建StsToken类方便用来存储临时身份信息 class StsInfo(object): """AssumeRole返回的临时身份信息 :param str access_key_id: 临时身份的access key id :param str access_key_secret: 临时身份的access key secret :param int expiration: 过期时间，UNIX时间，自1970年1月1日UTC零点的秒数 :param str security_token: 临时身份Token :param str request_id: 请求ID """ def __init__(self): self.access_key_id = '' self.access_key_secret = '' self.expiration = 0 self.security_token = '' self.request_id = '' # 在控制台将 AliyunSTSAssumeRoleAccess 权限授权给RAM子用户之后才能通过RAM用户获取临时身份信息 def fetch_sts_info(access_key_id, access_key_secret, sts_role_arn): """子用户角色扮演获取临时身份的信息 :param access_key_id: 子用户的 access key id :param access_key_secret: 子用户的 access key secret :param sts_role_arn: STS角色的Arn :return StsInfo 返回临时身份信息对象 """ # 配置要访问的STS endpoint _REGIONID = 'cn-hongkong' _ENDPOINT = 'sts.cn-hongkong.aliyuncs.com' region_provider.add_endpoint('Sts', _REGIONID, _ENDPOINT) clt = client.AcsClient(access_key_id, access_key_secret, 'cn-hongkong') request = AssumeRoleRequest.AssumeRoleRequest() #request.set_accept_format('json') #指定角色ARN request.set_RoleArn(sts_role_arn) #设置会话名称，审计服务使用此名称区分调用者 request.set_RoleSessionName('oss-python-sdk-example') #设置临时身份过期时间 request.set_DurationSeconds(DurationSeconds) #发起请求，并得到response response = clt.do_action_with_exception(request) #格式化输出返回结果，将字符串结果转化为字典类型 i = json.loads(oss2.to_unicode(response)) #实例化StsInfo类并将临时身份信息存入对象 global StsInfo StsInfo = StsInfo() StsInfo.access_key_id = i['Credentials']['AccessKeyId'] StsInfo.access_key_secret = i['Credentials']['AccessKeySecret'] StsInfo.security_token = i['Credentials']['SecurityToken'] StsInfo.request_id = i['RequestId'] StsInfo.expiration = oss2.utils.to_unixtime(i['Credentials']['Expiration'], '%Y-%m-%dT%H:%M:%SZ') #存储临时身份信息 save_info() #使用sts授权的临时身份上传文件到bucket def buck_put_object(sts_key_id, sts_key_secret, sts_secrity_token): """上传字符串到资源 :param sts_key_id: 临时身份的 access key id :param sts_key_secret: 临时身份的 access key secret :param sts_secrity_token: 临时身份的 secrity token :retu """ #实例化Bucket对象，并上传字符串 auth = oss2.StsAuth(sts_key_id, sts_key_secret, sts_secrity_token) bucket = oss2.Bucket(auth,endpoint,'fralychen') result = bucket.put_object('fralychen','good good study day day up') #根据需求，可将临时身份信息存储到json文件中，等到临时身份过期后再重新请求，避免重复请求，用户泛滥 def save_info(): #存储临时身份信息 with open('StsInfo.json','w',encoding='utf-8') as f: data = {'sts_key_id':StsInfo.access_key_id,'sts_key_secret':StsInfo.access_key_secret,'sts_secrity_token':StsInfo.security_token,'sts_expire_date':StsInfo.expiration,'sts_reques_id':StsInfo.request_id} json.dump(data,f,ensure_ascii=False) f.close def open_info(): #读取临时身份信息 with open('./StsInfo.json','r',encoding='utf-8') as f: global STSINFO STSINFO = json.load(f) return STSINFO #定义临时身份过期时间 DurationSeconds = 900 try: open_info() except IOError: print("Error: 没有用户信息文件或文件读取失败") print("初始化身份信息，临时身份信息存储中....") fetch_sts_info(access_key_id, access_key_secret, sts_role_arn) print("临时身份信息存储完毕，当前目录下StsInfo.json") else: if oss2.utils.http_to_unixtime(oss2.utils.http_date()) + DurationSeconds > STSINFO["sts_expire_date"]: buck_put_object(sts_key_id = STSINFO["sts_key_id"],sts_key_secret = STSINFO["sts_key_secret"], sts_secrity_token = STSINFO["sts_secrity_token"]) print("上传成功，good_lucky") else: print("更新临时用户信息，请稍后") fetch_sts_info(access_key_id, access_key_secret, sts_role_arn) buck_put_object(sts_key_id = STSINFO["sts_key_id"],sts_key_secret = STSINFO["sts_key_secret"], sts_secrity_token = STSINFO["sts_secrity_token"]) print("上传成功，good_lucky")
