Update CLRadio-Hybrid-A-1-ii-Integra.py
This commit is contained in:
226
CLRadio-Hybrid-A-1-ii-Integra.py
Normal file
226
CLRadio-Hybrid-A-1-ii-Integra.py
Normal file
@@ -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 <n> - Connect to nth profile\n/profiles add <nick> <ip>\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 <nick> <ip> 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("<any>")
|
||||
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()
|
||||
Reference in New Issue
Block a user