ASJServer/Ansjer/test/s3sts.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 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()