ASJServer/Ansjer/test/oss_sts.py

#!/usr/bin/env python3  
# -*- coding: utf-8 -*-  
"""
@Copyright (C) ansjer cop Video Technology Co.,Ltd.All rights reserved.
@AUTHOR: ASJRD018
@NAME: AnsjerFormal
@software: PyCharm
@DATE: 2018/11/8 9:23
@Version: python3.6
@MODIFY DECORD:ansjer dev
@file: oss_sts.py
@Contact: chanjunkai@163.com
"""
# -*- coding: utf-8 -*-
'''
生成sts上传授权
'''
from aliyunsdkcore import client
from aliyunsdksts.request.v20150401 import AssumeRoleRequest
import json
import oss2



'''
host = ""  # just for example
accessid = ""
accesskey = ""
bucket_name = ""
channel_name = "test-channel"
auth = Auth(accessid, accesskey)
bucket = Bucket(auth, host, bucket_name)
'''
# Endpoint以杭州为例，其它Region请按实际情况填写。
endpoint = 'oss-cn-shenzhen.aliyuncs.com'
# 阿里云主账号AccessKey拥有所有API的访问权限，风险很高。强烈建议您创建并使用RAM账号进行API访问或日常运维，请登录 https://ram.console.aliyun.com 创建RAM账号。
# access_key_id = 'LTAIXSozepSx81xe'
# access_key_secret = '19BlxRiAHDPLx4mhz6vQZRuhK0eKed'
access_key_id = 'LTAIKF7ohvwF5ugH'
access_key_secret = 'AfBIeMdrvMN3m0EsQYuRkmNyA7uWoz'
bucket_name = 'testrtmp'
# role_arn是角色的资源名称。
role_arn = 'acs:ram::1024237323330331:role/maxadmin'

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

# 设置返回值格式为JSON。
req.set_accept_format('json')
req.set_RoleArn(role_arn)
req.set_RoleSessionName('chanjunkai')
req.set_DurationSeconds(3600)
policys={
    "Version": "1",
    "Statement": [
        {
            "Action": [
                "oss:PutObject",
            ],
            "Resource": ["acs:oss:*:*:testrtmp/*"],
            "Effect": "Allow",
            "Condition": {
                "IpAddress": {
                    "acs:SourceIp": "120.237.157.184"
                }
            }
        }
    ]
}
req.set_Policy(Policy=json.dumps(policys))
body = clt.do_action(req)
# body = clt.do_action_with_exception(req)
# 使用RAM账号的AccessKeyId和AccessKeySecret向STS申请临时token。
token = json.loads(body)
print(token)
exit()

# 使用临时token中的认证信息初始化StsAuth实例。
auth = oss2.StsAuth(token['Credentials']['AccessKeyId'],
                    token['Credentials']['AccessKeySecret'],
                    token['Credentials']['SecurityToken'])
print(auth)
# 使用StsAuth实例初始化存储空间。
bucket = oss2.Bucket(auth, endpoint, bucket_name)
# 上传一个字符串。
bucket.put_object('object-name.txt', b'hello world')