n0mad1k/CoM-c2itall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add Chaos C2 deploy module with local/remote/manage options

Browse Source
This commit is contained in:
n0mad1k
2026-03-21 21:42:29 -04:00
parent acaffecbe8
commit ae3ecf863c
3 changed files with 453 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy.py
+6
View File
@@ -227,6 +227,7 @@ def tools_menu():
print(f"6) Infrastructure Health Check")
print(f"7) Ops Dashboard {COLORS['CYAN']}(Real-Time Engagement Monitor){COLORS['RESET']}")
print(f"8) Claude Bot {COLORS['CYAN']}(Matrix-Claude Code Bridge){COLORS['RESET']}")
print(f"9) Chaos C2 {COLORS['CYAN']}(Deploy / Manage Chaos teamserver){COLORS['RESET']}")
print(f"99) Return to Main Menu")
choice = input(f"\nSelect an option: ")
@@ -254,6 +255,11 @@ def tools_menu():
claude_bot_module = import_module_from_path('deploy_claude_bot', claude_bot_module_path)
if claude_bot_module:
claude_bot_module.claude_bot_menu()
elif choice == "9":
chaos_module_path = os.path.join(os.path.dirname(__file__), 'modules', 'c2', 'deploy_chaos.py')
chaos_module = import_module_from_path('deploy_chaos', chaos_module_path)
if chaos_module:
chaos_module.chaos_menu()
elif choice == "99":
return
else:
modules/c2/__init__.py
View File
modules/c2/deploy_chaos.py
+447
View File
@@ -0,0 +1,447 @@
#!/usr/bin/env python3
"""
Chaos C2 — c2itall integration module
Deploy and manage the Chaos C2 framework (Havoc fork) locally or on a remote host.
"""
import os
import sys
import subprocess
import glob
# Add parent paths for imports
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..'))
from utils.common import COLORS, clear_screen, print_banner, wait_for_input
# ─── Path Registry ────────────────────────────────────────────────────────────
CHAOS_PATH = os.path.expanduser('~/tools/chaos')
INSTALL_SH = os.path.join(CHAOS_PATH, 'Install.sh')
TEAMSERVER = os.path.join(CHAOS_PATH, 'teamserver')
DATA_DIR = os.path.join(CHAOS_PATH, 'data')
LOGS_DIR = os.path.join(DATA_DIR, 'logs')
WS_PATH_FILE = os.path.join(DATA_DIR, '.ws_path')
PROFILES_DIR = os.path.join(CHAOS_PATH, 'profiles')
# ─── SSH helpers ──────────────────────────────────────────────────────────────
def _build_ssh_cmd(host, ssh_user, ssh_port, ssh_key, command=None, interactive=False):
"""Build an SSH command list."""
cmd = ['ssh']
if ssh_key:
cmd.extend(['-i', ssh_key])
cmd.extend(['-p', str(ssh_port)])
known_hosts = os.path.expanduser('~/.ssh/c2deploy_chaos_known_hosts')
cmd.extend([
'-o', 'StrictHostKeyChecking=accept-new',
'-o', f'UserKnownHostsFile={known_hosts}',
'-o', 'IdentitiesOnly=yes',
])
if interactive:
cmd.append('-t')
cmd.append(f'{ssh_user}@{host}')
if command:
cmd.append(command)
return cmd
def _build_scp_cmd(ssh_port, ssh_key=None):
"""Build base SCP command with consistent SSH options."""
known_hosts = os.path.expanduser('~/.ssh/c2deploy_chaos_known_hosts')
cmd = [
'scp',
'-P', str(ssh_port),
'-o', 'StrictHostKeyChecking=accept-new',
'-o', f'UserKnownHostsFile={known_hosts}',
]
if ssh_key:
cmd.extend(['-i', ssh_key])
return cmd
# ─── Remote target prompt ─────────────────────────────────────────────────────
def _prompt_remote_target():
"""Prompt for host/user/port/key. Returns (host, user, port, key) or None on cancel."""
print(f"\n{COLORS['CYAN']}Remote Target{COLORS['RESET']}")
print(f"{COLORS['WHITE']}============={COLORS['RESET']}")
host = input(f" Remote host (IP or hostname): ").strip()
if not host:
print(f"{COLORS['YELLOW']}No host provided{COLORS['RESET']}")
return None, None, None, None
ssh_user = input(f" SSH user [{COLORS['CYAN']}root{COLORS['RESET']}]: ").strip() or 'root'
ssh_port_raw = input(f" SSH port [{COLORS['CYAN']}22{COLORS['RESET']}]: ").strip() or '22'
try:
ssh_port = int(ssh_port_raw)
except ValueError:
ssh_port = 22
ssh_key = input(f" SSH key path (blank for default): ").strip() or None
return host, ssh_user, ssh_port, ssh_key
# ─── Profile selection ────────────────────────────────────────────────────────
def _select_profile(label="Select profile"):
"""List available profiles and let the user choose one. Returns path or None."""
profiles = []
if os.path.isdir(PROFILES_DIR):
for ext in ('*.toml', '*.yaotl', '*.yaml', '*.yml'):
profiles.extend(glob.glob(os.path.join(PROFILES_DIR, ext)))
profiles.sort()
if not profiles:
print(f" {COLORS['YELLOW']}No profiles found in {PROFILES_DIR}{COLORS['RESET']}")
manual = input(f" Enter profile path manually (blank to cancel): ").strip()
return manual or None
print(f"\n {COLORS['CYAN']}{label}:{COLORS['RESET']}")
for i, p in enumerate(profiles, 1):
print(f" {i}) {os.path.basename(p)}")
raw = input(f"\n Select [1]: ").strip() or '1'
try:
idx = int(raw) - 1
if 0 <= idx < len(profiles):
return profiles[idx]
except ValueError:
pass
print(f" {COLORS['YELLOW']}Invalid selection — using first profile{COLORS['RESET']}")
return profiles[0]
# ─── Local actions ────────────────────────────────────────────────────────────
def _deploy_local():
"""Run Install.sh locally after a pre-flight check."""
if not os.path.exists(INSTALL_SH):
print(f"\n{COLORS['RED']}Install.sh not found at {INSTALL_SH}{COLORS['RESET']}")
print(f"{COLORS['YELLOW']}Clone the Chaos repo to ~/tools/chaos first{COLORS['RESET']}")
wait_for_input()
return
print(f"\n{COLORS['CYAN']}Running pre-flight check (Install.sh --check)...{COLORS['RESET']}")
check_result = subprocess.run(
['bash', INSTALL_SH, '--check'],
cwd=CHAOS_PATH,
capture_output=True,
text=True,
)
if check_result.stdout:
print(check_result.stdout)
if check_result.stderr:
print(check_result.stderr)
if check_result.returncode != 0:
print(f"{COLORS['YELLOW']}Pre-flight check reported issues (rc={check_result.returncode}) — continue anyway? (y/N): {COLORS['RESET']}", end='')
if input().strip().lower() != 'y':
wait_for_input()
return
print(f"\n{COLORS['CYAN']}Installing Chaos C2 locally...{COLORS['RESET']}")
print(f"{COLORS['YELLOW']}Note: script may require sudo — you may be prompted for your password.{COLORS['RESET']}\n")
try:
subprocess.run(['bash', INSTALL_SH], cwd=CHAOS_PATH)
except KeyboardInterrupt:
print(f"\n{COLORS['YELLOW']}Installation interrupted{COLORS['RESET']}")
wait_for_input()
def _deploy_remote():
"""SCP the chaos repo to a remote host and run Install.sh over SSH."""
if not os.path.isdir(CHAOS_PATH):
print(f"\n{COLORS['RED']}Chaos source not found at {CHAOS_PATH}{COLORS['RESET']}")
wait_for_input()
return
host, ssh_user, ssh_port, ssh_key = _prompt_remote_target()
if not host:
wait_for_input()
return
remote_staging = '/tmp/chaos-deploy'
print(f"\n{COLORS['CYAN']}Creating remote staging directory on {ssh_user}@{host}...{COLORS['RESET']}")
mkdir_cmd = _build_ssh_cmd(host, ssh_user, ssh_port, ssh_key,
f'rm -rf {remote_staging} && mkdir -p {remote_staging}')
result = subprocess.run(mkdir_cmd, capture_output=True, text=True)
if result.returncode != 0:
print(f"{COLORS['RED']}Failed to create remote staging dir: {result.stderr}{COLORS['RESET']}")
wait_for_input()
return
print(f"{COLORS['CYAN']}Copying Chaos to {ssh_user}@{host}:{remote_staging}...{COLORS['RESET']}")
scp_cmd = _build_scp_cmd(ssh_port, ssh_key)
scp_cmd.extend(['-r', CHAOS_PATH + '/'])
scp_cmd.append(f'{ssh_user}@{host}:{remote_staging}/')
result = subprocess.run(scp_cmd, capture_output=True, text=True)
if result.returncode != 0:
print(f"{COLORS['RED']}SCP failed: {result.stderr}{COLORS['RESET']}")
wait_for_input()
return
print(f" {COLORS['GREEN']}Files copied{COLORS['RESET']}")
print(f"\n{COLORS['CYAN']}Running Install.sh on {host}...{COLORS['RESET']}")
install_cmd_str = f'cd {remote_staging} && bash Install.sh'
ssh_install = _build_ssh_cmd(host, ssh_user, ssh_port, ssh_key,
install_cmd_str, interactive=True)
try:
subprocess.run(ssh_install)
except KeyboardInterrupt:
print(f"\n{COLORS['YELLOW']}Remote install interrupted{COLORS['RESET']}")
wait_for_input()
def _start_teamserver():
"""Start the Chaos teamserver with a selected profile."""
if not os.path.exists(TEAMSERVER):
print(f"\n{COLORS['RED']}teamserver binary not found at {TEAMSERVER}{COLORS['RESET']}")
print(f"{COLORS['YELLOW']}Run 'Deploy Chaos (local)' first to build it{COLORS['RESET']}")
wait_for_input()
return
profile = _select_profile("Select listener profile")
if not profile:
print(f"{COLORS['YELLOW']}No profile selected — aborting{COLORS['RESET']}")
wait_for_input()
return
print(f"\n{COLORS['CYAN']}Starting Chaos teamserver with profile: {os.path.basename(profile)}{COLORS['RESET']}")
print(f"{COLORS['GRAY']}Press Ctrl+C to stop{COLORS['RESET']}\n")
try:
subprocess.run([TEAMSERVER, '-p', profile], cwd=CHAOS_PATH)
except KeyboardInterrupt:
print(f"\n{COLORS['YELLOW']}Teamserver stopped by user{COLORS['RESET']}")
wait_for_input()
def _stop_teamserver():
"""Find and kill the teamserver process."""
print(f"\n{COLORS['CYAN']}Looking for running teamserver process...{COLORS['RESET']}")
result = subprocess.run(
['pgrep', '-f', 'teamserver'],
capture_output=True, text=True
)
pids = result.stdout.strip().splitlines()
if not pids:
print(f"{COLORS['YELLOW']}No teamserver process found{COLORS['RESET']}")
wait_for_input()
return
print(f" Found PID(s): {', '.join(pids)}")
confirm = input(f" {COLORS['YELLOW']}Kill these processes? (y/N): {COLORS['RESET']}").strip().lower()
if confirm != 'y':
print(f"{COLORS['GREEN']}Cancelled{COLORS['RESET']}")
wait_for_input()
return
for pid in pids:
kill_result = subprocess.run(['kill', pid], capture_output=True, text=True)
if kill_result.returncode == 0:
print(f" {COLORS['GREEN']}Killed PID {pid}{COLORS['RESET']}")
else:
print(f" {COLORS['RED']}Failed to kill PID {pid}: {kill_result.stderr.strip()}{COLORS['RESET']}")
wait_for_input()
def _show_status():
"""Check if teamserver is running, show WS path and active profile."""
print(f"\n{COLORS['CYAN']}Chaos C2 Status{COLORS['RESET']}")
print(f"{COLORS['WHITE']}==============={COLORS['RESET']}")
# Check for running process
result = subprocess.run(['pgrep', '-a', '-f', 'teamserver'], capture_output=True, text=True)
if result.stdout.strip():
print(f" Teamserver: {COLORS['GREEN']}RUNNING{COLORS['RESET']}")
for line in result.stdout.strip().splitlines():
print(f" {COLORS['GRAY']}{line}{COLORS['RESET']}")
else:
print(f" Teamserver: {COLORS['RED']}NOT RUNNING{COLORS['RESET']}")
# Show WS path
if os.path.exists(WS_PATH_FILE):
with open(WS_PATH_FILE) as f:
ws_path = f.read().strip()
print(f" WS Path: {COLORS['CYAN']}{ws_path}{COLORS['RESET']}")
else:
print(f" WS Path: {COLORS['GRAY']}(not set){COLORS['RESET']}")
# Show install dir
if os.path.isdir(CHAOS_PATH):
print(f" Install: {COLORS['GREEN']}{CHAOS_PATH}{COLORS['RESET']}")
else:
print(f" Install: {COLORS['RED']}NOT FOUND — {CHAOS_PATH}{COLORS['RESET']}")
# Show available profiles
if os.path.isdir(PROFILES_DIR):
profiles = []
for ext in ('*.toml', '*.yaotl', '*.yaml', '*.yml'):
profiles.extend(glob.glob(os.path.join(PROFILES_DIR, ext)))
if profiles:
print(f" Profiles:")
for p in sorted(profiles):
print(f" {COLORS['GRAY']}{os.path.basename(p)}{COLORS['RESET']}")
else:
print(f" Profiles: {COLORS['YELLOW']}none found in {PROFILES_DIR}{COLORS['RESET']}")
wait_for_input()
def _generate_payload():
"""Interactive payload generation: listener profile, arch, format, output path."""
if not os.path.exists(TEAMSERVER):
print(f"\n{COLORS['RED']}teamserver not found — build Chaos first{COLORS['RESET']}")
wait_for_input()
return
print(f"\n{COLORS['CYAN']}Payload Generation{COLORS['RESET']}")
print(f"{COLORS['WHITE']}=================={COLORS['RESET']}")
profile = _select_profile("Select listener profile for payload")
if not profile:
print(f"{COLORS['YELLOW']}No profile selected — aborting{COLORS['RESET']}")
wait_for_input()
return
print(f"\n Architecture:")
print(f" 1) x86_64 (64-bit)")
print(f" 2) x86 (32-bit)")
arch_choice = input(f" Select [1]: ").strip() or '1'
arch = 'x86_64' if arch_choice != '2' else 'x86'
print(f"\n Format:")
print(f" 1) exe (Windows executable)")
print(f" 2) dll (Windows DLL)")
print(f" 3) bin (raw shellcode)")
fmt_choice = input(f" Select [1]: ").strip() or '1'
fmt_map = {'1': 'exe', '2': 'dll', '3': 'bin'}
fmt = fmt_map.get(fmt_choice, 'exe')
default_out = os.path.join(CHAOS_PATH, 'payloads', f'payload_{arch}.{fmt}')
out_path = input(f"\n Output path [{COLORS['CYAN']}{default_out}{COLORS['RESET']}]: ").strip() or default_out
os.makedirs(os.path.dirname(out_path), exist_ok=True)
print(f"\n{COLORS['CYAN']}Generating payload...{COLORS['RESET']}")
cmd = [TEAMSERVER, 'generate', '--profile', profile,
'--arch', arch, '--format', fmt, '--output', out_path]
try:
result = subprocess.run(cmd, cwd=CHAOS_PATH, capture_output=True, text=True)
if result.stdout:
print(result.stdout)
if result.stderr:
print(result.stderr)
if result.returncode == 0:
print(f"{COLORS['GREEN']}Payload written to: {out_path}{COLORS['RESET']}")
else:
print(f"{COLORS['RED']}Payload generation failed (rc={result.returncode}){COLORS['RESET']}")
print(f"{COLORS['YELLOW']}Note: 'generate' subcommand may differ in your build — check teamserver --help{COLORS['RESET']}")
except FileNotFoundError:
print(f"{COLORS['RED']}teamserver binary not executable — did the build complete?{COLORS['RESET']}")
wait_for_input()
def _view_loot():
"""Tail structured JSON logs from data/logs/."""
if not os.path.isdir(LOGS_DIR):
print(f"\n{COLORS['YELLOW']}Logs directory not found: {LOGS_DIR}{COLORS['RESET']}")
print(f"{COLORS['GRAY']}Logs will appear here after the teamserver has been started{COLORS['RESET']}")
wait_for_input()
return
log_files = sorted(glob.glob(os.path.join(LOGS_DIR, '*.json')) +
glob.glob(os.path.join(LOGS_DIR, '*.log')))
if not log_files:
print(f"\n{COLORS['YELLOW']}No log files found in {LOGS_DIR}{COLORS['RESET']}")
wait_for_input()
return
print(f"\n{COLORS['CYAN']}Available log files:{COLORS['RESET']}")
for i, lf in enumerate(log_files, 1):
size = os.path.getsize(lf)
print(f" {i}) {os.path.basename(lf)} ({size} bytes)")
raw = input(f"\n Select log to tail [1]: ").strip() or '1'
try:
idx = int(raw) - 1
if 0 <= idx < len(log_files):
chosen = log_files[idx]
else:
chosen = log_files[0]
except ValueError:
chosen = log_files[0]
lines_raw = input(f" Lines to show [{COLORS['CYAN']}50{COLORS['RESET']}]: ").strip() or '50'
try:
lines = int(lines_raw)
except ValueError:
lines = 50
print(f"\n{COLORS['CYAN']}--- {os.path.basename(chosen)} (last {lines} lines) ---{COLORS['RESET']}\n")
try:
result = subprocess.run(['tail', '-n', str(lines), chosen],
capture_output=True, text=True)
print(result.stdout)
except Exception as e:
print(f"{COLORS['RED']}Error reading log: {e}{COLORS['RESET']}")
wait_for_input()
# ─── Main menu ────────────────────────────────────────────────────────────────
def chaos_menu():
"""Main entry point — called from c2itall deploy.py tools_menu()."""
while True:
clear_screen()
print_banner()
print(f"{COLORS['WHITE']}CHAOS C2{COLORS['RESET']}")
print(f"{COLORS['WHITE']}========{COLORS['RESET']}")
print(f" Havoc-based C2 framework")
# Show whether source is available
if os.path.isdir(CHAOS_PATH):
ts_label = (f"{COLORS['GREEN']}built{COLORS['RESET']}"
if os.path.exists(TEAMSERVER)
else f"{COLORS['YELLOW']}not built{COLORS['RESET']}")
print(f" Source: {COLORS['GREEN']}{CHAOS_PATH}{COLORS['RESET']} (teamserver: {ts_label})")
else:
print(f" Source: {COLORS['RED']}NOT FOUND — {CHAOS_PATH}{COLORS['RESET']}")
print()
print(f"1) Deploy Chaos {COLORS['CYAN']}(local — runs Install.sh){COLORS['RESET']}")
print(f"2) Deploy Chaos {COLORS['CYAN']}(remote SSH — SCP + install){COLORS['RESET']}")
print(f"3) Start Teamserver")
print(f"4) Stop Teamserver")
print(f"5) Show Status")
print(f"6) Generate Payload")
print(f"7) View Loot {COLORS['GRAY']}(tail data/logs/){COLORS['RESET']}")
print(f"99) Return to Tools Menu")
choice = input(f"\nSelect an option: ").strip()
if choice == '1':
_deploy_local()
elif choice == '2':
_deploy_remote()
elif choice == '3':
_start_teamserver()
elif choice == '4':
_stop_teamserver()
elif choice == '5':
_show_status()
elif choice == '6':
_generate_payload()
elif choice == '7':
_view_loot()
elif choice == '99':
return
else:
print(f"\n{COLORS['RED']}Invalid option. Please try again.{COLORS['RESET']}")
wait_for_input()