Operator
0799bfbae8
Sanitized version of red team infrastructure automation platform. Operational content (implant pipelines, lures, credential capture) replaced with documented stubs. Architecture and infrastructure automation code intact.
225 lines
8.6 KiB
Python
225 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Common utilities and constants for C2ingRed deployment system
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import random
|
|
import string
|
|
import logging
|
|
import subprocess
|
|
import re
|
|
from datetime import datetime
|
|
|
|
# Constants for providers
|
|
PROVIDERS = ["aws", "linode", "flokinet"]
|
|
DEFAULT_SSH_USER = {
|
|
"aws": "kali",
|
|
"linode": "root",
|
|
"flokinet": "root"
|
|
}
|
|
|
|
# Directory names - maintain correct case for each provider
|
|
PROVIDER_DIRS = {
|
|
"aws": "AWS",
|
|
"linode": "Linode",
|
|
"flokinet": "FlokiNET"
|
|
}
|
|
|
|
# Color codes for terminal output
|
|
COLORS = {
|
|
"RESET": "\033[0m",
|
|
"RED": "\033[91m",
|
|
"GREEN": "\033[92m",
|
|
"YELLOW": "\033[93m",
|
|
"BLUE": "\033[94m",
|
|
"PURPLE": "\033[95m",
|
|
"CYAN": "\033[96m",
|
|
"WHITE": "\033[97m",
|
|
"GRAY": "\033[90m"
|
|
}
|
|
|
|
def clear_screen():
|
|
"""Clear the terminal screen"""
|
|
os.system('cls' if os.name == 'nt' else 'clear')
|
|
|
|
def print_banner():
|
|
"""Print the C2ingRed banner"""
|
|
banner = f"""
|
|
{COLORS['BLUE']}========================================================{COLORS['RESET']}
|
|
{COLORS['BLUE']} ██████╗██████╗ ██╗███╗ ██╗ ██████╗ ██████╗ ███████╗██████╗{COLORS['RESET']}
|
|
{COLORS['BLUE']} ██╔════╝╚════██╗██║████╗ ██║██╔════╝ ██╔══██╗██╔════╝██╔══██╗{COLORS['RESET']}
|
|
{COLORS['BLUE']} ██║ █████╔╝██║██╔██╗ ██║██║ ███╗██████╔╝█████╗ ██║ ██║{COLORS['RESET']}
|
|
{COLORS['BLUE']} ██║ ██╔═══╝ ██║██║╚██╗██║██║ ██║██╔══██╗██╔══╝ ██║ ██║{COLORS['RESET']}
|
|
{COLORS['BLUE']} ╚██████╗███████╗██║██║ ╚████║╚██████╔╝██║ ██║███████╗██████╔╝{COLORS['RESET']}
|
|
{COLORS['BLUE']} ╚═════╝╚══════╝╚═╝╚═╝ ╚═══╝ ╚═════╝ ╚═╝ ╚═╝╚══════╝╚═════╝{COLORS['RESET']}
|
|
{COLORS['BLUE']} {COLORS['RESET']}
|
|
{COLORS['BLUE']} Red Team Infrastructure Deployment Tool {COLORS['RESET']}
|
|
{COLORS['BLUE']}========================================================{COLORS['RESET']}
|
|
"""
|
|
print(banner)
|
|
|
|
def generate_random_string(length=8):
|
|
"""Generate a random string of letters and digits."""
|
|
return ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))
|
|
|
|
def generate_deployment_id():
|
|
"""Generate a consistent deployment ID for all resources in this deployment"""
|
|
from utils.name_generator import generate_deployment_id as generate_verb_animal_id
|
|
return generate_verb_animal_id()
|
|
|
|
def archive_old_logs(max_logs_to_keep=2):
|
|
"""Archive old log files to keep logs directory clean - more aggressive archiving"""
|
|
import glob
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
log_dir = "logs"
|
|
archive_dir = os.path.join(log_dir, "archive")
|
|
|
|
if not os.path.exists(log_dir):
|
|
return
|
|
|
|
# Create archive directory if it doesn't exist
|
|
os.makedirs(archive_dir, exist_ok=True)
|
|
|
|
# Get all log files (excluding archive directory) - ANY existing logs should be archived
|
|
log_files = []
|
|
for pattern in ["deployment_*.log", "teardown_*.log", "20*.log"]:
|
|
log_files.extend(glob.glob(os.path.join(log_dir, pattern)))
|
|
|
|
# Remove duplicates and filter out archive directory
|
|
log_files = [f for f in set(log_files) if "archive" not in f]
|
|
|
|
# Sort by modification time (newest first)
|
|
log_files.sort(key=lambda x: os.path.getmtime(x), reverse=True)
|
|
|
|
# Archive ALL deployment logs to keep directory clean for new deployments
|
|
if len(log_files) > 0:
|
|
print(f"Found {len(log_files)} log files to archive")
|
|
|
|
archived_count = 0
|
|
for log_file in log_files:
|
|
try:
|
|
filename = os.path.basename(log_file)
|
|
# Add timestamp to archived filename to prevent conflicts
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
archived_filename = f"{timestamp}_{filename}"
|
|
archive_path = os.path.join(archive_dir, archived_filename)
|
|
|
|
shutil.move(log_file, archive_path)
|
|
archived_count += 1
|
|
print(f"Archived: {filename} -> archive/{archived_filename}")
|
|
except Exception as e:
|
|
print(f"Failed to archive {log_file}: {e}")
|
|
|
|
if archived_count > 0:
|
|
print(f"Archived {archived_count} log files to {archive_dir}")
|
|
|
|
# Also archive old deployment info files
|
|
info_files = glob.glob(os.path.join(log_dir, "deployment_info_*.txt"))
|
|
|
|
if len(info_files) > 0:
|
|
print(f"Found {len(info_files)} info files to archive")
|
|
for info_file in info_files:
|
|
try:
|
|
filename = os.path.basename(info_file)
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
archived_filename = f"{timestamp}_{filename}"
|
|
archive_path = os.path.join(archive_dir, archived_filename)
|
|
|
|
shutil.move(info_file, archive_path)
|
|
print(f"Archived: {filename} -> archive/{archived_filename}")
|
|
except Exception as e:
|
|
print(f"Failed to archive {info_file}: {e}")
|
|
|
|
def setup_logging(deployment_id=None, operation_type="deployment"):
|
|
"""Set up logging for the deployment or teardown"""
|
|
log_dir = "logs"
|
|
os.makedirs(log_dir, exist_ok=True)
|
|
|
|
# Archive old logs before starting new deployment
|
|
if operation_type == "deployment":
|
|
archive_old_logs()
|
|
|
|
# Create distinct log files for deployment vs teardown operations
|
|
if operation_type == "teardown":
|
|
log_file = os.path.join(log_dir, f"teardown_{deployment_id}.log")
|
|
else:
|
|
log_file = os.path.join(log_dir, f"deployment_{deployment_id}.log")
|
|
|
|
# Clear any existing handlers
|
|
root_logger = logging.getLogger()
|
|
root_logger.handlers.clear()
|
|
|
|
# Configure file handler to log DEBUG and above for full verbosity
|
|
file_handler = logging.FileHandler(log_file)
|
|
file_handler.setLevel(logging.DEBUG)
|
|
file_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
|
|
file_handler.setFormatter(file_formatter)
|
|
|
|
# Add console handler for INFO level and above
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setLevel(logging.INFO)
|
|
console_formatter = logging.Formatter('[%(levelname)s] %(message)s')
|
|
console_handler.setFormatter(console_formatter)
|
|
|
|
# Configure root logger
|
|
root_logger.setLevel(logging.DEBUG) # Capture everything at root level
|
|
root_logger.addHandler(file_handler)
|
|
root_logger.addHandler(console_handler)
|
|
|
|
logging.info(f"{operation_type.capitalize()} operation started")
|
|
logging.info(f"Deployment ID: {deployment_id}")
|
|
logging.info(f"Full verbose output will be captured in: {log_file}")
|
|
return log_file
|
|
|
|
def load_vars_file(provider):
|
|
"""Load vars.yaml for the specified provider"""
|
|
if provider not in PROVIDER_DIRS:
|
|
return {}
|
|
|
|
# Use correct case for directory
|
|
provider_dir = PROVIDER_DIRS[provider]
|
|
vars_file = f"providers/{provider_dir}/vars.yaml"
|
|
|
|
if os.path.exists(vars_file):
|
|
try:
|
|
import yaml
|
|
with open(vars_file, 'r') as f:
|
|
return yaml.safe_load(f) or {}
|
|
except Exception as e:
|
|
logging.error(f"Failed to load {vars_file}: {e}")
|
|
return {}
|
|
else:
|
|
logging.warning(f"Vars file not found: {vars_file}")
|
|
return {}
|
|
|
|
def validate_ip_address(ip):
|
|
"""Validate IP address format"""
|
|
pattern = r'^(\d{1,3}\.){3}\d{1,3}$'
|
|
return re.match(pattern, ip) is not None
|
|
|
|
def get_public_ip():
|
|
"""Get the user's public IP address"""
|
|
try:
|
|
import requests
|
|
return requests.get('https://api.ipify.org', timeout=5).text.strip()
|
|
except:
|
|
return None
|
|
|
|
def confirm_action(message, default=False):
|
|
"""Ask for user confirmation with a yes/no prompt"""
|
|
prompt = f"{message} ({'Y/n' if default else 'y/N'}): "
|
|
response = input(prompt).lower().strip()
|
|
|
|
if not response:
|
|
return default
|
|
|
|
return response in ['y', 'yes']
|
|
|
|
def wait_for_input(message="Press Enter to continue..."):
|
|
"""Wait for user input before continuing"""
|
|
input(f"\n{message}")
|