Nikola-Lab RFID 解密系统

简体中文

English

简体中文

English

Appearance

Python例程 ​

python
import asyncio
import hashlib
import json
import random
import threading
import time

import requests
import websockets

# 测试请求新的解密服务API


data_json_nested_v1 = {
    "uid": "814910980",
    "dist": "637",
    "keyA": 1,
    "block": "07",
    "cores": [
        {
            "nt1": "3032128150",
            "nt2": "432058316",
            "par": "5"
        },
        {
            "nt1": "808897476",
            "nt2": "904284173",
            "par": "3"
        }
    ]
}

data_json_nested_v2 = {
    "uid": "814910980",
    "keyA": 0,
    "block": "51",
    "cores": [
        {
            "ntp": "1721933979",
            "ks1": "2137132887"
        },
        {
            "ntp": "2755935295",
            "ks1": "2443342898"
        },
        {
            "ntp": "1254421459",
            "ks1": "2132982750"
        },
        {
            "ntp": "4059870459",
            "ks1": "3290103030"
        }
    ]
}

data_json_static_nested = {
    "uid": "961684779",
    "keyA": 0,
    "block": "51",
    "cores": [
        {
            "nt1": "18874693",
            "nt2": "2603586571"
        },
        {
            "nt1": "18874693",
            "nt2": "3619625478"
        }
    ]
}

data_json_darkside = {
    "uid": "2586461451",
    "keyA": 1,
    "block": "07",
    "cores": [
        {
            "ntp": "3617531615",
            "par": "14978751391669411775",
            "ks1": "1128099048457733",
            "nr": "0",
            "ar": "0"
        },
        {
            "ntp": "3617531615",
            "par": "6917397047991074623",
            "ks1": "867234333445718795",
            "nr": "1",
            "ar": "0"
        },
        {
            "ntp": "3617531615",
            "par": "549386136641275711",
            "ks1": "362823457123927557",
            "nr": "2",
            "ar": "0"
        }
    ]
}

data_json_mfkey32 = {
    "uid": "305419896",
    "keyA": 1,
    "block": "0",
    "cores": {
        "core1": {
            "nt": "450420523",
            "nr": "489775140",
            "ar": "1645146184"
        },
        "core2": {
            "nt": "819383047",
            "nr": "3307239394",
            "ar": "2205861402"
        }
    }
}

data_json_mfkey64 = {
    "uid": "341286761",
    "keyA": 1,
    "block": "0",
    "cores": {
        "nt": "3464774241",
        "nr": "4161051851",
        "ar": "86362191",
        "at": "2486291520"
    }
}

data_json_nested_rf08s = {
    "uid": "3779066835",
    "keyA": True,
    "block": "03",
    "cores": {
        "typeA": {
            "nt1": "4041266571",
            "nt2": "3748496197",
            "par": "0"
        },
        "typeB": {
            "nt1": "332352875",
            "nt2": "3968861009",
            "par": "15"
        }
    }
}


# ---------- 请修改以下信息 ----------

appId = "你的appId"
appSecret = "你的appSecret"

# ----------------------------------

baseHost = "decrypt.nikola-lab.cn"
openApiBaseUrl = f"https://{baseHost}/openapi/v1"


def create_sign(time_stamp, nt, name):
    """
        根据时间戳和随机数和appSecret创建签名，使用传入的名称的算法
    :param name: 签名算法名称
    :param time_stamp: 时间戳
    :param nt: 随机数
    :return: 签名结果
    """
    hash_obj = hashlib.new(name)
    hash_obj.update(f"{nt}:{time_stamp}:{appSecret}".encode(encoding="ascii"))
    return hash_obj.hexdigest()


def make_decrypt_task_url(url: str):
    return f'{openApiBaseUrl}/decrypt/task/{url}/'


