Add files via upload

Author: Rickidevs

core/__init__.py

@@ -0,0 +1,75 @@
+import socket
+import threading
+from typing import Callable, Optional
+
+from core.session_manager import SessionManager
+from interface.colors import Colors
+
+BIND_HOST = '0.0.0.0'
+
+
+class Listener(threading.Thread):
+    def __init__(
+        self,
+        host: str,
+        port: int,
+        session_manager: SessionManager,
+        on_connect: Optional[Callable[[int, str, int], None]] = None,
+    ):
+        super().__init__(daemon=True)
+        self.host = host
+        self.port = port
+        self.session_manager = session_manager
+        self.on_connect = on_connect
+        self._running = False
+        self._server_socket: Optional[socket.socket] = None
+
+    def run(self) -> None:
+        self._running = True
+        self._server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self._server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+
+        try:
+            self._server_socket.bind((BIND_HOST, self.port))
+            self._server_socket.listen(10)
+            print(
+                f"\n{Colors.success(f'[*] Listener started — binding {BIND_HOST}:{self.port}, payload LHOST: {self.host}')}"
+            )
+            print(Colors.dim(f"    Waiting for incoming connections...\n"))
+
+            while self._running:
+                try:
+                    self._server_socket.settimeout(1.0)
+                    conn, addr = self._server_socket.accept()
+                    session_id = self.session_manager.add_session(conn, addr)
+                    print(
+                        f"\n{Colors.bold(Colors.GREEN + '[+]' + Colors.ENDC)} "
+                        f"New connection from {Colors.info(addr[0])}:{Colors.warn(str(addr[1]))} "
+                        f"— {Colors.bold('Session ID:')} {Colors.success(str(session_id))}"
+                    )
+                    if self.on_connect:
+                        self.on_connect(session_id, addr[0], addr[1])
+                except socket.timeout:
+                    continue
+                except OSError:
+                    break
+        except Exception as exc:
+            print(Colors.error(f"[!] Listener error: {exc}"))
+        finally:
+            if self._server_socket:
+                try:
+                    self._server_socket.close()
+                except Exception:
+                    pass
+
+    def stop(self) -> None:
+        self._running = False
+        if self._server_socket:
+            try:
+                self._server_socket.close()
+            except Exception:
+                pass
+
+    @property
+    def is_running(self) -> bool:
+        return self._running and self.is_alive()

core/listener.py

@@ -0,0 +1,70 @@
+import socket
+import threading
+from typing import Any, Dict, List, Optional
+
+
+class Session:
+    def __init__(self, session_id: int, conn: socket.socket, addr: tuple):
+        self.id = session_id
+        self.conn = conn
+        self.addr = addr
+        self.ip: str = addr[0]
+        self.port: int = addr[1]
+
+    def send(self, data: str) -> None:
+        self.conn.sendall((data + '\n').encode())
+
+    def recv(self, timeout: float = 3.0) -> str:
+        self.conn.settimeout(timeout)
+        response = b''
+        try:
+            while True:
+                chunk = self.conn.recv(4096)
+                if not chunk:
+                    break
+                response += chunk
+        except Exception:
+            pass
+        finally:
+            self.conn.settimeout(None)
+        return response.decode(errors='replace')
+
+    def close(self) -> None:
+        try:
+            self.conn.close()
+        except Exception:
+            pass
+
+
+class SessionManager:
+    def __init__(self):
+        self._sessions: Dict[int, Session] = {}
+        self._counter: int = 0
+        self._lock = threading.Lock()
+
+    def add_session(self, conn: socket.socket, addr: tuple) -> int:
+        with self._lock:
+            session_id = self._counter
+            self._sessions[session_id] = Session(session_id, conn, addr)
+            self._counter += 1
+            return session_id
+
+    def get_session(self, session_id: int) -> Optional[Session]:
+        with self._lock:
+            return self._sessions.get(session_id)
+
+    def remove_session(self, session_id: int) -> None:
+        with self._lock:
+            if session_id in self._sessions:
+                self._sessions[session_id].close()
+                del self._sessions[session_id]
+
+    def list_sessions(self) -> List[Session]:
+        with self._lock:
+            return list(self._sessions.values())
+
+    def close_all(self) -> None:
+        with self._lock:
+            for session in self._sessions.values():
+                session.close()
+            self._sessions.clear()

core/session_manager.py

@@ -0,0 +1,57 @@
+import ipaddress
+import random
+import socket
+from typing import Dict
+
+import psutil
+
+
+def get_network_interfaces() -> Dict[str, str]:
+    interfaces: Dict[str, str] = {}
+
+    for iface, addrs in psutil.net_if_addrs().items():
+        for addr in addrs:
+            if addr.family == socket.AF_INET:
+                try:
+                    ipaddress.ip_address(addr.address)
+                    interfaces[iface] = addr.address
+                    break
+                except ValueError:
+                    continue
+
+    if not interfaces:
+        try:
+            ip = socket.gethostbyname(socket.gethostname())
+            interfaces['eth0'] = ip
+        except Exception:
+            interfaces['lo'] = '127.0.0.1'
+
+    return interfaces
+
+
+def get_random_port() -> int:
+    while True:
+        port = random.randint(1024, 65535)
+        try:
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                sock.bind(('', port))
+                return port
+        except OSError:
+            continue
+
+
+def is_valid_ip(ip: str) -> bool:
+    try:
+        ipaddress.ip_address(ip)
+        return True
+    except ValueError:
+        return False
+
+
+def is_valid_port(port: str) -> bool:
+    try:
+        p = int(port)
+        return 1 <= p <= 65535
+    except ValueError:
+        return False

core/utils.py

@@ -0,0 +1,35 @@
+class Colors:
+    HEADER = '\033[95m'
+    BLUE = '\033[94m'
+    CYAN = '\033[96m'
+    GREEN = '\033[92m'
+    WARNING = '\033[93m'
+    FAIL = '\033[91m'
+    ENDC = '\033[0m'
+    BOLD = '\033[1m'
+    DIM = '\033[2m'
+    UNDERLINE = '\033[4m'
+
+    @classmethod
+    def success(cls, text: str) -> str:
+        return f"{cls.GREEN}{text}{cls.ENDC}"
+
+    @classmethod
+    def error(cls, text: str) -> str:
+        return f"{cls.FAIL}{text}{cls.ENDC}"
+
+    @classmethod
+    def warn(cls, text: str) -> str:
+        return f"{cls.WARNING}{text}{cls.ENDC}"
+
+    @classmethod
+    def info(cls, text: str) -> str:
+        return f"{cls.CYAN}{text}{cls.ENDC}"
+
+    @classmethod
+    def bold(cls, text: str) -> str:
+        return f"{cls.BOLD}{text}{cls.ENDC}"
+
+    @classmethod
+    def dim(cls, text: str) -> str:
+        return f"{cls.DIM}{text}{cls.ENDC}"