Login API Examples

4. Cloud Login

Cloud API

"""
def cloud_login_api(self, userinfo: dict):
    """
    API Name: Cloud login
    Request Parameters: userinfo 
    Request Method: post
    Request Path: cloud_login_url = Prefix.UserServer + "/users/login"
    Parameter Example: userinfo: body {"username":xxx, "password": xxx}
    """
    return self.send(self.post, self.cloud_login_url, json=userinfo).json()

def check_username_api(self, username:str):
    """
    API Name: User validation
    Request Parameters: username       
    Request Method: post
    Request Path: cloud_login_url = Prefix.UserServer + "/users/check"
    """
    payload = {'username': username}
    return self.send(self.post, self.login_check, json=payload).json()
"""

Implementation

class LoginBusiness(RquestApi):

    def cloud_login_api(self, userinfo: dict):
        """
        Cloud login API
        @param userinfo:
        @return:
        """
        return self.send(self.post, cloud_login_url, json=userinfo).json()

    def check_username_api(self, username: str):
        """
        Validate username
        @param username:
        @return:
        """
        payload = {'username': username}
        return self.send(self.post, login_check, json=payload).json()

    def getCheckToken(self) -> str:
        """
        Validate Token
        @return:
        """
        getCheckUrl = os.path.join(SimOneUrl, user_check_url)
        payload = {'username': loginData['username']}
        response = self.send(self.post, getCheckUrl, json=payload)
        checkToken = json.loads(response.content.decode("utf-8"))['data']['checkToken']
        return checkToken

    def pad(self, text):
        """
        Padding function to make the byte length of encrypted data a multiple of block_size
        @param text:
        @return:
        """
        length = AES.block_size
        count = len(text.encode('utf-8'))
        add = length - (count % length)
        entext = text + (chr(add) * add)
        return entext

    def str_aes(self, string: str, key: str):
        """
        AES encryption
        @param string:
        @param key:
        @return:
        """
        aes = AES.new(key.encode("utf-8"), AES.MODE_ECB)
        res = aes.encrypt(self.pad(string).encode("utf8"))
        msg = str(base64.b64encode(res), encoding="utf8")
        return msg

    def get_str_aes(self, string: str, username: str, key: str = "eGeVQh0lRyq41I41") -> str:
        '''
        @param string:
        @param username:
        @param key: AES encryption key
        @return:
        '''
        check_token = self.check_username_api(username)
        target_string = check_token["data"]["checkToken"] + string
        msg = self.str_aes(target_string, key)
        return msg

    @property
    def _cloud_std_login(self):
        """
        Cloud login method
        @return:
        """
        try:
            new_login_data = copy.deepcopy(loginData)
            new_login_data['password'] = self.get_str_aes(string=new_login_data['password'],
                                                          username=new_login_data['username'])
            print('cloud login data:' + str(new_login_data))
            response = self.cloud_login_api(userinfo=new_login_data)
            print("-------", response)
            print('cloud login request result:' + str(response))
            token = response['data']['token']
            os.environ[EnvPath.token] = token
            print(token)
            return token
        except Exception as e:
            print(str(e))
            raise print('get token error ')


# Login entry point
class LoginPolicy():
    standalone = "standalone"
    cloud = "cloud"

    def __new__(cls, env_name):
        if env_name == cls.standalone:
            return LoginBusiness()._enterprise_std_login
        elif env_name == cls.cloud:
            return LoginBusiness()._cloud_std_login
        else:
            return print("----------input error---------------")

Cloud Usage Example

SimOneUrl = "http://127.0.0.1:30083"
loginData = {
    "username": 'admin',
    "password": 'admin',
}

if __name__ == '__main__':
    LoginPolicy("cloud")

Successful Response

cloud login request result:{'code': 0, 'data': {'token': 'f2e3e458-264b-44fb-9f95-319a978ddec3'}}

5. Standalone Login

Standalone API