# 开放API的URL和调试参数的映射表
OPEM_API_MAP = {
    'nested_raw': {
        'url': make_decrypt_task_url("create/nested/raw_response"),
        'data': data_json_nested_v1
    },
    'nested_ntpks1': {
        'url': make_decrypt_task_url("create/nested/ntp_ks1"),
        'data': data_json_nested_v2
    },
    'static_nested': {
        'url': make_decrypt_task_url("create/nested/static_fast"),
        'data': data_json_static_nested
    },
    'darkside': {
        'url': make_decrypt_task_url("create/darkside"),
        'data': data_json_darkside
    },
    'mfkey32': {
        'url': make_decrypt_task_url("create/mfkey32"),
        'data': data_json_mfkey32
    },
    'mfkey64': {
        'url': make_decrypt_task_url("create/mfkey64"),
        'data': data_json_mfkey64
    },
    'hardnested': {
        'url': make_decrypt_task_url("create/nested/hard_nonce"),
        'file': "./nonces.bin",
    },
    "nested_rf08s": {
        'url': make_decrypt_task_url("create/nested/rf08s_2x1nt"),
        'data': data_json_nested_rf08s
    }
}

print(f"当前运行环境支持的签名算法：{hashlib.algorithms_available}")


def create_openapi_key_args():
    """
        创建OpenApi需要的关键的验签参数
    :return:
    """
    time_stamp_ms = round(time.time() * 1000.0)
    nonce = random.randint(0, 0xFFFFFFFF)
    # time_stamp_ms = 1700818596082 # 模拟旧的时间戳，达到测试防重放第一阶段的效果
    # nonce = 0xFFFFFFFF  # 固定一个随机数，达到测试防重放第二阶段的效果
    sign_method = "SHA1"
    return {
        'timestamp': str(time_stamp_ms),
        'nonce': str(nonce),
        'method': sign_method,
        'sign': create_sign(time_stamp_ms, nonce, sign_method)
    }


def make_request_header():
    """
        构造一个请求头
    :return:
    """
    open_api_args = create_openapi_key_args()
    return {
        # OpenApi必要参数
        'X-CA-App-Id': appId,
        'X-CA-Timestamp': open_api_args['timestamp'],
        'X-CA-Nonce': open_api_args['nonce'],
        'X-CA-Signature': open_api_args['sign'],
        'X-CA-Signature-Method': open_api_args['method'],
    }


def request_and_download_body(url, method="GET"):
    """
        请求目标url，然后下载body数据流
    :param url: 目标url
    :param method: 请求方法
    :return: 目标数据流
    """
    openapi_header = make_request_header()
    # 调用 requests.request 开始请求
    resp = requests.request(method, url, headers=openapi_header)
    if resp.status_code == 200:
        return resp.content
    raise Exception(f"{url} 下载失败, status_code = {resp.status_code}")


def request_with_data(url: str, data: str | None, method: str, file_upload=False):
    """
        携带数据请求指定的url
    :param url: 要请求的url
    :param data: 数据
    :param method: 请求的方法
    :param file_upload: 是否是上传文件
    :return: 应答的json对象
    """
    openapi_header = make_request_header()
    # 如果判断到需要上传文件，我们就不需要设置内容为json的属性
    if file_upload:
        openapi_header['Content-Type'] = 'application/octet-stream'  # TODO 注意，纯上传接口，只接受一个二进制文件上传
    else:
        openapi_header['Content-Type'] = 'application/json'
    # 调用 requests.request 开始请求
    resp = requests.request(method, url, data=data, headers=openapi_header)
    # print(resp.content.decode())
    json_obj = json.loads(resp.content.decode())
    if resp.status_code >= 400:
        raise RuntimeError(f"调用接口 {json_obj['path']} 失败，"
                           f"状态码：{json_obj['error']}（{json_obj['status']}），"
                           f"消息：{json_obj['message']}")
    return json_obj


def request_create_task():
    """
        请求创建任务，并且返回创建成功之后的任务的UUID
    :return: 任务的UUID字符串
    """
    open_api_item = OPEM_API_MAP['nested_raw']  # 从映射表里面选择你需要测试解密的OPENAPI

    # 根据API需要的参数自动进行判断是否需要数据或者上传文件
    data = None
    is_file_upload = False
    if 'data' in open_api_item:
        data = json.dumps(open_api_item['data'])
    if 'file' in open_api_item:
        data = open(open_api_item['file'], 'rb')
        is_file_upload = True

    # 开始请求
    content_obj = request_with_data(open_api_item['url'], data, 'POST', is_file_upload)
    print(content_obj)
    return content_obj


