Spaces:
Paused
Paused
File size: 3,737 Bytes
53e18b4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 |
"""
DeepSeek Proof of Work Challenge Implementation
Author: @xtekky
Date: 2024
This module implements a proof-of-work challenge solver using WebAssembly (WASM)
for Custom sha3 hashing. It provides functionality to solve computational challenges
required for authentication or rate limiting purposes.
"""
import json
import base64
import wasmtime
import numpy as np
from typing import Dict, Any
import os
WASM_PATH = f'{os.path.dirname(__file__)}/wasm/sha3_wasm_bg.7b9ca65ddd.wasm'
class DeepSeekHash:
def __init__(self):
self.instance = None
self.memory = None
self.store = None
def init(self, wasm_path: str):
engine = wasmtime.Engine()
with open(wasm_path, 'rb') as f:
wasm_bytes = f.read()
module = wasmtime.Module(engine, wasm_bytes)
self.store = wasmtime.Store(engine)
linker = wasmtime.Linker(engine)
linker.define_wasi()
self.instance = linker.instantiate(self.store, module)
self.memory = self.instance.exports(self.store)["memory"]
return self
def _write_to_memory(self, text: str) -> tuple[int, int]:
encoded = text.encode('utf-8')
length = len(encoded)
ptr = self.instance.exports(self.store)["__wbindgen_export_0"](self.store, length, 1)
memory_view = self.memory.data_ptr(self.store)
for i, byte in enumerate(encoded):
memory_view[ptr + i] = byte
return ptr, length
def calculate_hash(self, algorithm: str, challenge: str, salt: str,
difficulty: int, expire_at: int) -> float:
prefix = f"{salt}_{expire_at}_"
retptr = self.instance.exports(self.store)["__wbindgen_add_to_stack_pointer"](self.store, -16)
try:
challenge_ptr, challenge_len = self._write_to_memory(challenge)
prefix_ptr, prefix_len = self._write_to_memory(prefix)
self.instance.exports(self.store)["wasm_solve"](
self.store,
retptr,
challenge_ptr,
challenge_len,
prefix_ptr,
prefix_len,
float(difficulty)
)
memory_view = self.memory.data_ptr(self.store)
status = int.from_bytes(bytes(memory_view[retptr:retptr + 4]), byteorder='little', signed=True)
if status == 0:
return None
value_bytes = bytes(memory_view[retptr + 8:retptr + 16])
value = np.frombuffer(value_bytes, dtype=np.float64)[0]
return int(value)
finally:
self.instance.exports(self.store)["__wbindgen_add_to_stack_pointer"](self.store, 16)
class DeepSeekPOW:
def __init__(self):
self.hasher = DeepSeekHash().init(WASM_PATH)
def solve_challenge(self, config: Dict[str, Any]) -> str:
"""Solves a proof-of-work challenge and returns the encoded response"""
answer = self.hasher.calculate_hash(
config['algorithm'],
config['challenge'],
config['salt'],
config['difficulty'],
config['expire_at']
)
result = {
'algorithm': config['algorithm'],
'challenge': config['challenge'],
'salt': config['salt'],
'answer': answer,
'signature': config['signature'],
'target_path': config['target_path']
}
return base64.b64encode(json.dumps(result).encode()).decode() |