diff --git a/CLRadio-Hybrid-A-1-i-Rich.py b/CLRadio-Hybrid-A-1-i-Rich.py new file mode 100644 index 0000000..7bfdc87 --- /dev/null +++ b/CLRadio-Hybrid-A-1-i-Rich.py @@ -0,0 +1,226 @@ +import os +import socket +import threading +import time +import sys +from pathlib import Path +from rich.console import Console +from rich.layout import Layout +from rich.panel import Panel +from prompt_toolkit import Application +from prompt_toolkit.layout import Layout as PTLayout, Window +from prompt_toolkit.layout.controls import FormattedTextControl +from prompt_toolkit.key_binding import KeyBindings +from prompt_toolkit.formatted_text import ANSI + +# ----- APP STATE & LOGIC ----- # + +class AppState: + def __init__(self): + self.left_input = "" + self.right_input = "" + self.active_side = "left" + self.logs = ["[bold blue]P7-InfraNET CLRadio-Hybrid A-1-ii Integra[/]"] + self.running = True + self.target_socket = None + self.target_ip = None + self.is_sudo = os.getuid() == 0 if os.name != 'nt' else True + self.cursor_blink = True + + def add_log(self, msg): + self.logs.append(msg) + if len(self.logs) > 100: self.logs.pop(0) + +state = AppState() +console = Console(force_terminal=True, color_system="truecolor", soft_wrap=True) +FILE_PATH = Path("known_connections.txt") + +# ----- FILE LOGIC ----- # + +def get_profiles(): + if not FILE_PATH.exists(): return [] + profiles = [] + with open(FILE_PATH, 'r') as f: + for line in f: + if ": " in line: + profiles.append(line.strip()) + return profiles + +def write_profile(nick, ip): + with open(FILE_PATH, 'a') as f: + f.write(f"{nick}: {ip}\n") + +# ----- NETWORK THREADS ----- # + +def background_receiver(app): + host = '0.0.0.0' + port = 443 + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + s.bind((host, port)) + s.listen(5) + state.add_log("[system] Listening on port 443...") + except PermissionError: + state.add_log("[bold red][!] ERROR: Process Needs Elevation to listen on 443.[/]") + return + + while state.running: + try: + s.settimeout(1.0) + c, addr = s.accept() + state.add_log(f"[bold green][+] Connection from {addr[0]}[/]") + while state.running: + data = c.recv(1024).decode() + if not data: break + state.add_log(f"[bold yellow][R] {addr[0]}:[/] {data}") + app.invalidate() + c.close() + except socket.timeout: + continue + except Exception as e: + state.add_log(f"[red]Receiver Error: {e}[/]") + +# ----- COMMANDS ----- # + +def handle_command(cmd_str): + parts = cmd_str.split() + if not parts: return + cmd = parts[0].lower() + + if cmd == "/help": + state.add_log("[bold cyan]COMMANDS:[/]\n/connect - Connect to nth profile\n/profiles add \n/profiles clear\n/listen - Status check\n/help - This menu") + + elif cmd == "/listen": + state.add_log("[system] Radio is already actively listening on 443.") + + elif cmd == "/profiles": + if len(parts) > 1: + mode = parts[1].lower() + if mode == "add" and len(parts) == 4: + write_profile(parts[2], parts[3]) + state.add_log(f"[system] Added {parts[2]}") + elif mode == "clear": + with open(FILE_PATH, 'w') as f: f.write("") + state.add_log("[system] Profiles cleared.") + else: + state.add_log("[system] Usage: /profiles add or /profiles clear") + + elif cmd == "/connect": + profiles = get_profiles() + if len(parts) == 1: + state.add_log("[bold cyan]PROFILES:[/]") + for i, p in enumerate(profiles, 1): + state.add_log(f" [{i}] {p}") + else: + try: + idx = int(parts[1]) - 1 + target = profiles[idx].split(": ")[1] + state.target_ip = target + # Attempt connection + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(3) + s.connect((target, 443)) + state.target_socket = s + state.add_log(f"[bold green][!] Connected to {target}.[/]") + except Exception as e: + state.add_log(f"[bold red][!] Connection failed: {e}[/]") + +# ----- UI RENDERING ----- # + +def generate_rich_layout(): + try: + size = os.get_terminal_size() + term_width, term_height = size.columns - 2, size.lines + visible_log_lines = max(2, term_height - 10) + except OSError: + term_width, visible_log_lines = 80, 10 + + layout = Layout() + layout.split_row(Layout(name="left", ratio=3), Layout(name="right", ratio=1)) + layout["left"].split_column(Layout(name="left_output", ratio=9), Layout(name="left_input", size=3)) + layout["right"].split_column(Layout(name="right_output", ratio=9), Layout(name="right_input", size=3)) + + # Cursor Logic + cursor = "█" if state.cursor_blink else " " + l_input = state.left_input + (cursor if state.active_side == "left" else "") + r_input = state.right_input + (cursor if state.active_side == "right" else "") + + # Panels + log_content = "\n".join(state.logs[-visible_log_lines:]) + layout["left_output"].update(Panel(log_content, title="Live Console", border_style="blue")) + + status_text = f"Focus: [bold yellow]{state.active_side.upper()}[/]\n\n" + if not state.is_sudo: + status_text += "[bold red]NON-SUDO MODE[/]\nCan only receive.\n\n" + status_text += "[bold cyan]Quick Commands:[/]\n/connect\n/profiles\n/help\n/listen" + + layout["right_output"].update(Panel(status_text, title="Status/Help", border_style="green")) + layout["left_input"].update(Panel(l_input, title="Chat", border_style="cyan" if state.active_side == "left" else "white")) + layout["right_input"].update(Panel(r_input, title="Cmd", border_style="cyan" if state.active_side == "right" else "white")) + + with console.capture() as capture: + console.print(layout, width=term_width) + return ANSI(capture.get()) + +# ----- INPUT HANDLING ----- # + +kb = KeyBindings() + +@kb.add("tab") +def _(event): + state.active_side = "right" if state.active_side == "left" else "left" + +@kb.add("c-q") +def _(event): + state.running = False + if state.target_socket: state.target_socket.close() + event.app.exit() + +@kb.add("enter") +def _(event): + if state.active_side == "left": + if state.left_input and state.target_socket: + try: + state.target_socket.send(state.left_input.encode()) + state.add_log(f"[bold white][S]:[/] {state.left_input}") + except: + state.add_log("[bold red][!] FATAL: Connection lost![/]") + state.target_socket = None + elif not state.target_socket: + state.add_log("[red]No active connection. Use /connect in Cmd panel.[/]") + state.left_input = "" + else: + if state.right_input.startswith("/"): + handle_command(state.right_input) + state.right_input = "" + +@kb.add("backspace") +def _(event): + if state.active_side == "left": state.left_input = state.left_input[:-1] + else: state.right_input = state.right_input[:-1] + +@kb.add("") +def _(event): + if state.active_side == "left": state.left_input += event.data + else: state.right_input += event.data + +# ----- MAIN LOOP ----- # + +def blink_thread(app): + while state.running: + state.cursor_blink = not state.cursor_blink + app.invalidate() + time.sleep(0.5) + +if __name__ == "__main__": + app = Application(layout=PTLayout(Window(content=FormattedTextControl(text=generate_rich_layout))), key_bindings=kb, full_screen=True) + + # Threads + threading.Thread(target=background_receiver, args=(app,), daemon=True).start() + threading.Thread(target=blink_thread, args=(app,), daemon=True).start() + + if not state.is_sudo: + state.add_log("[bold red][!] WARNING: Not running as sudo. Receiver will fail to bind port 443.[/]") + + app.run()