ik wil een 100% betrouwbare exploit.py the server.py from qiskit import QuantumCircuit, transpile from qiskit_aer import Aer from math import pi import json class PhaseEncoder: def init(self, data: bytes): self.base_circuit = QuantumCircuit(len(data), 1) for i in range(0, len(data), 3): self.base_circuit.rx(self.degrees_to_radians(data[i]), i) if i + 1 < len(data): self.base_circuit.ry(self.degrees_to_radians(data[i + 1]), i + 1) if i + 2 < len(data): self.base_circuit.h(i + 2) self.base_circuit.rz(self.degrees_to_radians(data[i + 2]), i + 2) self.backend = Aer.get_backend("qasm_simulator") self.backend.set_max_qubits(len(data)) def degrees_to_radians(self, degrees: int): return degrees * (pi / 180) def measure(self, circuit: QuantumCircuit, qubit: int): circuit.measure(qubit, 0) compiled = transpile(circuit, self.backend) results = self.backend.run(compiled, shots = 100_000).result() return results.get_counts() def complete_circuit_and_measure(self, qubit: int, instructions: str): if qubit >= self.base_circuit.num_qubits: print(f"Index {qubit} out of range for size {self.base_circuit.num_qubits}") return None if len(instructions) == 0: return self.measure(self.base_circuit, qubit) circuit = self.base_circuit.copy() instructions = instructions.split(";") for instr in instructions: parts = instr.split(":") if len(parts) != 2: print(f"Invalid instruction: {instr}. Expected format: <gate>:<params>") print("Examples: H:<target> | RX:<phase>,<target>") return None gate, params = parts try: params = [ int(p) for p in params.split(",") ] except: print("Quantum gate input parameters must be integers.") print("Examples: H:0 | RX:90,1") return None if len(params) == 2: if params[1] >= self.base_circuit.num_qubits: print(f"Qubit indexes must be less than {self.base_circuit.num_qubits}") return None phase = self.degrees_to_radians(params[0]) if gate == "RX": circuit.rx(phase, params[1]) elif gate == "RY": circuit.ry(phase, params[1]) elif gate == "RZ": circuit.rz(phase, params[1]) else: print(f"Quantum gate '{gate}' is invalid or unexpected with 2 parameters.") return None else: print(f"Unsupported number of parameters ({len(params)}) for quantum gate '{gate}'.") return None return self.measure(circuit, qubit) def main(): encoder = PhaseEncoder(open('flag.txt', 'rb').read()) while True: qubit = input('Specify the qubit index you want to measure : ') try: qubit = int(qubit) except: print('The qubit index must be an integer.') continue instructions = input('Specify the instructions : ') results = encoder.complete_circuit_and_measure(qubit, instructions) if results is None: continue print(json.dumps(results)) if name == 'main': main() ///////////////////////////////// the exploit.py #!/usr/bin/env python3 """ Fast exploit using fresh connections per measurement """ import socket import json import math import re from typing import Dict, Optional, Tuple from concurrent.futures import ThreadPoolExecutor, as_completed class FastQuantumExploit: def init(self, host: str, port: int): self.host = host self.port = port self.recovered_bytes = bytearray() def single_measurement(self, qubit: int, instructions: str = "") -> Optional[Dict]: """Single measurement with fresh connection""" sock = None try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(5) sock.connect((self.host, self.port)) sock.recv(4096) sock.sendall(f"{qubit}\n".encode()) sock.recv(4096) sock.sendall(f"{instructions}\n".encode()) result = sock.recv(8192).decode() json_match = re.search(r'{[^}]+}', result) if json_match: return json.loads(json_match.group()) return None except Exception: return None finally: if sock: sock.close() def fast_estimate_byte(self, qubit: int) -> Optional[Tuple[int, float, float]]: """Fast byte estimation using single measurement""" pattern = qubit % 3 if pattern == 2: instructions = f"RY:-90,{qubit}" else: instructions = "" counts = self.single_measurement(qubit, instructions) if not counts: return None p0 = counts.get('0', 0) / sum(counts.values()) best_byte = None best_error = float('inf') search_range = list(range(32, 127)) + list(range(0, 32)) + list(range(127, 256)) for candidate in search_range: theta = candidate * math.pi / 180 if pattern == 0: expected = math.cos(theta / 2) ** 2 elif pattern == 1: expected = math.cos(theta / 2) ** 2 else: expected = (1 + math.cos(theta)) / 2 error = abs(expected - p0) if error < best_error: best_error = error best_byte = candidate if error < 0.001: break return best_byte, p0, 1 - best_error def parallel_extract_chunk(self, start_idx: int, chunk_size: int, workers: int = 8): """Extract multiple bytes in parallel""" results = {} with ThreadPoolExecutor(max_workers=workers) as executor: futures = { executor.submit(self.fast_estimate_byte, i): i for i in range(start_idx, start_idx + chunk_size) } for future in as_completed(futures): qubit_idx = futures[future] try: result = future.result(timeout=10) if result: results[qubit_idx] = result except Exception as e: print(f"[!] Failed qubit {qubit_idx}: {e}") return results def extract_flag_fast(self, max_length: int = 100): """Fast sequential extraction with fresh connections""" print("[] Fast extraction mode (fresh connection per measurement)") for i in range(max_length): result = self.fast_estimate_byte(i) if result is None: print(f"\n[!] Failed at position {i}") break byte_val, p0, confidence = result self.recovered_bytes.append(byte_val) char = chr(byte_val) if 32 <= byte_val < 127 else '?' display = self.recovered_bytes.decode('ascii', errors='replace') print(f"\r[{i:3d}] {char} (0x{byte_val:02x}) conf={confidence:.3f} p0={p0:.4f} | {display}", end='', flush=True) if byte_val == ord('}') and b'HTB{' in self.recovered_bytes: print("\n[+] Flag complete!") break return self.recovered_bytes def extract_flag_parallel(self, chunk_size: int = 10): """Parallel extraction""" print(f"[] Parallel extraction mode (chunk size: {chunk_size})") position = 0 while position < 100: print(f"\n[*] Extracting chunk {position}-{position+chunk_size-1}...") results = self.parallel_extract_chunk(position, chunk_size, workers=8) for i in range(position, position + chunk_size): if i in results: byte_val, p0, conf = results[i] self.recovered_bytes.append(byte_val) char = chr(byte_val) if 32 <= byte_val < 127 else '?' print(f" [{i:3d}] {char} (0x{byte_val:02x}) conf={conf:.3f}") else: print(f" [{i:3d}] FAILED") self.recovered_bytes.append(ord('?')) display = self.recovered_bytes.decode('ascii', errors='replace') print(f"\n[+] Current: {display}") if ord('}') in self.recovered_bytes[max(0, len(self.recovered_bytes)-chunk_size):]: if b'HTB{' in self.recovered_bytes: print("\n[+] Flag complete!") break position += chunk_size return self.recovered_bytes def run(self, mode: str = "fast"): """Main exploit execution""" try: print(f"[+] Target: {self.host}:{self.port}") if mode == "parallel": flag = self.extract_flag_parallel(chunk_size=10) else: flag = self.extract_flag_fast() print("\n" + "="*60) print("[+] EXTRACTION COMPLETE!") print("="*60) print(f"Hex: {flag.hex()}") print(f"ASCII: {flag.decode('ascii', errors='replace')}") if b'HTB{' in flag: start = flag.index(b'HTB{') try: end = flag.index(b'}', start) + 1 clean_flag = flag[start:end] print(f"Flag: {clean_flag.decode('ascii')}") except: pass print("="*60) except KeyboardInterrupt: print("\n[!] Interrupted") print(f"Partial: {self.recovered_bytes.decode('ascii', errors='replace')}") except Exception as e: print(f"\n[!] Error: {e}") import traceback traceback.print_exc() def main(): import sys if len(sys.argv) < 3: print(f"Usage: {sys.argv[0]} <host> <port> [mode]") print(f" mode: fast (default) or parallel") print(f"Example: {sys.argv[0]} 1.1.1.1 6666 fast") sys.exit(1) host = sys.argv[1] port = int(sys.argv[2]) mode = sys.argv[3] if len(sys.argv) > 3 else "fast" exploit = FastQuantumExploit(host, port) exploit.run(mode=mode) if name == "main": main()