全部产品
Search
文档中心

对象存储 OSS:Python授权访问

更新时间:Mar 13, 2024

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

注意事项

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

  • 本文以华东1(杭州)外网Endpoint为例。如果您希望通过与OSS同地域的其他阿里云产品访问OSS,请使用内网Endpoint。关于OSS支持的Region与Endpoint的对应关系,请参见访问域名和数据中心

  • 本文以从环境变量读取访问凭证为例。如何配置访问凭证,请参见配置访问凭证

  • 本文以OSS域名新建OSSClient为例。如果您希望通过自定义域名、STS等方式新建OSSClient,请参见初始化

使用STS进行临时授权

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

STS的优势如下:

  • 您无需透露您的长期密钥(AccessKey)给第三方应用,只需生成一个访问令牌并将令牌交给第三方应用。您可以自定义这个令牌的访问权限及有效期限。

  • 您无需关心权限撤销问题,访问令牌过期后自动失效。

通过STS临时授权访问OSS的步骤如下:

步骤1:创建RAM用户。

  1. 登录RAM控制台

  2. 在左侧导航栏,选择身份管理>用户

  3. 单击创建用户

  4. 输入登录名称显示名称

  5. 访问方式区域下,选择OpenAPI调用访问,然后单击确定

  6. 单击复制,保存访问密钥(AccessKey ID 和 AccessKey Secret)。

步骤2:为RAM用户授予请求AssumeRole的权限。

  1. 单击已创建RAM用户右侧对应的添加权限

  2. 添加权限页面,选择AliyunSTSAssumeRoleAccess系统策略。image.png

  3. 单击确定

步骤3:创建用于获取临时访问凭证的角色。

  1. 在左侧导航栏,选择身份管理>角色

  2. 单击创建角色,选择可信实体类型为阿里云账号,单击下一步

  3. 创建角色对话框,角色名称填写为RamOssTest,选择信任的云账号当前云账号

  4. 单击完成。角色创建完成后,单击关闭

  5. 角色页面,搜索框输入角色名称RamOssTest,然后单击RamOssTest

  6. 单击ARN右侧的复制,保存角色的ARN。arn

步骤4:为角色授予上传和下载文件的权限。

  1. 为角色授予上传文件和下载文件的自定义权限策略。

    1. 在左侧导航栏,选择权限管理>权限策略

    2. 权限策略页面,单击创建权限策略

    3. 创建权限策略页面,单击脚本编辑,然后在策略文档输入框中赋予角色向目标存储空间examplebucket执行简单上传(oss:PutObject)和下载文件(oss:GetObject)的权限。具体配置示例如下。

      警告

      以下示例仅供参考。您需要根据实际需求配置更细粒度的授权策略,防止出现权限过大的风险。关于更细粒度的授权策略配置详情,请参见通过RAM或STS服务向其他用户授权

      {
          "Version": "1",
          "Statement": [
           {
                 "Effect": "Allow",
                 "Action": [
                   "oss:PutObject",
                   "oss:GetObject"
                 ],
                 "Resource": [
                   "acs:oss:*:*:examplebucket",
                   "acs:oss:*:*:examplebucket/*"
                 ]
           }
          ]
      }
    4. 策略配置完成后,单击继续编辑基本信息

    5. 基本信息区域,填写策略名称RamTestPolicy,然后单击确定

  2. 为RAM角色RamOssTest授予自定义权限策略。

    1. 在左侧导航栏,选择身份管理 > 角色

    2. 角色页面,找到目标RAM角色RamOssTest

    3. 单击RAM角色RamOssTest右侧的添加权限

    4. 添加权限页面下的自定义策略页签,选择已创建的自定义权限策略RamTestPolicy

    5. 单击确定

步骤5:生成临时访问凭证。

临时访问凭证包括临时访问密钥(AccessKey ID和AccessKey Secret)和安全令牌(SecurityToken)。临时访问凭证有效时间单位为秒,最小值为900,最大值以当前角色设定的最大会话时间为准。更多信息,请参见设置RAM角色最大会话时间

关于STS应用的完整示例代码,请参见GitHub

重要

获取临时访问凭证前,您需要使用pip install aliyun-python-sdk-sts命令安装官方的Python STS客户端。

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

from aliyunsdkcore import client
from aliyunsdkcore.request import CommonRequest
import json
import oss2

# yourEndpoint填写Bucket所在地域对应的Endpoint。以华东1(杭州)为例,Endpoint填写为https://oss-cn-hangzhou.aliyuncs.com。
endpoint = 'https://oss-cn-hangzhou.aliyuncs.com'
# 填写步骤1创建的RAM用户AccessKey。
access_key_id = 'yourAccessKeyId'
access_key_secret = 'yourAccessKeySecret'
# role_Arn填写步骤3获取的角色ARN,例如acs:ram::175708322470****:role/ramtest。
role_arn = 'acs:ram::175708322470****:role/ramtest'

# 创建权限策略。
# 仅允许对examplebucket执行上传(PutObject)和下载(GetObject)操作。
policy_text = '{"Version": "1", "Statement": [{"Action": ["oss:PutObject","oss:GetObject"], "Effect": "Allow", "Resource": ["acs:oss:*:*:examplebucket/*"]}]}'

clt = client.AcsClient(access_key_id, access_key_secret, 'cn-hangzhou')
request = CommonRequest(product="Sts", version='2015-04-01', action_name='AssumeRole')
request.set_method('POST')
request.set_protocol_type('https')
request.add_query_param('RoleArn', role_arn)
# 指定自定义角色会话名称,用来区分不同的令牌,例如填写为sessiontest。
request.add_query_param('RoleSessionName', 'sessiontest')
# 指定临时访问凭证有效时间单位为秒,最小值为900,最大值为3600。
request.add_query_param('DurationSeconds', '3000')
# 如果policy为空,则RAM用户默认获得该角色下所有权限。如果有特殊权限控制要求,请参考上述policy_text设置。
request.add_query_param('Policy', policy_text)
request.set_accept_format('JSON')

