#!/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}")