ASJServer/Ansjer/test/oss_sts.py

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


# Endpoint以杭州为例，其它egion请按实际情况填写。
endpoint = 'oss-cn-shenzhen.aliyuncs.com'
access_key_id = 'LTAIyMkGfEdogyL9'
access_key_secret = '71uIjpsqVOmF7DAITRyRuc259jHOjO'
bucket_name = 'cloudvod1'
# role_arn是角色的资源名称。
role_arn = 'acs:ram::1901342792446414:role/stsoss'

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('uid13241234123')
req.set_DurationSeconds(3600)
policys = {
    "Version": "1",
    "Statement": [
        {
            "Action": [
                "oss:PutObject",
                "oss:DeleteObject",
            ],
            # "Resource": ["acs:oss:*:*:cloudvod1/*"],
            "Resource": ["acs:oss:*:*:cloudvod1/test/*"],
            "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()
# tokens = {'RequestId': '2D83A43D-8D6B-44C5-83A4-3530BFB032EC', 'AssumedRoleUser': {'AssumedRoleId': '394329055954717182:chanjunkai', 'Arn': 'acs:ram::1901342792446414:role/stsoss/chanjunkai'}, 'Credentials': {'AccessKeySecret': 'FtPRvMDBRitAX8X7ZGoQ2DohrMWvMjP2EXVVSVqhmdyv', 'AccessKeyId': 'STS.NJHBY1uKvmvy2rRunerymDq7y', 'Expiration': '2018-11-30T07:23:54Z', 'SecurityToken': 'CAIS0wJ1q6Ft5B2yfSjIr4n9CeOFmJRX2rSSMFTjkW4wfvZBq7Scmzz2IHFIf3NhAe0bv/kzm2lX7/YYlqN4S5ZDR1HCbsJxtgf5G/c/J9ivgde8yJBZoljMewHKedGSvqL7Z+H+U6mMGJOEYEzFkSle2KbzcS7YMXWuLZyOj+wuDLEQRRLqVSdaI91UKwB+yrcdLmCDGfupPwLnpWDSAUF0wFce71ly8qOi2MaRxwPDhVnhsI8vqp/2P4KvYrsdXuh2WMzn2/dtJOiTknxb7x9X+LlxgPUDu02D3LC8G3Bd7w+LKPGultRkN11CYbUgEKNltIqF95oh4LSVzd+ojU8QYqcMA3j9PNnxmJKfRLn2Z4xlKeinYij3v4rRZsWvgWQNemkGMQ5GQd0lJ0JrBAYkIjOgcf77owqbOlr+GvXZj/hpiMov1Tvz4cGNPEOf5T4EVOPY4vQagAGFFahWtPsgkWPPJhSOu/bsFZGJFeQvb/Ij5nT5dPNiuEfvECu+ns2GmUB5sAQO9xeMmIBr9y2yteTIwoG8FQx5S4ybj051MK+K53GeRZOSd05iZRU8hpU3+O2mTlQI4FicDjG6C6my1IJpNPYjO+8Y9cWecm+aodAeBJLJS1WP6w=='}}

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