#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Quantum Eraser – CHAOS ENGINE (v6.0) ==================================== Features: 1. MULTI-ALGORITHM: Brute Force, Genetic, Simulated Annealing. 2. NON-LINEAR SCHEDULER: Scanează lungimi aleatoriu, nu secvențial. 3. HEURISTIC LOCK: Insistă pe lungimi care dau semnale slabe. 4. LIVE CHAOS VISUALS: Grafice și log-uri dinamice. """ import tkinter as tk from tkinter import ttk, messagebox import numpy as np from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg from matplotlib.figure import Figure import threading import time import random from concurrent.futures import ProcessPoolExecutor import multiprocessing import math # ==================== MATH ENGINE ==================== def generate_keystream_fast(key_int: int, length: int) -> np.ndarray: np.random.seed(key_int) return np.random.randint(0, 2, length, dtype=np.int8) def evaluate_key_score(args): """Funcție pură pentru calcul scor.""" key_int, s_bits, cum_shots = args I_guess = generate_keystream_fast(key_int, len(s_bits)) scores = [] idx_start = 0 for idx_end in cum_shots[1:]: seg_I = I_guess[idx_start:idx_end] seg_S = s_bits[idx_start:idx_end] m0 = (seg_I == 0) c0 = np.sum(m0) c1 = len(seg_S) - c0 if c0 > 0 and c1 > 0: p0 = np.sum(seg_S[m0]) / c0 p1 = np.sum(seg_S[~m0]) / c1 scores.append(abs(p0 - p1)) idx_start = idx_end if not scores: return 0.0 return np.mean(scores) # ==================== GUI & CONTROLLER ==================== class ChaosHackerApp(tk.Tk): def __init__(self): super().__init__() self.title("QUANTUM BREACH // CHAOS ENGINE v6.0") self.geometry("1600x900") self.state('zoomed') # Cyberpunk Theme self.bg_color = "#050505" self.fg_color = "#00ff41" self.err_color = "#ff3333" self.configure(bg=self.bg_color) self.style = ttk.Style(self) self.style.theme_use('clam') self.style.configure(".", background=self.bg_color, foreground=self.fg_color, font=("Consolas", 10)) self.style.configure("TLabel", background=self.bg_color, foreground=self.fg_color) self.style.configure("TButton", background="#111", foreground=self.fg_color, bordercolor=self.fg_color) self.style.map("TButton", background=[('active', '#222')]) self.style.configure("TLabelframe", background=self.bg_color, foreground=self.fg_color, bordercolor="#004400") self.style.configure("TLabelframe.Label", background=self.bg_color, foreground=self.fg_color, font=("Consolas", 11, "bold")) # Simulation State self.running = False self.experiment_data = None # (s_bits, cum_shots, phis) self.target_key_len = 0 # Doar pentru generare, atacul NU o știe! self.found_key = None # Visual State self.active_algo = "IDLE" self.active_len = 0 self.best_score_global = 0.0 self._build_ui() self._animate_visuals() def _build_ui(self): # Top Header tk.Label(self, text="QUANTUM BREACH // AUTONOMOUS AGENT", font=("Consolas", 22, "bold"), bg="#001100", fg=self.fg_color).pack(fill=tk.X, pady=0) main = tk.Frame(self, bg=self.bg_color) main.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) # Left: Controls col1 = tk.Frame(main, bg=self.bg_color, width=350) col1.pack(side=tk.LEFT, fill=tk.Y, padx=10) grp_set = ttk.LabelFrame(col1, text=" [ TARGET CONFIGURATION ] ") grp_set.pack(fill=tk.X, pady=10) ttk.Label(grp_set, text="Encryption Difficulty (Bits):").pack(anchor="w", padx=5) self.scale_diff = tk.Scale(grp_set, from_=8, to=40, orient=tk.HORIZONTAL, bg=self.bg_color, fg=self.fg_color, highlightthickness=0) self.scale_diff.set(16) self.scale_diff.pack(fill=tk.X, padx=5, pady=5) self.btn_start = ttk.Button(grp_set, text="DEPLOY CHAOS AGENT", command=self.start_simulation) self.btn_start.pack(fill=tk.X, padx=5, pady=10) grp_stat = ttk.LabelFrame(col1, text=" [ AGENT STATUS ] ") grp_stat.pack(fill=tk.X, pady=10) self.lbl_algo = ttk.Label(grp_stat, text="MODULE: STANDBY", font=("Consolas", 12, "bold")) self.lbl_algo.pack(anchor="w", padx=5, pady=2) self.lbl_target = ttk.Label(grp_stat, text="SCANNING: ---") self.lbl_target.pack(anchor="w", padx=5, pady=2) self.lbl_conf = ttk.Label(grp_stat, text="CONFIDENCE: 0.00%") self.lbl_conf.pack(anchor="w", padx=5, pady=2) self.log_text = tk.Text(col1, height=25, width=40, bg="black", fg="#00aa00", font=("Consolas", 9), relief="flat") self.log_text.pack(fill=tk.BOTH, expand=True, pady=10) # Center: Graph col2 = tk.Frame(main, bg=self.bg_color) col2.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=10) self.fig = Figure(figsize=(5, 4), dpi=100, facecolor=self.bg_color) self.ax = self.fig.add_subplot(111) self.ax.set_facecolor("black") self.ax.grid(color='#003300', linewidth=0.5) for spine in self.ax.spines.values(): spine.set_color(self.fg_color) self.ax.tick_params(colors=self.fg_color) self.ax.set_title("SIGNAL RECOVERY FEED", color=self.fg_color) self.canvas = FigureCanvasTkAgg(self.fig, col2) self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) # Right: Algorithms Monitor (The "Brain") col3 = ttk.LabelFrame(main, text=" [ STRATEGY QUEUE ] ") col3.pack(side=tk.LEFT, fill=tk.Y, padx=10) self.queue_text = tk.Text(col3, width=35, bg="black", fg="#55ff55", font=("Consolas", 10), relief="flat") self.queue_text.pack(fill=tk.BOTH, expand=True) self.queue_text.tag_config("curr", background="#004400", foreground="#ffffff") self.queue_text.tag_config("done", foreground="#555555") def log(self, msg): self.log_text.insert(tk.END, f"> {msg}\n") self.log_text.see(tk.END) def _animate_visuals(self): # Efecte vizuale "idle" sau active if self.running and self.active_algo != "IDLE": # Adaugă noise pe grafic pentru a arăta că "gândește" if self.experiment_data: phis = self.experiment_data[2] noise = np.random.rand(len(phis)) * 0.2 + 0.4 # Noise centrat pe 0.5 # Curățăm liniile vechi de noise lines = self.ax.lines while len(lines) > 1: # Păstrăm linia verde (index 0) dacă există lines[-1].remove() color = "#ff0000" if self.best_score_global < 0.8 else "#ffff00" self.ax.plot(phis, noise, color=color, alpha=0.1, linewidth=1) self.canvas.draw_idle() self.after(100, self._animate_visuals) # ==================== LOGICĂ DE SIMULARE ==================== def start_simulation(self): if self.running: return self.running = True self.btn_start.config(state="disabled") self.best_score_global = 0.0 self.found_key = None self.active_algo = "INITIALIZING" bits = self.scale_diff.get() self.target_key_len = bits threading.Thread(target=self._run_experiment_and_attack, args=(bits,), daemon=True).start() def _run_experiment_and_attack(self, bits): # 1. Generează Datele "Secrete" self.log(f"GENERATING {bits}-BIT ENCRYPTED STREAM...") true_key = random.getrandbits(bits) # Setup fizic n_angles = 16 shots = 2000 phis = np.linspace(0, 2*np.pi, n_angles) np.random.seed(true_key) full_idler = np.random.randint(0, 2, shots*n_angles, dtype=np.int8) s_bits, cum_shots = [], [0] idx = 0 for phi in phis: prob = (1 + np.cos(phi))/2 rnd = np.random.rand(shots) seg_I = full_idler[idx:idx+shots] seg_S = np.zeros(shots) m0 = (seg_I==0) seg_S[m0] = (rnd[m0] < prob).astype(np.int8) seg_S[~m0] = (rnd[~m0] < 0.5).astype(np.int8) s_bits.extend(seg_S) cum_shots.append(cum_shots[-1]+shots) idx += shots self.experiment_data = (np.array(s_bits), np.array(cum_shots), phis) self.log("DATA INTERCEPTED. STARTING CHAOS ENGINE.") # Inițializare grafic (linie plată) self.ax.clear() self.ax.grid(color='#003300') self.ax.set_ylim(0, 1) self.ax.plot(phis, [0.5]*len(phis), color=self.fg_color, linewidth=2) self.canvas.draw() # 2. START CHAOS SCHEDULER self._chaos_scheduler() def _chaos_scheduler(self): """ Creierul atacului. Nu știe lungimea cheii. Generează task-uri aleatorii și le prioritizează pe cele promițătoare. """ pool = ProcessPoolExecutor(max_workers=multiprocessing.cpu_count()) # Lista inițială de ipoteze (Randomized) possible_lengths = [8, 10, 12, 14, 16, 18, 20, 24, 28, 32, 36, 40] random.shuffle(possible_lengths) # NU le luăm la rând! # Coada de task-uri: (Length, Algorithm, Priority) # Algoritmi: BF (BruteForce), GA (Genetic), SA (Simulated Annealing) task_queue = [] for l in possible_lengths: algo = "BF" if l <= 14 else random.choice(["GA", "SA"]) task_queue.append({"len": l, "algo": algo, "status": "pending"}) while self.running and task_queue: # Luăm primul task task = task_queue.pop(0) length = task['len'] algo = task['algo'] # Update UI self.active_algo = algo self.active_len = length self.after(0, lambda: self.lbl_algo.config(text=f"MODULE: {self._get_algo_name(algo)}")) self.after(0, lambda: self.lbl_target.config(text=f"SCANNING LEN: {length} bits")) self._update_queue_display(task, task_queue) self.log(f"Trying Length {length} with {algo}...") # Execută Task result_score, result_key = 0.0, 0 if algo == "BF": result_score, result_key = self._run_bruteforce(pool, length) elif algo == "GA": result_score, result_key = self._run_genetic(pool, length) elif algo == "SA": result_score, result_key = self._run_annealing(pool, length) # Analiză Rezultat self.after(0, lambda: self.lbl_conf.config(text=f"CONFIDENCE: {result_score*100:.2f}%")) if result_score > self.best_score_global: self.best_score_global = result_score self._update_graph_best(result_key, length) if result_score > 0.85: self.log(f"!!! MATCH FOUND AT LEN {length} !!!") self.found_key = result_key self._success_sequence(result_key, length) break # HEURISTIC FEEDBACK (Important!) # Dacă scorul e mediocru (>0.53) dar nu perfect, e SUSPECT. # Poate algoritmul Genetic nu a avut timp, sau Annealing s-a blocat. # Re-adăugăm task-ul în coadă cu prioritate și alt algoritm! if 0.53 < result_score < 0.85: self.log(f"-> Suspicious signal at {length} bits ({result_score:.3f}). Retrying aggressively.") new_algo = "SA" if algo == "GA" else "GA" # Schimbăm strategia task_queue.insert(0, {"len": length, "algo": new_algo, "status": "RETRY"}) # Dacă scorul e praf (<0.51), e zgomot. Trecem mai departe. pool.shutdown() if not self.found_key and self.running: self.log("QUEUE EMPTY. ATTACK FAILED.") self.running = False self.btn_start.config(state="normal") # ==================== ALGORITHMS ==================== def _run_bruteforce(self, pool, length): space = 1 << length limit = 4000 # Limităm pentru demo candidates = [random.getrandbits(length) for _ in range(min(space, limit))] # Dacă spațiul e mic, asigurăm acoperire totală if space < limit: candidates = list(range(space)) tasks = [(c, self.experiment_data[0], self.experiment_data[1]) for c in candidates] results = list(pool.map(evaluate_key_score, tasks)) best_idx = np.argmax(results) return results[best_idx], candidates[best_idx] def _run_genetic(self, pool, length): pop_size = 800 gens = 25 population = [random.getrandbits(length) for _ in range(pop_size)] best_s, best_k = 0.0, 0 for g in range(gens): if not self.running: break tasks = [(c, self.experiment_data[0], self.experiment_data[1]) for c in population] scores = list(pool.map(evaluate_key_score, tasks)) # Sort pop_scores = sorted(zip(scores, population), key=lambda x: x[0], reverse=True) best_s, best_k = pop_scores[0] if best_s > 0.85: return best_s, best_k # Evolve new_pop = [x[1] for x in pop_scores[:50]] # Elitism while len(new_pop) < pop_size: p1 = random.choice(pop_scores[:200])[1] p2 = random.choice(pop_scores[:200])[1] mask = random.getrandbits(length) child = (p1 & mask) | (p2 & ~mask) if random.random() < 0.2: child ^= (1 << random.randint(0, length-1)) new_pop.append(child) population = new_pop return best_s, best_k def _run_annealing(self, pool, length): # Simulated Annealing simplificat (paralel) # Lansăm 20 de "căutători" care pleacă din puncte random walkers = 20 steps = 50 current_keys = [random.getrandbits(length) for _ in range(walkers)] # Evaluăm starea inițială tasks = [(k, self.experiment_data[0], self.experiment_data[1]) for k in current_keys] current_scores = list(pool.map(evaluate_key_score, tasks)) best_s_local = max(current_scores) best_k_local = current_keys[current_scores.index(best_s_local)] temp = 1.0 cooling = 0.90 for step in range(steps): if best_s_local > 0.85: break # Generăm vecini (flip 1 bit) neighbors = [] for k in current_keys: neighbors.append(k ^ (1 << random.randint(0, length-1))) # Evaluăm vecinii tasks = [(k, self.experiment_data[0], self.experiment_data[1]) for k in neighbors] neighbor_scores = list(pool.map(evaluate_key_score, tasks)) # Decizie Metropolis for i in range(walkers): delta = neighbor_scores[i] - current_scores[i] if delta > 0 or math.exp(delta / temp) > random.random(): current_keys[i] = neighbors[i] current_scores[i] = neighbor_scores[i] if current_scores[i] > best_s_local: best_s_local = current_scores[i] best_k_local = current_keys[i] temp *= cooling return best_s_local, best_k_local # ==================== HELPERS & VISUALS ==================== def _get_algo_name(self, code): return {"BF": "BRUTE FORCE", "GA": "GENETIC ALGO", "SA": "SIM. ANNEALING"}.get(code, code) def _update_queue_display(self, current, queue): self.queue_text.delete("1.0", tk.END) # Current self.queue_text.insert(tk.END, f">> EXEC: Len={current['len']} [{current['algo']}]\n", "curr") # Next for t in queue[:10]: tag = "retry" if t.get("status") == "RETRY" else "" self.queue_text.insert(tk.END, f" WAIT: Len={t['len']} [{t['algo']}] {t.get('status', '')}\n") if len(queue) > 10: self.queue_text.insert(tk.END, f" ... (+{len(queue)-10} tasks)") def _update_graph_best(self, key, length): s_bits, cum_shots, phis = self.experiment_data I_guess = generate_keystream_fast(key, len(s_bits)) vals = [] idx_start = 0 for idx_end in cum_shots[1:]: seg_I = I_guess[idx_start:idx_end] seg_S = s_bits[idx_start:idx_end] m0 = (seg_I == 0) if np.sum(m0)>0: vals.append(np.mean(seg_S[m0])) else: vals.append(0.5) idx_start = idx_end self.ax.clear() self.ax.grid(color='#003300') self.ax.set_ylim(0, 1) self.ax.plot(phis, vals, 'o-', color=self.fg_color, linewidth=2, label="Decoded") self.canvas.draw() def _success_sequence(self, key, length): messagebox.showinfo("BREACH SUCCESSFUL", f"KEY FOUND!\nLength: {length}\nBinary: {bin(key)}") self.running = False self.btn_start.config(state="normal") self.lbl_algo.config(text="SYSTEM: COMPROMISED", foreground="#ffffff", background="#00aa00") if __name__ == "__main__": multiprocessing.freeze_support() app = ChaosHackerApp() app.mainloop()