def request_query_task(uuid: str):
    """
        请求查询任务信息
    :param uuid: 任务的uuid
    :return: 任务的状态信息
    """
    url = make_decrypt_task_url(f"{uuid}/information")
    print(f"组合之后的查询信息的URL是：{url}")
    result_obj = request_with_data(url, None, 'GET')
    print(result_obj)
    return result_obj


async def client():
    """
        连接到解密结果通知服务进行监听，如果解密完成此websocket将回调一次通知某个UUID的任务完成
    :return:
    """
    open_api_args = create_openapi_key_args()
    url_for_result_notify = (f"wss://{baseHost}/openapi/v1/decrypt/result/notify?"
                             f"appId={appId}&timestamp={open_api_args['timestamp']}&nonce={open_api_args['nonce']}"
                             f"&signature={open_api_args['sign']}&signature_method={open_api_args['method']}")
    async with websockets.connect(url_for_result_notify) as websocket:
        # Process messages received on the connection.
        async for message in websocket:
            print(message)
            json_obj = json.loads(message)
            print(f"服务端通知UUID为 {json_obj['uuid']} 的任务已完成：{json_obj['keysOrLink']}")


def download_keys_from_task(uuid, task_info):
    """
        从任务信息中下载解密成功的密钥
    :param uuid: 任务的uuid
    :param task_info: 解密任务信息
    :return:
    """
    url = make_decrypt_task_url(f"{uuid}/keys")
    # {'keysOrLink': 'daf7fabfff41,ffffffffff41,ba6a550955e1', 'finish': True}
    # {'keysOrLink': {'size': 16411, 'link': 'b82a906b4bd708b7f56c65c85930d6818fd8d0c0'}, 'finish': True}
    if task_info['keysOrLink'] is None:
        return  # 空的密钥，没解密成功或者解密成功了但是真的解不出密钥
    try:
        # 如果能格式化成json，说明解密结果是一个大量密钥的结果，我们需要进行结果的下载
        link_info = task_info['keysOrLink']
        print(f"发现多密钥解密结果，密钥数量为 {link_info['size']}，密钥文件关联信息为：{link_info['link']}")
        key_bytes = request_and_download_body(url)
        file_name = link_info['link'] + ".bin"
        with open(file_name, 'wb+') as f:
            f.write(key_bytes)
        print(f"下载密钥文件成功，保存到为 {file_name}，大小为{len(key_bytes)}bytes")
    except Exception as e:
        # print(e)
        # 而如果无法格式化成json，则说明是普通的解密少量结果，我们可以直接展示
        print(f"解密结果为：{task_info['keysOrLink']}")


def test():
    """
        测试单元
    :return:
    """
    test_count = 1
    task_list = []

    # 创建任务阶段
    for i in range(test_count):
        response = request_create_task()
        task_list.append({
            'uuid': response['uuid'],
            'done': False
        })

    # TODO 暂时屏蔽查询
    skip_result_query = False
    if skip_result_query:
        return

    # 查询结果阶段
    while True:
        is_all_done = True  # 标志当前是否所有的解密任务都完成了
        # 在循环里一个一个查询
        for i in range(test_count):
            if task_list[i]['done']:
                continue
            is_all_done = False  # 发现一个未完成的，我们需要继续查询
            # 在此处实现查询任务状态
            uuid = task_list[i]['uuid']
            task_info = request_query_task(uuid)
            if task_info is not None:
                if task_info['finish']:
                    task_list[i]['done'] = True
                    download_keys_from_task(uuid, task_info)

        time.sleep(2)  # 不要请求太快了，可能会被限流风控拉黑

        # 所有的任务都完成了，我们可以结束查询了
        if is_all_done:
            break
        print("")


# 可测试多线程请求
# for thread_i in range(50):
#     threading.Thread(target=test, ).start()

# 启动ws客户端处理
# threading.Thread(target=lambda: asyncio.run(client()), ).start()

# 单线程测试
test()

本系统由Nikola-LAB强力驱动

Copyright © 2022 NIKOLAT.LAB 版权所有 粤ICP备2022019306号