開發與維運

STS Python_SDK授權臨時用戶讀寫OSS資源

2020-04-21

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 API AssumeRole可以獲得角色的安全令牌，使用安全令牌可以訪問雲服務API。

這裡將手動定義 授權策略（Policy），將 授權策略 授權給 角色 ,然後子賬號（RAM account）通過 扮演角色_方法 _獲取 角色 的 安全令牌 對 資源 進行操作.
RAM 用戶 可以使用 API 扮演 RAM 角色。當 RAM 用戶被授予 AliyunSTSAssumeRoleAccess 權限策略 之後，可以使用其訪問密鑰調用 STS API AssumeRole 接口，以獲取某個角色的 安全令牌，從而使用安全令牌訪問資源。

新建用戶，角色，授權策略

新建RAM用戶

  1. 登錄 RAM 控制檯。
  2. 在左側導航欄的人員管理菜單下，單擊用戶。
  3. 單擊新建用戶，輸入登錄名稱和顯示名稱。
    說明 單擊添加用戶，可一次性創建多個 RAM 用戶。

  4. 在訪問方式區域下，選擇控制檯密碼登錄或編程訪問。

    • 控制檯密碼登錄：可以完成對登錄安全的基本設置，包括自動生成或自定義登錄密碼、是否要求下次登錄時重置密碼以及是否要求開啟多因素認證。
    • 編程訪問：將會自動為 RAM 用戶創建訪問密鑰（AccessKey）。RAM 用戶可以通過 API 或其他開發工具訪問阿里雲。
  5. 說明 為了保障賬號安全，建議僅為 RAM 用戶選擇一種登錄方式。避免 RAM 用戶離開組織後仍可以通過訪問密鑰訪問阿里雲資源。
  6. 單擊確認。

授權RAM用戶AssumeRole接口權限

  1. 登錄 RAM 控制檯。
  2. 在左側導航欄的人員管理菜單下，單擊用戶。
  3. 在用戶列表中找到剛才創建的用戶，單擊列表【操作】欄下的【添加權限】
  4. 在彈出對話框中的【系統權限策略】搜索並添加 _AliyunSTSAssumeRoleAccess_
  5. 單擊確認。

新建自定義權限策略test_policy

  1. 登錄 RAM 控制檯。
  2. 在左側導航欄的權限管理菜單下，單擊權限策略管理。
  3. 單擊新建權限策略。
  4. 填寫策略名稱和備註。

  5. 配置模式選擇可視化配置或腳本配置。

    • 若選擇可視化配置：單擊添加授權語句，根據界面提示，對權限效力、操作名稱和資源等進行配置。
    • 若選擇腳本配置，請參考語法結構編輯策略內容。
  6. 單擊確認。

下面為OSS所有權限語法示例

{
    "Statement": [
        {
            "Action": "oss:*",
            "Effect": "Allow",
            "Resource": "*"
        }
    ],
    "Version": "1"
}

新建角色testRole

注:這裡角色有三種類型,創建可信實體為阿里雲賬號的RAM角色
  1. 雲賬號登錄RAM控制檯。
  2. 在左側導航欄，單擊RAM角色管理。
  3. 單擊新建RAM角色。
  4. 選擇可信實體類型為阿里雲賬號，單擊下一步。
  5. 輸入角色名稱和備註。
  6. 選擇雲賬號後，單擊完成。

將自定義權限策略授權給角色

  1. 登錄 RAM 控制檯。
  2. 在左側導航欄的權限管理菜單下，單擊授權。
  3. 單擊新增授權。
  4. 在被授權主體區域下，輸入 RAM 角色名稱後，單擊需要授權的 RAM 角色。
  5. 在左側權限策略名稱列表下，單擊需要授予 RAM 角色的權限策略。
    說明 在右側區域框，選擇某條策略並單擊 ×，可撤銷該策略。
  6. 單擊確定。

完成授權後可在角色管理中查看授權詳情
image.png

注:這裡策略主體類型有兩種,權限策略也可直接授權給用戶    

image.png

通過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 API AssumeRole可以獲得角色的安全令牌，使用安全令牌可以訪問雲服務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 oss2
pip install aliyun-python-sdk-core
pip install aliyun-python-sdk-sts
sts sdk github 地址

 獲取臨時身份信息

# 在控制檯將 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")

join US

未標題-1.png

image.png
暴走小二官方群，歡迎來躁

Leave a Reply

Your email address will not be published. Required fields are marked *