#!/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 17:30 @Version: python3.6 @MODIFY DECORD:ansjer dev @file: s3sts.py @Contact: chanjunkai@163.com """ def test_get_credential(): import boto3 REGION_NAME = 'us-east-1' # e.g import json boto3_sts = boto3.client( 'sts', aws_access_key_id='AKIA2E67UIMD45Y3HL53', aws_secret_access_key='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', region_name=REGION_NAME ) Policy = {"Version": "2012-10-17", "Statement": [ {"Effect": "Allow", "Action": "s3:*", "Resource": ["arn:aws:s3:::azvod1/*"]}]} response = boto3_sts.get_federation_token( Name='chanjunkai@166.com', Policy=json.dumps(Policy), DurationSeconds=7200 ) print(response) # test_get_credential() # exit() def getCredentials(): import boto3 REGION_NAME = 'us-east-1' # e.g import json sts = boto3.client( 'sts', aws_access_key_id='AKIA2E67UIMD45Y3HL53', aws_secret_access_key='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', region_name=REGION_NAME ) Policy = {"Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Action": "kinesisvideo:*", "Resource": [ "arn:aws:kinesisvideo:us-east-1:697864307463:stream/wpy_stream/1573460250304"]}]} credentials = sts.get_federation_token( Name='chanjunkai@163.com', # or any unique text related to user Policy=json.dumps(Policy), DurationSeconds=3600, ) print(credentials) access_key_id = credentials['Credentials']['AccessKeyId'] session_token = credentials['Credentials']['SessionToken'] secret_access_key = credentials['Credentials']['SecretAccessKey'] res = { 'access_key_id': access_key_id, 'session_token': session_token, 'secret_access_key': secret_access_key, } print(res) # getCredentials() def test(): print('test') import boto3 REGION_NAME = 'us-east-1' # e.g import json crd = {'Credentials': {'AccessKeyId': 'ASIA2E67UIMDZ3G2JTGG', 'SecretAccessKey': 'hZLPMgOiOUR08uZhEorG7Hf2KURo+U62n8pz4h9J', 'SessionToken': 'FwoGZXIvYXdzEEsaDFDNHGJymzOcRe21LiLFARGel4C5k3cqtnon8+yrpzZW5+tTsUONkQlBVeM6XjIgn9LUmlwroGktRBAOLuNb++licd8bUugeCU9+XMoWivJi7c3B3G5LVbofxPqs6MPfXdx4XXVa7PtMZPCyqjIhoIWcbrNuY21j3j3Mj1oggdLxPF1qPiAL1auVW8cTvygaRr6X5wgGGl0dOvBLOfGJzyCIOyLQ4BjwNCFaT4awgfrfK7rYnKvBGUWcM5UEkJ6O6dIkCHb+eHCV9RwqosXBcSE4KNk6KLKtpvMFMin3p5wPsYYvAAJVwDmFUoPyRuKom8HL8mlya7/aCnAbIM5u5lPO7TionA==', 'Expiration': 3600}, 'FederatedUser': {'FederatedUserId': '697864307463:chanjunkai@166.com', 'Arn': 'arn:aws:sts::697864307463:federated-user/chanjunkai@166.com'}, 'PackedPolicySize': 8, 'ResponseMetadata': {'RequestId': '65904087-9678-4c25-9042-fbb1b165a928', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': '65904087-9678-4c25-9042-fbb1b165a928', 'content-type': 'text/xml', 'content-length': '1138', 'date': 'Thu, 12 Mar 2020 01:56:02 GMT'}, 'RetryAttempts': 0}} # s3_resource = boto3.resource( s3_resource = boto3.client( 's3', aws_access_key_id=crd['Credentials']['AccessKeyId'], aws_secret_access_key=crd['Credentials']['SecretAccessKey'], aws_session_token=crd['Credentials']['SessionToken'], ) response = s3_resource.put_object( Body='xxxxx', Key='test1111111111111.log', Bucket='azvod1', # Expires=datetime() ) print(response) test()