from fastapi import FastAPI, HTTPException from pydantic import BaseModel import hashlib from solders.keypair import Keypair from solders.pubkey import Pubkey from solana.rpc.api import Client from solders.transaction import Transaction from solders.system_program import TransferParams, transfer from solana.rpc.types import TxOpts import base58 import hashlib import os #from helpers.mongodbconnection import provideClient from fastapi.concurrency import run_in_threadpool import pybase64 from web3 import Web3 import hexbytes app = FastAPI() solana_client = Client("https://devnet.helius-rpc.com/?api-key=4e833ada-d32c-48c5-b020-c11b2253f25b") ETH_RPC_URL = "https://mainnet.infura.io/v3/7ad20d8703134068a4e4563f4a5a2279" w3 = Web3(Web3.HTTPProvider(ETH_RPC_URL)) seed_text="shellinfo" seed = hashlib.sha256(seed_text.encode()).digest() SERVER_KEY = Keypair.from_seed(seed) class TransactionPayload(BaseModel): signed_hex: str class EthTransactionPayload(BaseModel): signed_hex: str message: str # The original message signature: str # Base64 encoded signature public_key: str # Base64 encoded public key def verify_dc(message,signature,public_key): msg_bytes = message.encode('utf-8') sig_bytes = pybase64.b64decode(signature) pk_bytes = pybase64.b64decode(public_key) with oqs.Signature("Dilithium5") as verifier: # 3. Perform Verification is_valid = verifier.verify(msg_bytes, sig_bytes, pk_bytes) if is_valid: return True else: return False ''' CHAIN_LENGTH = 256 N = 32 # SHA256 output bytes def verify_wots(signature, message, public_key): """ signature: list[str] (base58 encoded, length 32) message: bytes public_key: list[str] (base58 encoded, length 32) """ # Basic sanity checks if len(signature) != N or len(public_key) != N: print("Invalid lengths") return False # Hash the message msg_hash = hashlib.sha256(message).digest() for i in range(N): try: sig_i = base58.b58decode(signature[i]) pk_i = base58.b58decode(public_key[i]) except Exception as e: print(f"Base58 decode error at index {i}: {e}") return False check = sig_i # Walk forward in hash chain steps = CHAIN_LENGTH - msg_hash[i] for _ in range(steps): check = hashlib.sha256(check).digest() # Compare with public key element if check != pk_i: print(f"Mismatch at index {i}") return False return True ''' CHAIN_LENGTH = 256 N_MESSAGE = 32 # Standard hash length N_TOTAL = 34 # 32 (Message) + 2 (Checksum) def verify_wots(signature, message, public_key): """ signature: list[str] (base58 encoded, length 34) public_key: list[str] (base58 encoded, length 34) """ if len(signature) != N_TOTAL or len(public_key) != N_TOTAL: print(f"Invalid lengths: expected {N_TOTAL}") return False # 1. Re-calculate the message hash msg_hash = hashlib.sha256(message).digest() # 2. Re-calculate the SAME Checksum as the client checksum = 0 for byte in msg_hash: checksum += (255 - byte) # Convert checksum to 2 bytes cks_bytes = bytes([(checksum >> 8) & 0xff, checksum & 0xff]) # 3. Combine them exactly like the client did combined = msg_hash + cks_bytes # 4. Verify all 34 chains for i in range(N_TOTAL): try: sig_i = base58.b58decode(signature[i]) pk_i = base58.b58decode(public_key[i]) except Exception as e: return False # Current revealed link check = sig_i # WALK FORWARD to the anchor (Public Key) # The client walked 'combined[i]' steps. # We walk the remaining steps to reach 256. steps = CHAIN_LENGTH - combined[i] for _ in range(steps): check = hashlib.sha256(check).digest() if check != pk_i: print(f"Mismatch at index {i}") return False return True @app.get("/challenge") def get_challenge(): nonce = os.urandom(16) return {"nonce": base58.b58encode(nonce).decode()} ''' @app.post("/relay-txn") async def relay_txn(payload: dict): # 1. Verification Math message = base58.b58decode(payload["challange"]) signature = payload["signature"] # list[str] public_key = payload["anchorB58"] # list[str] is_valid = verify_wots(signature, message, public_key) if not is_valid: return {"success": False, "error": "Invalid WOTS signature"} return {"success": True} ''' @app.post("/relay-txn") async def relay_txn(payload: dict): # Move the heavy math to a background thread is_valid = await run_in_threadpool( verify_wots, payload["signature"], base58.b58decode(payload["challange"]), payload["anchorB58"] ) if not is_valid: return {"success": False, "error": "Invalid WOTS signature"} return {"success": True} @app.post("/sign-message") async def sign_message(payload: dict): #message = bytes(payload["message"]) # 1️⃣ verify quantum proof (same as your txn logic) #proof = base58.b58decode(payload["winternitz_proof"]) #anchor = base58.b58decode(payload["anchor"]) #lap = payload["lap"] #check = proof #for _ in range(256-lap): # check = hashlib.sha256(check).digest() #if check != anchor: # return {"success": False, "error": "Quantum auth failed"} #return { # "success": True #} message = base58.b58decode(payload["message"]) signature = payload["signature"] # list[str] public_key = payload["anchorB58"] # list[str] is_valid = verify_wots(signature, message, public_key) if not is_valid: return {"success": False, "error": "Invalid WOTS signature"} return {"success": True} @app.post("/relay/broadcast/solana") async def broadcast_transaction(payload: TransactionPayload): """ Receives a pre-signed transaction, strips identifying headers and broadcasts to the Solana mainnet. """ try: raw_tx = pybase64.b64decode(payload.signed_hex) response = solana_client.send_raw_transaction(raw_tx,opts=TxOpts( skip_preflight=True, preflight_commitment="processed" )) if not response.value: raise HTTPException(status_code=400, detail="Broadcast failed: No signature returned") return { "success": True, "signature": str(response.value), "status": "Broadcasted via Relay" } except Exception as e: print(f"Relay Error: {str(e)}") return { "success": False, "signature": '', "status": "Failed" } @app.post("/relay/broadcast/ethereum") async def broadcast_ethereum_transaction(payload: EthTransactionPayload): """ Receives a signed ETH hex, strips headers, and broadcasts to the Ethereum network. """ try: # 1. Connection Check if not w3.is_connected(): raise HTTPException(status_code=500, detail="Relayer lost connection to Ethereum Gateway") signed_tx = payload.signed_hex if payload.signed_hex.startswith("0x") else f"0x{payload.signed_hex}" tx_hash = w3.eth.send_raw_transaction(signed_tx) return { "success": True, "signature": tx_hash.hex(), "status": "Relayed to Ethereum Network" } except ValueError as ve: print(f"Validation Error: {str(ve)}") return { "success": False, "signature": tx_hash.hex(), "status": "Validation Error" } except Exception as e: print(f"Relay Error: {str(e)}") return { "success": False, "signature": tx_hash.hex(), "status": "Relay error" }