ASJServer/Controller/S3GetStsController.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import hashlib
import json
import time
import uuid

import boto3
from django.http import HttpResponse, JsonResponse
from django.views import View

from Ansjer.config import AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_ARN
from Controller.DeviceConfirmRegion import Device_Region
from Model.models import Device_User, Device_Info, iotdeviceInfoModel, UIDCompanySerialModel, \
    SerialNumberModel
from Object.IOTCore.IotObject import IOTClient
from Object.ResponseObject import ResponseObject
from Service.CommonService import CommonService


class S3GetStsView(View):

    def get(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        request_dict = request.GET
        operation = kwargs.get('operation', None)
        return self.validate(operation, request_dict, request)

    def post(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        request_dict = request.POST
        operation = kwargs.get('operation', None)
        return self.validate(operation, request_dict, request)

    def validate(self, operation, request_dict, request):

        response = ResponseObject()

        if operation == 'ota':
            return self.ota(request_dict, response, request)
        else:
            return response.json(404)


    # 授权ota升级s3预签名
    def ota(self, request_dict, response, request):
        mold = request_dict.get('mold', None)  #0国内， 1国外
        region_name = request_dict.get('region_name', None)
        bucket_name = request_dict.get('bucket_name', None)
        endpoint = request_dict.get('endpoint', None)
        filepath = request_dict.get('filepath', None)
        role_name = request_dict.get('jobname', None)

        try:
            aws_access_key_id = AWS_ACCESS_KEY_ID[mold]
            aws_secret_access_key = AWS_SECRET_ACCESS_KEY[mold]
            aws_arn = AWS_ARN[mold]
        except:
            res = {'code': 404, 'msg': 'mold not exists!'}
            return HttpResponse(json.dumps(res, ensure_ascii=False), content_type="application/json,charset=utf-8")

        ###############
        boto3_sts = boto3.client(
            'sts',
            aws_access_key_id=aws_access_key_id,
            aws_secret_access_key=aws_secret_access_key,
            region_name=region_name
        )
        Policy = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": "s3:*",
                    "Resource": ["{aws_arn}:::{bucket_name}/{filepath}*".
                                     format(aws_arn=aws_arn, bucket_name=bucket_name, filepath=filepath)]
                }
            ]
        }
        response = boto3_sts.get_federation_token(
            Name='{role_name}'.format(role_name=role_name),
            Policy=json.dumps(Policy),
            DurationSeconds=7200
        )
        ##############
        res = {
            'AccessKeyId': response['Credentials']['AccessKeyId'],
            'AccessKeySecret': response['Credentials']['SecretAccessKey'],
            'SessionToken': response['Credentials']['SessionToken'],
            'Expiration': response['Credentials']['Expiration'],
            'expire': 900,
            'endpoint': endpoint,
            'bucket_name': bucket_name,
            'arn': response['FederatedUser']['Arn'],
            'region': region_name,
            'bucket_mold': mold
        }
        return JsonResponse(status=200, data=res)