body = clt.do_action_with_exception(request)

# 使用RAM用户的AccessKey ID和AccessKey Secret向STS申请临时访问凭证。
token = json.loads(oss2.to_unicode(body))
# 打印STS返回的临时访问密钥(AccessKey ID和AccessKey Secret)、安全令牌(SecurityToken)以及临时访问凭证过期时间(Expiration)。
print('AccessKeyId: '+token['Credentials']['AccessKeyId'])
print('AccessKeySecret: '+token['Credentials']['AccessKeySecret'])
print('SecurityToken: '+token['Credentials']['SecurityToken'])
print('Expiration: '+token['Credentials']['Expiration'])

步骤6:使用临时访问凭证上传和下载文件。

  • 使用临时访问凭证上传文件

    # -*- coding: utf-8 -*-
    import oss2
    
    # yourEndpoint填写Bucket所在地域对应的Endpoint。以华东1(杭州)为例,Endpoint填写为https://oss-cn-hangzhou.aliyuncs.com。
    endpoint = 'https://oss-cn-hangzhou.aliyuncs.com'
    # 从STS服务获取的临时访问密钥(AccessKey ID和AccessKey Secret)。
    sts_access_key_id = 'yourAccessKeyId'
    sts_access_key_secret = 'yourAccessKeySecret'
    # 填写Bucket名称。
    bucket_name = 'examplebucket'
    # 填写Object完整路径和字符串。Object完整路径中不能包含Bucket名称。
    object_name = 'exampleobject.txt'
    # 从STS服务获取的安全令牌(SecurityToken)。
    security_token = 'yourSecurityToken'
    
    
    # 使用临时访问凭证中的认证信息初始化StsAuth实例。
    auth = oss2.StsAuth(sts_access_key_id,
                        sts_access_key_secret,
                        security_token)
    
    # 使用StsAuth实例初始化存储空间。
    bucket = oss2.Bucket(auth, endpoint, bucket_name)
    
    # 上传Object。
    result = bucket.put_object(object_name, "hello world")
    print(result.status)
  • 使用临时访问凭证下载文件

    # -*- coding: utf-8 -*-
    import oss2
    
    # yourEndpoint填写Bucket所在地域对应的Endpoint。以华东1(杭州)为例,Endpoint填写为https://oss-cn-hangzhou.aliyuncs.com。
    endpoint = 'https://oss-cn-hangzhou.aliyuncs.com'
    # 从STS服务获取的临时访问密钥(AccessKey ID和AccessKey Secret)。
    sts_access_key_id = 'yourAccessKeyId'
    sts_access_key_secret = 'yourAccessKeySecret'
    # 填写Bucket名称,例如examplebucket。
    bucket_name = 'examplebucket'
    # 填写Object完整路径和字符串。Object完整路径中不能包含Bucket名称。
    object_name = 'exampleobject.txt'
    # 从STS服务获取的安全令牌(SecurityToken)。
    security_token = 'yourSecurityToken'
    
    # 使用临时访问凭证中的认证信息初始化StsAuth实例。
    auth = oss2.StsAuth(sts_access_key_id,
                        sts_access_key_secret,
                        security_token)
    
    # 使用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转给第三方实现授权访问。具体操作,请参见签名版本1

生成带versionId的签名URL

以下代码用于生成带versionId的签名URL。

# -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
import requests

# 从环境变量中获取访问凭证。运行本代码示例之前,请确保已设置环境变量OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())

# yourEndpoint填写Bucket所在地域对应的Endpoint。以华东1(杭州)为例,Endpoint填写为https://oss-cn-hangzhou.aliyuncs.com。
# 填写Bucket名称,例如examplebucket。
bucket = oss2.Bucket(auth, 'https://oss-cn-hangzhou.aliyuncs.com', 'examplebucket')
# 填写Object完整路径,例如exampledir/exampleobject.txt。Object完整路径中不能包含Bucket名称。
object_name = 'exampledir/exampleobject.txt'

# 指定Header。
headers = dict()
# 填写Object的versionId。
headers["versionId"] = "CAEQARiBgID8rumR2hYiIGUyOTAyZGY2MzU5MjQ5ZjlhYzQzZjNlYTAyZDE3****"

# 生成上传文件的签名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上传文件

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

# -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
import requests

# 从环境变量中获取访问凭证。运行本代码示例之前,请确保已设置环境变量OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())

# yourEndpoint填写Bucket所在地域对应的Endpoint。以华东1(杭州)为例,Endpoint填写为https://oss-cn-hangzhou.aliyuncs.com。
# 填写Bucket名称,例如examplebucket。
bucket = oss2.Bucket(auth, 'https://oss-cn-hangzhou.aliyuncs.com', '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
from oss2.credentials import EnvironmentVariableCredentialsProvider
import requests

# 从环境变量中获取访问凭证。运行本代码示例之前,请确保已设置环境变量OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())

# yourEndpoint填写Bucket所在地域对应的Endpoint。以华东1(杭州)为例,Endpoint填写为https://oss-cn-hangzhou.aliyuncs.com。
# 填写Bucket名称,例如examplebucket。
bucket = oss2.Bucket(auth, 'https://oss-cn-hangzhou.aliyuncs.com', '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-t4nlw426y44rd3iq4xxxx"
# 指定是否允许转发请求。
# 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)