本文介绍如何使用STS以及签名URL临时授权访问OSS资源。

注意 由于STS临时账号以及签名URL均需设置有效时长,当您使用STS临时账号生成签名URL执行相关操作(例如上传、下载文件)时,以最小的有效时长为准。例如您的STS临时账号的有效时长设置为1200秒、签名URL设置为3600秒时,当有效时长超过1200秒后,您无法使用此STS临时账号生成的签名URL上传文件。

使用STS进行临时授权

OSS可以通过阿里云STS(Security Token Service)进行临时授权访问。阿里云STS是为云计算用户提供临时访问令牌的Web服务。通过STS,您可以为第三方应用或子用户(即用户身份由您自己管理的用户)颁发一个自定义时效和权限的访问凭证。关于STS的更多信息,请参见STS介绍

STS的优势如下:

  • 您无需透露您的长期密钥(AccessKey)给第三方应用,只需生成一个访问令牌并将令牌交给第三方应用。您可以自定义这个令牌的访问权限及有效期限。
  • 您无需关心权限撤销问题,访问令牌过期后自动失效。
注意 关于搭建STS服务的具体操作,请参见使用STS临时访问凭证访问OSS。您可以通过调用STS服务的AssumeRole接口或者使用各语言STS SDK来获取临时访问凭证。临时访问凭证包括临时访问密钥(AccessKeyId和AccessKeySecret)和安全令牌(SecurityToken)。临时访问凭证有效时间单位为秒,最小值为900,最大值以当前角色设定的最大会话时间为准。更多信息,请参见设置角色最大会话时间

执行pip install aliyun-python-sdk-sts命令,安装官方的Python STS客户端。关于STS应用的完整示例代码,请参见GitHub

请确保使用2.0.6及以上版本SDK。以下代码用于使用STS临时授权下载文件。

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

from aliyunsdkcore import client
from aliyunsdksts.request.v20150401 import AssumeRoleRequest
import json
import oss2

# yourEndpoint填写Bucket所在地域对应的Endpoint。以华东1(杭州)为例,Endpoint填写为https://oss-cn-hangzhou.aliyuncs.com。
endpoint = 'yourEndpoint'
# 阿里云账号AccessKey拥有所有API的访问权限,风险很高。强烈建议您创建并使用RAM用户进行API访问或日常运维,请登录RAM控制台创建RAM用户。
access_key_id = 'yourAccessKeyId'
access_key_secret = 'yourAccessKeySecret'
# 填写Bucket名称,例如examplebucket。
bucket_name = 'examplebucket'
# 填写Object完整路径,例如exampledir/exampleobject.txt。Object完整路径中不能包含Bucket名称。
object_name = 'exampledir/exampleobject.txt'
# 您可以登录RAM控制台,在RAM角色管理页面,搜索创建的RAM角色后,单击RAM角色名称,在RAM角色详情界面查看和复制角色的ARN信息。
# 填写角色的ARN信息。格式为acs:ram::$accountID:role/$roleName。
# $accountID为阿里云账号ID。您可以通过登录阿里云控制台,将鼠标悬停在右上角头像的位置,直接查看和复制账号ID,或者单击基本资料查看账号ID。
# $roleName为RAM角色名称。您可以通过登录RAM控制台,单击左侧导航栏的RAM角色管理,在RAM角色名称列表下进行查看。
role_arn = 'acs:ram::17464958********:role/ossststest'

# 创建权限策略。
# 只允许对名称为examplebucket的Bucket下的所有资源执行GetObject操作。
policy_text = '{"Version": "1", "Statement": [{"Action": ["oss:GetObject"], "Effect": "Allow", "Resource": ["acs:oss:*:*:examplebucket/*"]}]}'

clt = client.AcsClient(access_key_id, access_key_secret, 'cn-hangzhou')
req = AssumeRoleRequest.AssumeRoleRequest()

# 设置返回值格式为JSON。
req.set_accept_format('json')
req.set_RoleArn(role_arn)
# 自定义角色会话名称,用来区分不同的令牌,例如可填写为session-test。
req.set_RoleSessionName('session-test')
req.set_Policy(policy_text)
body = clt.do_action_with_exception(req)

# 使用RAM用户的AccessKeyId和AccessKeySecret向STS申请临时访问凭证。
token = json.loads(oss2.to_unicode(body))

# 使用临时访问凭证中的认证信息初始化StsAuth实例。
auth = oss2.StsAuth(token['Credentials']['AccessKeyId'],
                    token['Credentials']['AccessKeySecret'],
                    token['Credentials']['SecurityToken'])

# 使用StsAuth实例初始化存储空间。
bucket = oss2.Bucket(auth, endpoint, bucket_name)

# 下载Object。
read_obj = bucket.get_object(object_name)
print(read_obj.read())            

使用签名URL进行临时授权

以下介绍使用签名URL临时授权的常见示例。

您可以将生成的签名URL提供给访客进行临时访问。生成签名URL时,您可以自定义URL的过期时间来限制访客的访问时长。

注意 通过以下示例生成的签名URL中如果包含特殊符号+,可能出现无法正常访问该签名URL的现象。如需正常访问该签名URL,请将签名URL中的+替换为%2B