"""
def enterprise_login_api(self):
    """
    API Name: Standalone enterprise login
    Request Parameters: None 
    Request Method: get
    Request Path: SimOneUrl + standalone_login_url
    Parameter Example: userinfo: body {"username":xxx, "password": xxx}
    """
    getLoginUrl = os.path.join(SimOneUrl, standalone_login_url)
    response = self.send(self.get, url=getLoginUrl)
    return response
"""

Implementation

class LoginBusiness(RquestApi):

    def cloud_login_api(self, userinfo: dict):
        """
        Cloud login API
        @param userinfo:
        @return:
        """
        return self.send(self.post, cloud_login_url, json=userinfo).json()

    def check_username_api(self, username: str):
        """
        Validate username
        @param username:
        @return:
        """
        payload = {'username': username}
        return self.send(self.post, login_check, json=payload).json()

    def getCheckToken(self) -> str:
        """
        Validate Token
        @return:
        """
        getCheckUrl = os.path.join(SimOneUrl, user_check_url)
        payload = {'username': loginData['username']}
        response = self.send(self.post, getCheckUrl, json=payload)
        checkToken = json.loads(response.content.decode("utf-8"))['data']['checkToken']
        return checkToken

    def pad(self, text):
        """
        Padding function to make the byte length of encrypted data a multiple of block_size
        @param text:
        @return:
        """
        length = AES.block_size
        count = len(text.encode('utf-8'))
        add = length - (count % length)
        entext = text + (chr(add) * add)
        return entext

    def str_aes(self, string: str, key: str):
        """
        AES encryption
        @param string:
        @param key:
        @return:
        """
        aes = AES.new(key.encode("utf-8"), AES.MODE_ECB)
        res = aes.encrypt(self.pad(string).encode("utf8"))
        msg = str(base64.b64encode(res), encoding="utf8")
        return msg

    def get_str_aes(self, string: str, username: str, key: str = "eGeVQh0lRyq41I41") -> str:
        '''
        @param string:
        @param username:
        @param key: AES encryption key
        @return:
        '''
        check_token = self.check_username_api(username)
        target_string = check_token["data"]["checkToken"] + string
        msg = self.str_aes(target_string, key)
        return msg

    def enterprise_login_api(self):
        """
        @return:
        """
        getLoginUrl = os.path.join(SimOneUrl, standalone_login_url)
        response = self.send(self.get, url=getLoginUrl)
        return response
    @property
    def _enterprise_std_login(self):
        """
        Standalone login method
        @return:
        """
        try:
            getLoginUrl = os.path.join(SimOneUrl, standalone_login_url)
            response = self.send(self.get, url=getLoginUrl)
            token = json.loads(response.content.decode("utf-8"))['data']['token']
            os.environ[EnvPath.token] = token
        except Exception as e:
            print(str(e))
            token = ""
            # raise AssertionError('get token error ')
        return token

    @property
    def _cloud_std_login(self):
        """
        Cloud login method
        @return:
        """
        try:
            new_login_data = copy.deepcopy(loginData)
            new_login_data['password'] = self.get_str_aes(string=new_login_data['password'],
                                                          username=new_login_data['username'])
            print('cloud login data:' + str(new_login_data))
            response = self.cloud_login_api(userinfo=new_login_data)
            print("-------", response)
            print('cloud login request result:' + str(response))
            token = response['data']['token']
            os.environ[EnvPath.token] = token
            print(token)
            return token
        except Exception as e:
            print(str(e))
            raise print('get token error ')


class LoginPolicy():
    standalone = "standalone"
    cloud = "cloud"

    def __new__(cls, env_name):
        if env_name == cls.standalone:
            return LoginBusiness()._enterprise_std_login
        elif env_name == cls.cloud:
            return LoginBusiness()._cloud_std_login
        else:
            return print("----------input error---------------")

Usage Example

if __name__ == '__main__':
    LoginPolicy("standalone")

Response

{'method': 'GET', 'url': 'http://172.31.9.85:30083/api-user/users/standalone', 'headers': {'Content-Type': 'application/json', 'Authorization': None}}
response:{"code":0,"data":{"token":"b557f84a-070d-4ce3-8f70-8c0021428bed"}}