Python Example
- Python Version:
3.8 - Requirements: websockets
python
import asyncio
import hashlib
import json
import random
import threading
import time
import requests
import websockets
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"
}
}
}
# ---------- Change the information here ----------
appId = "Your appId"
appSecret = "Your appSecret"
# ---------------------------------------------------
baseHost = "decrypt.nikola-lab.cn"
openApiBaseUrl = f"https://{baseHost}/openapi/v1"
def create_sign(time_stamp, nt, name):
"""
Create a signature based on timestamp, random number, and appSecret, using the algorithm of the passed in name
:param name: The hash method.
:param time_stamp: Current unix timestamp
:param nt: Random number
:return: The result for the signed.
"""
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}/'
# Open API URL and Test Parameter Mapping Table
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"The signature algorithms supported by the current operating environment: {hashlib.algorithms_available}")
def create_openapi_key_args():
"""
Key verification parameters required for creating OpenApi
:return:
"""
time_stamp_ms = round(time.time() * 1000.0)
nonce = random.randint(0, 0xFFFFFFFF)
# time_stamp_ms = 1700818596082 # Simulate old timestamps to achieve the effect of testing the first stage of replay prevention
# nonce = 0xFFFFFFFF # Fix a random number to achieve the effect of testing the second stage of anti replay
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():
"""
Create a http headers object.
:return:
"""
open_api_args = create_openapi_key_args()
return {
'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"):
"""
Request url to download body stream.
:param url: The URL to be requested
:param method: http method
:return: body bytes
"""
openapi_header = make_request_header()
resp = requests.request(method, url, headers=openapi_header)
if resp.status_code == 200:
return resp.content
raise Exception(f"{url} Dwonload failed, status_code = {resp.status_code}")
def request_with_data(url: str, data: str | None, method: str, file_upload=False):
"""
Carrying data request specified URL
:param url: The URL to be requested
:param data: data
:param method: http method
:param file_upload: Is it uploading a file
:return: JSON object for response
"""
openapi_header = make_request_header()
# If it is determined that a file needs to be uploaded, we do not need to set the content as a JSON attribute
if file_upload:
openapi_header['Content-Type'] = 'application/octet-stream' # TODO Note that the pure upload interface only accepts one binary file upload
else:
openapi_header['Content-Type'] = 'application/json'
# Call requests.request to start the 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"Request {json_obj['path']} failed, "
f"Status: {json_obj['error']}({json_obj['status']}), "
f"Message: {json_obj['message']}")
return json_obj
def request_create_task():
"""
Request to create a task and return the UUID of the task after successful creation
:return: The UUID string of the task
"""
open_api_item = OPEM_API_MAP['nested_raw'] # TODO Select the OPENAPI you need to test decryption from the mapping table
# Automatically determine whether data or file upload is required based on the parameters required by the 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
# Start request
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):
"""
Request to query task information
:param uuid: The uuid of the task
:return: Task status information
"""
url = make_decrypt_task_url(f"{uuid}/information")
print(f"The informations of task: {url}")
result_obj = request_with_data(url, None, 'GET')
print(result_obj)
return result_obj
async def client():
"""
Connect to the decryption result notification service to listen.
If decryption is completed, this websocket will call back once to notify a UUID that the task has been completed
:return:
"""
open_api_args = create_openapi_key_args()
url_for_result_notify = (f"wss://{baseHost}/openapi/v1/decrypt/result/notify?"
f"appId={appId}×tamp={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"The server notification task completed, The UUID of task is {json_obj['uuid']}, result: {json_obj['keysOrLink']}")
def download_keys_from_task(uuid, task_info):
"""
Download keys from task result
:param uuid: the uuid of task
:param task_info: info of the task
: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 # Empty, maybe decrypt task is failed.
try:
# If it can be formatted as JSON, it means that the decryption result is the result of a large number of keys, and we need to download the result
link_info = task_info['keysOrLink']
print(f"Keys count = {link_info['size']}，Keys link = {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"Dwonload success, save to {file_name}, the size of file is{len(key_bytes)}bytes")
except Exception as e:
# print(e)
# If it cannot be formatted as JSON, it means it is a normal decryption with a small amount of results, and we can directly display it
print(f"Decrypt result = {task_info['keysOrLink']}")
def test():
"""
TEST
:return:
"""
test_count = 1
task_list = []
# Create task
for i in range(test_count):
response = request_create_task()
task_list.append({
'uuid': response['uuid'],
'done': False
})
# TODO You can temporarily skip queries
skip_result_query = False
if skip_result_query:
return
# Query in loop
while True:
is_all_done = True # Flag whether all decryption tasks have been completed currently
# Query one by one in the loop
for i in range(test_count):
if task_list[i]['done']:
continue
is_all_done = False # Found an unfinished task, we need to continue querying
# Request task informations, to get result of task.
task_info = request_query_task(task_list[i]['uuid'])
if task_info is not None:
if task_info['finish']:
task_list[i]['done'] = True
time.sleep(2) # TODO Don't request too quickly, it may be blocked by current limiting risk control
# All tasks have been completed, we can now end the query
if is_all_done:
break
print("")
# Optional testing of multi-threaded requests
# for thread_i in range(50):
# threading.Thread(target=test, ).start()
# Start ws client processing
# threading.Thread(target=lambda: asyncio.run(client()), ).start()
# Single threaded testing
test()