在URL中加入签名信息,以便将该URL转给第三方实现授权访问。具体操作,请参见在URL中包含签名

  • 使用签名URL上传文件

    以下代码用于生成上传的签名URL,并使用签名URL上传文件。

    # -*- coding: utf-8 -*-
    import oss2
    import requests
    
    # 阿里云账号AccessKey拥有所有API的访问权限,风险很高。强烈建议您创建并使用RAM用户进行API访问或日常运维,请登录RAM控制台创建RAM用户。
    auth = oss2.Auth('yourAccessKeyId', 'yourAccessKeySecret')
    # 如果使用STS授权,则填写从STS服务获取的临时访问密钥(AccessKey ID和AccessKey Secret)以及安全令牌(SecurityToken)。
    # auth = oss2.StsAuth('yourAccessKeyId', 'yourAccessKeySecret', 'yourToken')
    
    # yourEndpoint填写Bucket所在地域对应的Endpoint。以华东1(杭州)为例,Endpoint填写为https://oss-cn-hangzhou.aliyuncs.com。
    # 填写Bucket名称,例如examplebucket。
    bucket = oss2.Bucket(auth, 'yourEndpoint', 'examplebucket')
    # 填写Object完整路径,例如exampledir/exampleobject.txt。Object完整路径中不能包含Bucket名称。
    object_name = 'exampledir/exampleobject.txt'
    
    # 指定Header。
    headers = dict()
    # 指定Content-Type。
    # headers['Content-Type'] = 'text/txt'
    # 指定存储类型。
    # headers["x-oss-storage-class"] = "Standard"
    
    # 生成上传文件的签名URL,有效时间为60秒。
    # 生成签名URL时,OSS默认会对Object完整路径中的正斜线(/)进行转义,从而导致生成的签名URL无法直接使用。
    # 设置slash_safe为True,OSS不会对Object完整路径中的正斜线(/)进行转义,此时生成的签名URL可以直接使用。
    url = bucket.sign_url('PUT', object_name, 60, slash_safe=True, headers=headers)
    print('签名URL的地址为:', url)
    
    # 通过签名URL上传文件,以requests为例说明。
    # 填写本地文件路径,例如D:\\exampledir\\examplefile.txt。
    requests.put(url, data=open('D:\\exampledir\\examplefile.txt', 'rb').read(), headers=headers)       
  • 使用签名URL下载文件

    以下代码用于生成下载的签名URL,并使用签名URL下载文件。

    # -*- coding: utf-8 -*-
    import oss2
    import requests
    
    # 阿里云账号AccessKey拥有所有API的访问权限,风险很高。强烈建议您创建并使用RAM用户进行API访问或日常运维,请登录RAM控制台创建RAM用户。
    auth = oss2.Auth('yourAccessKeyId', 'yourAccessKeySecret')
    # 如果使用STS授权,则填写从STS服务获取的临时访问密钥(AccessKey ID和AccessKey Secret)以及安全令牌(SecurityToken)。
    # auth = oss2.StsAuth('yourAccessKeyId', 'yourAccessKeySecret', 'yourToken')
    
    # yourEndpoint填写Bucket所在地域对应的Endpoint。以华东1(杭州)为例,Endpoint填写为https://oss-cn-hangzhou.aliyuncs.com。
    # 填写Bucket名称,例如examplebucket。
    bucket = oss2.Bucket(auth, 'yourEndpoint', 'examplebucket')
    # 填写Object完整路径,例如exampledir/exampleobject.txt。Object完整路径中不能包含Bucket名称。
    object_name = 'exampledir/exampleobject.txt'
    
    # 指定Header。
    headers = dict()
    # 指定Accept-Encoding。
    headers['Accept-Encoding'] = 'gzip'
    
    # 指定HTTP查询参数。
    params = dict()
    # 设置单链接限速,单位为bit,例如限速100 KB/s。
    # params['x-oss-traffic-limit'] = str(100 * 1024 * 8)
    # 指定IP地址或者IP地址段。
    # params['x-oss-ac-source-ip'] = "127.0.0.1"
    # 指定子网掩码中1的个数。
    # params['x-oss-ac-subnet-mask'] = "32"
    # 指定VPC ID。
    # params['x-oss-ac-vpc-id'] = "vpc-t4nlw426y44rd3iq4****"
    # 指定是否允许转发请求。
    # params['x-oss-ac-forward-allow'] = "true"
    
    # 生成上传文件的签名URL,有效时间为60秒。
    # 生成签名URL时,OSS默认会对Object完整路径中的正斜线(/)进行转义,从而导致生成的签名URL无法直接使用。
    # 设置slash_safe为True,OSS不会对Object完整路径中的正斜线(/)进行转义,此时生成的签名URL可以直接使用。
    url = bucket.sign_url('GET', object_name, 60, slash_safe=True, headers=headers, params=params)
    print('签名URL的地址为:', url)
    
    # 通过签名URL下载文件,以requests为例说明。
    resp = requests.get(url, headers=headers)
    
    # 填写本地文件路径,例如D:\\exampledir\\examplefile.txt。
    with open("D:\\exampledir\\examplefile.txt", "wb") as code:
        code.write(resp.content)