ASJServer/Ansjer/test/aws_debug.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: 2019/11/13 14:38
@Version: python3.6
@MODIFY DECORD:ansjer dev
@file: aws_debug.py
@Contact: chanjunkai@163.com
"""
import boto3

REGION_NAME = 'us-east-1'  # e.g
import json


def generate_sts_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": "kinesisvideo:*", "Resource": ['*']}]}
    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,
        'secret_access_key': secret_access_key,
        'session_token': session_token,
    }
    print(res)


token_dict = {'access_key_id': 'ASIA2E67UIMDVTHGAEEV', 'secret_access_key': 'GncLAqpmcPS+4OjZt+xfg5hZrXDVYDH0jvaulunY',
              'session_token': 'FwoGZXIvYXdzEBAaDKypPx1gg3gqmDdzfCK+AZ1Z/DS1ZK8jSzrOQnAeB70PnMBK+B7rVPXyiqIilx7gqCAg/xVne9X2bHAItg3b02dGkRrD57CWqHnAeLkpsLMXYxAZsZhgEXW11NmjIAQVc5ZiNyRI8LIvp8s8FrsLLm7zs57TdbPy6i9o1di1silEiwYRQc65iKMOfVH8cOg0iZA0mfDYksuxAmeAo5lQF6JX/7DujtJR/LNzqetv83F400I47vIStBbXtXkqp/qBVMfNIrN9wVA4fmLpvBEop8Gu7gUyKT34Cd4XhwvxdXt4bq8bKm3Bya553WMEsqh0xbG7WJLWP4+8GitU8YWb'}

key = token_dict['access_key_id']
secret = token_dict['secret_access_key']
token = token_dict['session_token']

boto3_session = boto3.Session(aws_access_key_id=key, aws_secret_access_key=secret, aws_session_token=token,
                              region_name='us-east-1')

kv_client = boto3_session.client('kinesisvideo')
kv_client.create_stream(DeviceName='string',
    StreamName='string',
    MediaType='string',
    KmsKeyId='string',
    DataRetentionInHours=123,
    Tags={
        'string': 'string'
    })

res = kv_client.describe_stream(StreamName='test_stream')
print(res)