oss_sts.py 2.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465
  1. from aliyunsdkcore import client
  2. from aliyunsdksts.request.v20150401 import AssumeRoleRequest
  3. import json
  4. import oss2
  5. # Endpoint以杭州为例,其它egion请按实际情况填写。
  6. endpoint = 'oss-cn-shenzhen.aliyuncs.com'
  7. access_key_id = 'LTAIyMkGfEdogyL9'
  8. access_key_secret = '71uIjpsqVOmF7DAITRyRuc259jHOjO'
  9. bucket_name = 'cnvod1'
  10. # role_arn是角色的资源名称。
  11. role_arn = 'acs:ram::1901342792446414:role/stsoss'
  12. clt = client.AcsClient(access_key_id, access_key_secret, 'cn-shenzhen')
  13. req = AssumeRoleRequest.AssumeRoleRequest()
  14. # 设置返回值格式为JSON。
  15. req.set_accept_format('json')
  16. req.set_RoleArn(role_arn)
  17. req.set_RoleSessionName('uid13241234123')
  18. req.set_DurationSeconds(3600)
  19. policys = {
  20. "Version": "1",
  21. "Statement": [
  22. {
  23. "Action": [
  24. "oss:PutObject",
  25. "oss:DeleteObject",
  26. ],
  27. # "Resource": ["acs:oss:*:*:cloudvod1/*"],
  28. "Resource": ["acs:oss:*:*:cloudvod1/test/*"],
  29. "Effect": "Allow",
  30. "Condition": {
  31. "IpAddress": {
  32. "acs:SourceIp": "120.237.157.184"
  33. }
  34. }
  35. }
  36. ]
  37. }
  38. req.set_Policy(Policy=json.dumps(policys))
  39. body = clt.do_action(req)
  40. # body = clt.do_action_with_exception(req)
  41. # 使用RAM账号的AccessKeyId和AccessKeySecret向STS申请临时token。
  42. token = json.loads(body)
  43. print(token)
  44. exit()
  45. # 使用临时token中的认证信息初始化StsAuth实例。
  46. auth = oss2.StsAuth(token['Credentials']['AccessKeyId'],
  47. token['Credentials']['AccessKeySecret'],
  48. token['Credentials']['SecurityToken'])
  49. print(auth)
  50. # 使用StsAuth实例初始化存储空间。
  51. bucket = oss2.Bucket(auth, endpoint, bucket_name)
  52. # 上传一个字符串。
  53. # res = bucket.put_object('oss_media_hls.ts', b'hello world')
  54. # res = bucket.put_object('test/test-name.txt', b'hello world')
  55. # print(res)
  56. # oss append obj
  57. result = bucket.append_object('mio', 0, 'content of first append')
  58. print(result)
  59. # 如果不是首次上传,可以通过bucket.head_object方法或上次追加返回值的next_position属性,得到追加位置。
  60. # bucket.append_object('<yourObjectName>', result.next_position, 'content of second append')