python

一.python(2.7)

#!/bin/python
#-*-coding=utf-8-*-

import requests
import json
import hashlib
import base64
from datetime import datetime
from Crypto.Cipher import AES
from Crypto import Random

SRV_URL = "https://api.threathunter.cn/zhbpro/"
POST_URL = SRV_URL + "phone_no_check/v2"
SNUSER = "******"
SNKEY = "******"

def AesEncryptSeg(phoneno):
    remainder = len(phoneno) % 16
    if remainder:
        padded_value = phoneno + '\0' * (16 - remainder)
    else:
        padded_value = phoneno
    # a random 16 byte key
    iv = Random.new().read(AES.block_size)
    # CFB mode
    cipher = AES.new(SNKEY, AES.MODE_CFB, iv, segment_size=128)
    # drop the padded value(phone number length is short the 16bytes)
    value = cipher.encrypt(padded_value)[:len(phoneno)]
    ciphertext = iv + value
    return base64.encodestring(ciphertext).strip()

def AesDecryptSeg(phoneno):
    data = base64.decodestring(phoneno)
    cihpertxt = data[AES.block_size:]
    remainder = len(cihpertxt) % 16
    if remainder:
        padded_value = cihpertxt + '\0' * (16 - remainder)
    else:
        padded_value = cihpertxt
    cryptor = AES.new(SNKEY, AES.MODE_CFB, data[0:AES.block_size], segment_size=128)
    plain_text  = cryptor.decrypt(padded_value)
    return plain_text[0:len(cihpertxt)]


def TestGetStatus(userlist):
    users = []
    for user in userlist:
        u = {}
        usha1 = hashlib.sha1(user).hexdigest()
        u["user"] = usha1
        users.append(u)

    # user list origin text
    pstr = json.dumps(users)
    print "====>request plain<===="
    print pstr
    # encrypt the origin text
    cstr = AesEncryptSeg(pstr)

    payload = {
            "snuser":SNUSER,
            "data":cstr
    }
    print "====>request body<===="
    print payload
    r = requests.post(POST_URL, data=json.dumps(payload), verify=False)
    rjson = json.loads(r.text)
    print "====>response body<===="
    print rjson
    if rjson["status"] == 200:
        print "====>response plain<===="
        print AesDecryptSeg(rjson["data"])
    else:
        print "if status is 503, please check you user, snkey, and if ip in white list"
        print "如果返回是503，请确认你的用户，密钥是否正确，以及ip是否在白名单内"


if __name__ == "__main__":
    TestGetStatus(['13978049038','19831258560','13408050087','14000085517','14000085515','15906198201','13099630356'])

二·python(3.7)

# -*- encoding: utf-8 -*-
"""
# install requirement
pip install requests
pip install pycryptodome

"""
import requests
import json
import hashlib
import base64

from Crypto.Cipher import AES
from Crypto import Random

SRV_URL = "https://api.yazx.com/zhbpro/"
POST_URL = SRV_URL + "phone_no_check/v2"
SNUSER = "******"
SNKEY = "******"


def encrypt(encrypt_str: str, cecret: str):
    """
    aes加密数据后，再进行base54编码后返回
    :param encrypt_str:
    :param cecret:
    :return:
    """
    remainder = len(encrypt_str) % AES.block_size
    if remainder:
        padded_value = encrypt_str + '\0' * (AES.block_size - remainder)
    else:
        padded_value = encrypt_str
    # a random 16 byte key
    iv = Random.new().read(AES.block_size)
    # CFB mode
    cipher = AES.new(bytes(cecret, encoding="utf-8"), AES.MODE_CFB, iv, segment_size=128)
    # drop the padded value(phone number length is short the 16bytes)
    value = cipher.encrypt(bytes(padded_value, encoding="utf8")[:len(encrypt_str)])
    ciphertext = iv + value
    return str(base64.encodebytes(ciphertext).strip(), encoding="utf8")


def decrypt(encrypt_str: str, cecret: str):
    """
    base64解码后，再进行aes解密
    :param encrypt_str:
    :param cecret:
    :return:
    """
    data = base64.decodebytes(bytes(encrypt_str, encoding="utf8"))
    cihpertxt = data[AES.block_size:]
    remainder = len(cihpertxt) % AES.block_size
    if remainder:
        padded_value = cihpertxt + b'\0' * (AES.block_size - remainder)
    else:
        padded_value = cihpertxt
    cryptor = AES.new(bytes(cecret, encoding="utf-8"), AES.MODE_CFB, data[0:AES.block_size], segment_size=128)
    plain_text = cryptor.decrypt(padded_value)
    return str(plain_text[0:len(cihpertxt)], encoding="utf8")


def TestGetStatus(userlist):
    users = []
    for user in userlist:
        u = {}
        usha1 = hashlib.sha1(user.encode()).hexdigest()
        u["user"] = usha1
        users.append(u)

    # user list origin text
    pstr = json.dumps(users)
    print("====>request plain<====")
    print(pstr)
    # encrypt the origin text
    cstr = encrypt(pstr, SNKEY)

    payload = {
            "snuser":SNUSER,
            "data":cstr
    }
    print("====>request body<====")
    print(payload)
    r = requests.post(POST_URL, data=json.dumps(payload), verify=True)
    rjson = json.loads(r.text)
    print("====>response body<====")
    print(rjson)
    if rjson["status"] == 200:
        print("====>response plain<====")
        print(decrypt(rjson["data"], SNKEY))
    else:
        print("if status is 503, please check you user, snkey, and if ip in white list")
        print("如果返回是503，请确认你的用户，密钥是否正确，以及ip是否在白名单内")


if __name__ == "__main__":
    TestGetStatus(["15118376562","13520438342", "13900100033"])