aliyun_cf.py

# encoding: utf-8

import logging
import hashlib
import json

from aliyunsdkcore.client import AcsClient
from aliyunsdkcf.request.v20151127 import AuthenticateRequest

from rumble import settings

logger = logging.getLogger(__name__)

acs_client = None


def init_client():
    global acs_client
    if acs_client is None:
        acs_client = \
            AcsClient(settings.ALIYUN_ACCESSKEY, settings.ALIYUN_SECRET,
                      region_id='cn-hangzhou')

init_client()


def authenticate_request(token, sig, sessionid, remoteip):
    """ authenticate request

    :param token: request token
    :param sig: request signature
    :param sessionid: request sessionid
    :param remoteip: client ip
    :return: bool
    """
    appkey = settings.ALICF_APPKEY
    accesskey = settings.ALICF_ACCESSKEY
    enc = '{}{}{}{}'.format(
        appkey, accesskey, token, sig
    )
    serno = hashlib.md5(enc).hexdigest()
    request = AuthenticateRequest.AuthenticateRequest()
    request.set_accept_format('JSON')
    request.set_AppKey(appkey)
    request.set_Token(token)
    request.set_Sig(sig)
    request.set_Serno(serno)
    request.set_SessionId(sessionid)
    request.set_RemoteIp(remoteip)

    try:
        response = json.loads(acs_client.do_action(request))
    except Exception as e:
        logger.exception(e, exc_info=True)
        return False
    else:
        return _handle_authenticate_response(response)


def _handle_authenticate_response(response):
    ret = False
    if response.get('Success', False):
        auth_code = \
            int(response.get('SigAuthenticateResult', {}).get('code', 0))
        if 100 <= auth_code < 300:
            ret = True

    if not ret:
        logger.warning(
            'aliyun authenticate request error, get response : {}'.format(
                response
            )
        )
    return ret