OSS可以通过阿里云STS（Security Token Service）进行临时授权访问。通过STS，您可以为第三方应用或子用户（即用户身份由您自己管理的用户）颁发一个自定义时效和权限的访问凭证。

实现逻辑：前端用户点击上传图片， 前端向后端发起请求，后端返回一个签名后的url ，前端拿到签名后url再进行用Put方法

调用OSS的SDK时报“SignatureDoesNotMatch”错误， 由于前端直接向签名url put 存在跨域问题，需要把oss的跨域给打开， 使用：set_oss_cors 方法

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @File  : sts_token.py
# @Author: 往事随风
# @Email: gujiwork@outlook.com
# @Date  : 2020/12/24
# @Desc  :
# pip install aliyun-python-sdk-sts
# pip install oss2
from aliyunsdkcore import client
from aliyunsdksts.request.v20150401 import AssumeRoleRequest
import json
import oss2
import requests
from oss2.models import BucketCors, CorsRule


class AliStsGenerateToKey:
    def __init__(self, endpoint, access_key_id, access_key_secret, bucket_name):
        """
        :param endpoint: 地域
        :param access_key_id: ram子用户key
        :param access_key_secret: ram子用户 secret
        :param bucket_name: oss bucket名称
        """
        self.endpoint = endpoint
        self.access_key_id = access_key_id
        self.access_key_secret = access_key_secret
        self.bucket_name = bucket_name

    @staticmethod
    def generate_sts_key(self, role_arn):
        object_name = ''
        # 设置sts token获取的权限策略, action及resource
        policy_text = '{"Statement": [{"Action": ["*"],"Effect": "Allow","Resource": ["*"]}],"Version":"1"}'
        # policy_text = '{"Statement": [{"Action": ["oss:GetObject"],"Effect": "Allow","Resource": ["acs:oss:*:*:oss-test/*"]}],"Version":"1"}'
        clt = client.AcsClient(self.access_key_id, self.access_key_secret, 'cn-hangzhou')
        req = AssumeRoleRequest.AssumeRoleRequest()
        req.set_accept_format('json')
        req.set_RoleArn(role_arn)
        # 设置会话名称，审计服务使用此名称区分调用者
        req.set_RoleSessionName('test')
        req.set_Policy(policy_text)
        body = clt.do_action_with_exception(req)

        return body

    def set_oss_sign(self, role_arn, save_oss_file_name, exp_time, upload_file):
        """
        :param role_arn: 角色的资源名称
        :param save_oss_file_name:  保存到oss目录/文件名  settlement_excel_download/20200812200526.png
        :param exp_time: 过期时间
        :param upload_file: 本地上传文件名
        :return: 签名后的url地址
        """
        body = AliStsGenerateToKey.generate_sts_key(self, role_arn=role_arn)

        # 使用RAM账号的AccessKeyId和AccessKeySecret向STS申请临时token。
        token = json.loads(oss2.to_unicode(body))

        # 使用临时token中的认证信息初始化StsAuth实例。
        auth = oss2.StsAuth(token['Credentials']['AccessKeyId'],
                            token['Credentials']['AccessKeySecret'],
                            token['Credentials']['SecurityToken'])
        # 使用StsAuth实例初始化存储空间。
        bucket = oss2.Bucket(auth, self.endpoint, self.bucket_name)

        sign_url_put = bucket.sign_url('PUT', save_oss_file_name, exp_time)
        print(sign_url_put)
        rsp = requests.put(url=sign_url_put, data=open(upload_file, 'rb'))
        if rsp.status_code == 200:
            print('请求上传文件返回状态码： 200')

        file_exists = bucket.object_exists(key=save_oss_file_name)
        if file_exists:
            sign_url = bucket.sign_url('GET', save_oss_file_name, exp_time)
            print(sign_url)
            if requests.get(url=sign_url).status_code != 200:
                return False, '设置签名未生效或已过期, 请检查oss是否设置跨域、 oss权限及sts权限设置是否正确、过期策略时间！'
            return True, sign_url
        print('文件上传失败！')
        return False, '文件上传失败！'


class AliOssCors:
    def __init__(self, access_key_id, access_key_secret, endpoint, bucket_name):
        self.access_key_id = access_key_id
        self.access_key_secret = access_key_secret
        self.endpoint = endpoint
        self.bucket_name = bucket_name
        self.auth = oss2.Auth(self.access_key_id, self.access_key_secret)

    def get_oss_cors(self):
        bucket = oss2.Bucket(self.auth, self.endpoint, self.bucket_name)
        try:
            cors = bucket.get_bucket_cors()
        except oss2.exceptions.NoSuchCors:
            print('OSS未设置跨域! 开始设置跨域----->')
            AliOssCors.set_oss_cors(self)
            AliOssCors.get_oss_cors(self)
        else:
            print('获取跨域规则----->')
            for rule in cors.rules:
                print('AllowedOrigins={0}'.format(rule.allowed_origins))
                print('AllowedMethods={0}'.format(rule.allowed_methods))
                print('AllowedHeaders={0}'.format(rule.allowed_headers))
                print('ExposeHeaders={0}'.format(rule.expose_headers))
                print('MaxAgeSeconds={0}'.format(rule.max_age_seconds))

    @staticmethod
    def set_oss_cors(self):
        bucket = oss2.Bucket(self.auth, self.endpoint, self.bucket_name)
        rule = CorsRule(allowed_origins=['*'],
                        allowed_methods=['GET', 'HEAD', 'PUT', 'POST'],
                        allowed_headers=['*'],
                        max_age_seconds=0)
        # 注意：如果已存在的规则将会被覆盖。
        bucket.put_bucket_cors(BucketCors([rule]))

    def delete_oss_cors(self):
        bucket = oss2.Bucket(self.auth, self.endpoint, self.bucket_name)
        bucket.delete_bucket_cors()


oss_obj = AliOssCors(
    access_key_id='xxxxxxxxxxxxxxxxx',
    access_key_secret='xxxxxxxxxxxxxxxxx',
    endpoint='oss-cn-hangzhou.aliyuncs.com',
    bucket_name='xxxxxxxxxxxxxxxxx')
oss_obj.get_oss_cors()


p = AliStsGenerateToKey(
    endpoint='oss-cn-hangzhou.aliyuncs.com',
    access_key_id='xxxxxxxxxxxxxxxxx',
    access_key_secret='xxxxxxxxxxxxxxxxx',
    bucket_name='xxxxxxxxxxxxxxxxx')
p.set_oss_sign(
    role_arn='acs:ram::{xxxxxxxxxxxxxxxxx}',
    save_oss_file_name='test.png',
    exp_time=300,
    upload_file='22.png'  # 本地图片
)

STS中临时授权时出现“You are not authorized to do this action. You should be authorized by RAM“报错

代码中使用的AccessKey和AccessKeySecret是主账号的，并非RAM用户的。 必须要创建子帐号才key才可以

https://help.aliyun.com/document_detail/100624.html?spm=a2c4g.11186623.2.10.5e474529lXELjN#concept-xzh-nzk-2gb

https://help.aliyun.com/document_detail/28798.html?spm=a2c4g.11186623.2.10.29bc203dUpOWmQ#reference-smb-tzy-xdb

https://help.aliyun.com/document_detail/32033.html?spm=a2c4g.11186623.2.23.64e33b49Q6fLpK#section-zx1-55k-kfc