Python - Advanced Python Debugging: Debugging Performance and Memory Issues

# Basic variable types - critical for understanding memory usage and type-related bugs
integer_var = 42                    # int - whole numbers
float_var = 3.14159                 # float - decimal numbers
string_var = "exploit_payload"      # str - text data
boolean_var = True                  # bool - True/False values

# Collections - understanding these is key for memory profiling
list_var = [1, 2, 3, 4, 5]          # list - mutable ordered collection
tuple_var = (1, 2, 3, 4, 5)         # tuple - immutable ordered collection
dict_var = {"key": "value"}          # dict - key-value pairs
set_var = {1, 2, 3, 4, 5}           # set - unordered unique elements

# Security implication: Mutable defaults can cause data leakage
def insecure_function(data=[]):      # BUG: mutable default argument
    data.append("new_item")
    return data

# Correct approach
def secure_function(data=None):
    if data is None:
        data = []
    data.append("new_item")
    return data

# Arithmetic operators
result = 10 + 5 * 2  # 20, not 30 (multiplication has higher precedence)

# Comparison operators - critical for validation logic
is_admin = user_role == "admin"
is_valid_port = 1 <= port_number <= 65535

# Logical operators - essential for conditional logic in exploits
can_access = is_authenticated and has_permission
should_alert = not is_whitelisted or is_malicious

# Bitwise operators - useful for low-level operations and flag manipulation
flags = 0b1010 & 0b1100  # Bitwise AND
permissions = 0b0100 | 0b0001  # Bitwise OR
inverted = ~0b1010           # Bitwise NOT

# Identity vs equality - common source of bugs
list1 = [1, 2, 3]
list2 = [1, 2, 3]
print(list1 == list2)  # True - values are equal
print(list1 is list2)  # False - different objects in memory

# Conditional statements - core of validation and decision logic
def validate_target(target_ip):
    if not target_ip:
        raise ValueError("Target IP cannot be empty")

    if target_ip.startswith("127."):
        return "localhost"
    elif target_ip.startswith("192.168.") or target_ip.startswith("10."):
        return "private_network"
    else:
        return "public_network"

# Loops - understanding iteration is key for performance debugging
targets = ["192.168.1.1", "10.0.0.1", "8.8.8.8"]

# for loop - explicit iteration
for target in targets:
    result = scan_target(target)
    if result == "vulnerable":
        exploit_target(target)
        break  # Exit loop on first success

# while loop with sentinel - careful with infinite loop potential
attempts = 0
max_attempts = 3
connection_established = False

while not connection_established and attempts < max_attempts:
    try:
        establish_c2_connection()
        connection_established = True
    except ConnectionError:
        attempts += 1
        time.sleep(2 ** attempts)  # Exponential backoff

# List comprehensions - memory-efficient alternatives to loops
vulnerable_hosts = [host for host in network_scan if is_vulnerable(host)]
payload_sizes = [len(generate_payload(target)) for target in targets]

def scan_port(target_ip, port, timeout=5):
    """
    Scan a single port on target IP with timeout.
    Returns tuple: (port, status, banner)
    """
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        result = sock.connect_ex((target_ip, port))

        if result == 0:
            # Port is open, try to get banner
            try:
                banner = sock.recv(1024).decode('utf-8', errors='ignore').strip()
                return (port, "open", banner)
            except:
                return (port, "open", "no_banner")
        else:
            return (port, "closed", "")

    except socket.gaierror:
        raise ValueError(f"Invalid IP address: {target_ip}")
    except Exception as e:
        return (port, "error", str(e))
    finally:
        try:
            sock.close()
        except:
            pass

# Lambda functions - useful for simple operations
is_open_port = lambda result: result[1] == "open"
format_scan_result = lambda ip, results: f"{ip}: {sum(1 for r in results if r[1] == 'open')} open ports"

# Decorators - powerful for adding debugging/logging to functions
def log_execution_time(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"{func.__name__} executed in {end_time - start_time:.4f} seconds")
        return result
    return wrapper

@log_execution_time
def perform_intrusion(target):
    # Complex intrusion logic here
    return "success"

import cProfile
import io
import pstats
from functools import wraps

def profile_function(func):
    """Decorator to profile function execution"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        pr = cProfile.Profile()
        pr.enable()
        result = func(*args, **kwargs)
        pr.disable()

        # Sort by cumulative time
        s = io.StringIO()
        sortby = 'cumulative'
        ps = pstats.Stats(pr, stream=s).sort_stats(sortby)
        ps.print_stats()
        print(s.getvalue())

        return result
    return wrapper

@profile_function
def brute_force_password(hashes, wordlist):
    """Simulated password cracking - watch for performance bottlenecks"""
    cracked = {}
    for hash_value in hashes:
        for password in wordlist:
            if hash_password(password) == hash_value:
                cracked[hash_value] = password
                break
    return cracked

# Direct profiling
hashes = ["hash1", "hash2", "hash3"]
wordlist = ["password123", "admin", "letmein"] * 1000  # Large wordlist

cProfile.run('brute_force_password(hashes, wordlist)', 'profile_output.prof')

# Analyze results
stats = pstats.Stats('profile_output.prof')
stats.sort_stats('cumulative').print_stats(10)  # Top 10 time consumers

import timeit
import hashlib

def hash_password(password, algorithm='sha256'):
    """Hash a password using specified algorithm"""
    if algorithm == 'md5':
        return hashlib.md5(password.encode()).hexdigest()
    elif algorithm == 'sha1':
        return hashlib.sha1(password.encode()).hexdigest()
    elif algorithm == 'sha256':
        return hashlib.sha256(password.encode()).hexdigest()
    else:
        raise ValueError(f"Unsupported algorithm: {algorithm}")

# Compare different hashing approaches
test_password = "my_secret_password_123"

# Setup code for timeit
setup_code = """
import hashlib
def hash_password(password, algorithm='sha256'):
    if algorithm == 'md5':
        return hashlib.md5(password.encode()).hexdigest()
    elif algorithm == 'sha1':
        return hashlib.sha1(password.encode()).hexdigest()
    elif algorithm == 'sha256':
        return hashlib.sha256(password.encode()).hexdigest()
    else:
        raise ValueError(f"Unsupported algorithm: {algorithm}")

test_password = "my_secret_password_123"
"""

# Test different algorithms
md5_time = timeit.timeit('hash_password(test_password, "md5")', setup=setup_code, number=10000)
sha1_time = timeit.timeit('hash_password(test_password, "sha1")', setup=setup_code, number=10000)
sha256_time = timeit.timeit('hash_password(test_password, "sha256")', setup=setup_code, number=10000)

print(f"MD5: {md5_time:.4f}s")
print(f"SHA1: {sha1_time:.4f}s")
print(f"SHA256: {sha256_time:.4f}s")

# Test string concatenation vs join performance
concat_setup = """
words = ['word'] * 10000
"""

concat_time = timeit.timeit('"".join(words)', setup=concat_setup, number=1000)
join_time = timeit.timeit('result = ""; [result := result + word for word in words]', setup=concat_setup, number=1000)

print(f"Join method: {join_time:.4f}s")
print(f"Concat method: {concat_time:.4f}s")

from line_profiler import LineProfiler
import time

def fibonacci_recursive(n):
    """Inefficient recursive Fibonacci - great for profiling"""
    if n <= 1:
        return n
    return fibonacci_recursive(n-1) + fibonacci_recursive(n-2)

def fibonacci_iterative(n):
    """Efficient iterative Fibonacci"""
    if n <= 1:
        return n
    a, b = 0, 1
    for _ in range(2, n+1):
        a, b = b, a + b
    return b

# Profile both functions
lp = LineProfiler()
lp.add_function(fibonacci_recursive)
lp.add_function(fibonacci_iterative)

# Run and profile
lp.runcall(fibonacci_recursive, 30)
lp.runcall(fibonacci_iterative, 30)

lp.print_stats()

from memory_profiler import profile, memory_usage
import numpy as np
import gc

@profile
def memory_intensive_scan():
    """Simulate memory-intensive network scanning"""
    # Create large data structures
    ip_ranges = []
    for i in range(100):
        # Simulate IP ranges with associated data
        ip_range = {
            'network': f'192.168.{i}.0/24',
            'hosts': [f'192.168.{i}.{j}' for j in range(1, 255)],
            'scan_results': {},
            'timestamps': []
        }
        ip_ranges.append(ip_range)

    # Simulate scanning with memory spikes
    total_hosts = sum(len(ip_range['hosts']) for ip_range in ip_ranges)
    print(f"Scanning {total_hosts} hosts...")

    for ip_range in ip_ranges:
        for host in ip_range['hosts'][:10]:  # Scan first 10 hosts per range
            # Simulate scan result storage
            ip_range['scan_results'][host] = {
                'open_ports': np.random.choice([22, 80, 443], size=np.random.randint(0, 4), replace=False),
                'os_fingerprint': 'Linux/Unix',
                'response_time': np.random.uniform(0.1, 2.0)
            }
            ip_range['timestamps'].append(time.time())

    return ip_ranges

def monitor_memory_usage():
    """Monitor memory usage of a function over time"""
    mem_usage = memory_usage((memory_intensive_scan,), interval=0.1)
    print(f"Peak memory usage: {max(mem_usage):.2f} MiB")
    print(f"Average memory usage: {sum(mem_usage)/len(mem_usage):.2f} MiB")

if __name__ == "__main__":
    # Run memory profiling
    results = memory_intensive_scan()

    # Force garbage collection and check
    gc.collect()
    monitor_memory_usage()

import tracemalloc
import linecache

def display_top_memory_users(snapshot, key_type='lineno', limit=10):
    """Display top memory-consuming lines"""
    snapshot = snapshot.filter_traces((
        tracemalloc.Filter(False, "<frozen importlib._bootstrap>"),
        tracemalloc.Filter(False, "<unknown>"),
    ))
    top_stats = snapshot.statistics(key_type)

    print(f"Top {limit} memory allocations:")
    for index, stat in enumerate(top_stats[:limit], 1):
        frame = stat.traceback[0]
        filename = frame.filename
        line = linecache.getline(filename, frame.lineno).strip()
        print(f"#{index}: {filename}:{frame.lineno}: {line}")
        print(f"    Memory: {stat.size / 1024:.1f} KiB")
        print(f"    Count: {stat.count}")
    print()

def memory_intensive_operation():
    """Operation that may have memory issues"""
    tracemalloc.start()

    # Take initial snapshot
    snapshot1 = tracemalloc.take_snapshot()

    # Perform memory-intensive operations
    large_lists = []
    for i in range(100):
        large_lists.append(list(range(10000)))

    # Take second snapshot
    snapshot2 = tracemalloc.take_snapshot()

    # Compare snapshots
    top_stats = snapshot2.compare_to(snapshot1, 'lineno')

    print("Memory allocations between snapshots:")
    for stat in top_stats[:10]:
        frame = stat.traceback[0]
        filename = frame.filename
        line = linecache.getline(filename, frame.lineno).strip()
        print(f"{filename}:{frame.lineno}: {line}")
        print(f"    Size: {stat.size / 1024:.1f} KiB (+{stat.size_diff / 1024:.1f} KiB)")

    # Display top memory users
    display_top_memory_users(snapshot2)

    tracemalloc.stop()

memory_intensive_operation()

import gc
import weakref
import threading
import time

class MemoryLeakDetector:
    def __init__(self):
        self.objects_created = set()
        self.weak_refs = []

    def track_object(self, obj, name=""):
        """Track an object for potential leaks"""
        obj_id = id(obj)
        self.objects_created.add((obj_id, name, type(obj).__name__))

        # Create weak reference to detect when object is deleted
        def callback(ref):
            print(f"Object {name} (id: {obj_id}) was garbage collected")
        self.weak_refs.append(weakref.ref(obj, callback))

    def check_for_leaks(self):
        """Check for objects that haven't been garbage collected"""
        gc.collect()  # Force garbage collection

        leaked_objects = []
        for obj_id, name, type_name in self.objects_created:
            # Check if object still exists
            try:
                obj = ctypes.cast(obj_id, ctypes.py_object).value
                if obj is not None:
                    leaked_objects.append((name, type_name, obj_id))
            except:
                pass  # Object was properly deleted

        if leaked_objects:
            print("POTENTIAL MEMORY LEAKS DETECTED:")
            for name, type_name, obj_id in leaked_objects:
                print(f"  - {name} ({type_name}, id: {obj_id})")
        else:
            print("No memory leaks detected")

        return leaked_objects

def problematic_function(detector):
    """Function that creates potential memory leaks"""
    # Circular reference - common cause of memory leaks
    class Parent:
        def __init__(self):
            self.children = []

    class Child:
        def __init__(self, parent):
            self.parent = parent
            parent.children.append(self)

    # Create circular reference
    parent = Parent()
    child = Child(parent)

    # Track objects
    detector.track_object(parent, "parent")
    detector.track_object(child, "child")

    # Don't return references, creating potential leak
    return "operation_complete"

# Demonstrate memory leak detection
detector = MemoryLeakDetector()

print("Running operation...")
result = problematic_function(detector)

print("\\nChecking for memory leaks...")
leaks = detector.check_for_leaks()

# Force cleanup
gc.collect()
print("\\nAfter forced garbage collection:")
detector.check_for_leaks()

import pdb
import sys
import traceback

class ExploitDebugger:
    def __init__(self):
        self.breakpoints = []
        self.watch_variables = {}

    def debug_exploit(self, exploit_func, *args, **kwargs):
        """Debug an exploit function with comprehensive error handling"""
        try:
            # Set up debugging
            pdb.set_trace()

            # Run exploit
            result = exploit_func(*args, **kwargs)

            print(f"Exploit completed successfully: {result}")
            return result

        except Exception as e:
            print(f"Exploit failed with error: {e}")
            print("Full traceback:")
            traceback.print_exc()

            # Attempt recovery or cleanup
            self.attempt_recovery()

            raise

    def attempt_recovery(self):
        """Attempt to recover from exploit failure"""
        print("Attempting recovery...")

        # Clean up any resources
        # Close connections, remove temp files, etc.

        # Log failure for analysis
        with open('exploit_failures.log', 'a') as f:
            f.write(f"{time.ctime()}: Exploit failure\\n")
            f.write(traceback.format_exc())
            f.write("\\n---\\n")

def buffer_overflow_exploit(target_ip, payload):
    """
    Simulated buffer overflow exploit with debugging hooks
    """
    try:
        # Validate inputs
        if not isinstance(target_ip, str) or not target_ip.count('.') == 3:
            raise ValueError("Invalid target IP format")

        if len(payload) > 1024:
            raise ValueError("Payload too large for buffer")

        # Simulate exploit steps
        print(f"Connecting to {target_ip}...")
        # socket_connection = socket.connect(target_ip, 80)

        print("Sending payload...")
        # socket_connection.send(payload)

        print("Checking for shell...")
        # result = check_shell_access()

        return "exploit_successful"

    except Exception as e:
        print(f"Exploit step failed: {e}")
        raise

# Usage
debugger = ExploitDebugger()

try:
    result = debugger.debug_exploit(
        buffer_overflow_exploit,
        "192.168.1.100",
        b"A" * 512 + b"\\x90\\x90"  # NOP sled + shellcode
    )
except KeyboardInterrupt:
    print("Debugging interrupted by user")
except Exception as e:
    print(f"Final failure: {e}")

import asyncio
import aiohttp
import time
from concurrent.futures import ThreadPoolExecutor
import threading

class OptimizedNetworkScanner:
    def __init__(self, max_concurrent=100):
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.results = {}
        self.lock = threading.Lock()

    async def scan_port_async(self, session, ip, port):
        """Asynchronous port scanning with timeout"""
        async with self.semaphore:
            try:
                start_time = time.time()
                timeout = aiohttp.ClientTimeout(total=5)

                # Attempt connection
                async with session.get(f"http://{ip}:{port}", timeout=timeout) as response:
                    response_time = time.time() - start_time

                    with self.lock:
                        self.results[f"{ip}:{port}"] = {
                            'status': 'open',
                            'response_time': response_time,
                            'banner': response.headers.get('Server', 'unknown')
                        }

            except asyncio.TimeoutError:
                with self.lock:
                    self.results[f"{ip}:{port}"] = {'status': 'timeout'}
            except Exception as e:
                with self.lock:
                    self.results[f"{ip}:{port}"] = {'status': 'closed', 'error': str(e)}

    async def scan_network_async(self, network_prefix, ports):
        """Scan entire network asynchronously"""
        async with aiohttp.ClientSession() as session:
            tasks = []

            # Generate all IP-port combinations
            for i in range(1, 255):  # 192.168.1.1-254
                ip = f"{network_prefix}.{i}"
                for port in ports:
                    task = self.scan_port_async(session, ip, port)
                    tasks.append(task)

            # Execute all scans concurrently
            await asyncio.gather(*tasks, return_exceptions=True)

    def scan_network_threaded(self, network_prefix, ports, max_threads=50):
        """Threaded scanning alternative for comparison"""
        def scan_host(ip):
            for port in ports:
                # Synchronous scan logic here
                pass

        with ThreadPoolExecutor(max_workers=max_threads) as executor:
            futures = []
            for i in range(1, 255):
                ip = f"{network_prefix}.{i}"
                future = executor.submit(scan_host, ip)
                futures.append(future)

            # Wait for completion
            for future in futures:
                future.result()

    def benchmark_scanning(self, network_prefix="192.168.1", ports=[80, 443, 8080]):
        """Compare async vs threaded performance"""
        print("Benchmarking network scanning approaches...")

        # Async approach
        start_time = time.time()
        asyncio.run(self.scan_network_async(network_prefix, ports))
        async_time = time.time() - start_time

        # Reset results
        self.results = {}

        # Threaded approach
        start_time = time.time()
        self.scan_network_threaded(network_prefix, ports)
        threaded_time = time.time() - start_time

        print(f"Async scanning: {async_time:.2f} seconds")
        print(f"Threaded scanning: {threaded_time:.2f} seconds")
        print(".2f")

        return self.results

# Usage example
scanner = OptimizedNetworkScanner(max_concurrent=200)
results = scanner.benchmark_scanning()
print(f"Scanned {len(results)} endpoints")

import hashlib
import itertools
import string
import time
from multiprocessing import Pool, cpu_count
import os
import psutil
from typing import Optional, Tuple, List

class AdvancedPasswordCracker:
    def __init__(self, hash_algorithm: str = 'sha256', charset: Optional[str] = None):
        self.hash_algorithm = hash_algorithm
        self.charset = charset or string.ascii_letters + string.digits + string.punctuation

        # Pre-compute common passwords for hybrid attacks
        self.common_passwords = [
            'password', 'password123', '123456', '123456789', 'qwerty',
            'abc123', 'password1', 'admin', 'letmein', 'welcome',
            'monkey', '1234567890', 'password123456', 'qwerty123'
        ]

        # Performance tracking
        self.stats = {
            'total_attempts': 0,
            'start_time': None,
            'end_time': None,
            'memory_peak': 0
        }

    def hash_password(self, password: str) -> str:
        """Hash password using specified algorithm with salt support"""
        if self.hash_algorithm == 'md5':
            return hashlib.md5(password.encode()).hexdigest()
        elif self.hash_algorithm == 'sha1':
            return hashlib.sha1(password.encode()).hexdigest()
        elif self.hash_algorithm == 'sha256':
            return hashlib.sha256(password.encode()).hexdigest()
        elif self.hash_algorithm == 'sha512':
            return hashlib.sha512(password.encode()).hexdigest()
        else:
            raise ValueError(f"Unsupported hash algorithm: {self.hash_algorithm}")

    def salted_hash_password(self, password: str, salt: str = '') -> str:
        """Hash password with salt for enhanced security"""
        return self.hash_password(password + salt)

    def brute_force_worker(self, args: Tuple[str, int, int, int, str]) -> Tuple[Optional[str], int, int]:
        """Worker function for multiprocessing brute force"""
        target_hash, start_length, end_length, worker_id, salt = args
        attempts = 0

        for length in range(start_length, end_length + 1):
            for candidate in itertools.product(self.charset, repeat=length):
                candidate_str = ''.join(candidate)

                # Apply salt if provided
                if salt:
                    candidate_hash = self.salted_hash_password(candidate_str, salt)
                else:
                    candidate_hash = self.hash_password(candidate_str)

                attempts += 1

                if candidate_hash == target_hash:
                    return candidate_str, attempts, worker_id

                # Progress reporting (every 10000 attempts)
                if attempts % 10000 == 0:
                    memory_usage = psutil.Process(os.getpid()).memory_info().rss / 1024 / 1024
                    print(f"Worker {worker_id}: Tried {attempts} combinations... "
                          f"Memory: {memory_usage:.1f} MB")

        return None, attempts, worker_id

    def brute_force_parallel(self, target_hash: str, max_length: int = 8,
                           num_processes: Optional[int] = None, salt: str = '') -> Optional[str]:
        """Parallel brute force cracking with memory monitoring"""
        if num_processes is None:
            num_processes = min(cpu_count(), 8)  # Cap at 8 processes

        print(f"Starting parallel brute force with {num_processes} processes...")
        print(f"Charset size: {len(self.charset)} characters")
        print(f"Search space: {sum(len(self.charset) ** i for i in range(1, max_length + 1)):,} possibilities")

        self.stats['start_time'] = time.time()

        # Distribute work across processes
        work_items = []
        length_per_worker = max(max_length // num_processes, 1)

        for i in range(num_processes):
            start_len = i * length_per_worker + 1
            end_len = min((i + 1) * length_per_worker, max_length)
            work_items.append((target_hash, start_len, end_len, i, salt))

        with Pool(processes=num_processes) as pool:
            results = pool.map(self.brute_force_worker, work_items)

        # Check results
        total_attempts = 0
        for result in results:
            password, attempts, worker_id = result
            total_attempts += attempts
            if password:
                self.stats['end_time'] = time.time()
                self.stats['total_attempts'] = total_attempts

                duration = self.stats['end_time'] - self.stats['start_time']
                attempts_per_second = total_attempts / duration

                print(f"PASSWORD CRACKED: {password}")
                print(f"Time taken: {duration:.2f} seconds")
                print(f"Total attempts: {total_attempts:,}")
                print(".2f")
                print(".0f")

                return password

        self.stats['end_time'] = time.time()
        print("Password not found in search space")
        return None

    def hybrid_attack(self, target_hash: str, wordlist_path: str,
                     max_mutations: int = 3) -> Optional[str]:
        """Hybrid dictionary + mutation attack"""
        print("Starting hybrid dictionary attack...")
        start_time = time.time()
        attempts = 0

        if not os.path.exists(wordlist_path):
            print(f"Wordlist file not found: {wordlist_path}")
            return None

        try:
            with open(wordlist_path, 'r', encoding='utf-8', errors='ignore') as f:
                base_words = [line.strip() for line in f if line.strip()]

            print(f"Loaded {len(base_words):,} base words from dictionary")

            # Generate mutations for each base word
            for base_word in base_words:
                if len(base_word) < 3:  # Skip very short words
                    continue

                # Test base word
                candidate_hash = self.hash_password(base_word)
                attempts += 1

                if candidate_hash == target_hash:
                    duration = time.time() - start_time
                    print(f"PASSWORD CRACKED: {base_word}")
                    print(f"Time taken: {duration:.2f} seconds")
                    print(f"Attempts: {attempts}")
                    return base_word

                # Generate mutations (limited to avoid explosion)
                mutations = self.generate_mutations(base_word, max_mutations)

                for mutation in mutations:
                    candidate_hash = self.hash_password(mutation)
                    attempts += 1

                    if candidate_hash == target_hash:
                        duration = time.time() - start_time
                        print(f"PASSWORD CRACKED: {mutation}")
                        print(f"Time taken: {duration:.2f} seconds")
                        print(f"Attempts: {attempts}")
                        return mutation

                    if attempts % 50000 == 0:
                        elapsed = time.time() - start_time
                        print(".2f")

        except Exception as e:
            print(f"Error during hybrid attack: {e}")
            return None

        print("Password not found in hybrid attack")
        return None

    def generate_mutations(self, base_word: str, max_mutations: int) -> List[str]:
        """Generate common password mutations"""
        mutations = []

        # Common substitutions
        substitutions = {
            'a': ['@', '4'],
            'e': ['3'],
            'i': ['1', '!'],
            'o': ['0'],
            's': ['$', '5'],
            't': ['7']
        }

        # Simple mutations
        mutations.extend([
            base_word.upper(),
            base_word.lower(),
            base_word.capitalize(),
            base_word + '123',
            base_word + '!',
            base_word + '2023',
            '123' + base_word,
            base_word + '123!',
        ])

        # Character substitutions (limited)
        if max_mutations >= 2:
            for char, subs in substitutions.items():
                if char in base_word.lower():
                    for sub in subs[:1]:  # Limit substitutions
                        mutations.append(base_word.lower().replace(char, sub))

        return mutations[:50]  # Limit total mutations

    def rainbow_table_attack(self, target_hash: str, rainbow_table_path: str) -> Optional[str]:
        """Simulate rainbow table lookup (would need precomputed table)"""
        print("Rainbow table attacks require precomputed tables...")
        print("This is a simulation - real implementation would load massive precomputed tables")

        # In a real implementation, you'd load a rainbow table
        # and perform chain walking to find the original password

        return None

    def benchmark_cracking_methods(self, test_passwords: List[str]) -> dict:
        """Benchmark different cracking approaches"""
        results = {
            'brute_force': [],
            'dictionary': [],
            'hybrid': []
        }

        for test_password in test_passwords:
            test_hash = self.hash_password(test_password)
            print(f"\\nBenchmarking password: {test_password} (hash: {test_hash[:16]}...)")

            # Brute force (limited length)
            if len(test_password) <= 4:
                start_time = time.time()
                result = self.brute_force_parallel(test_hash, max_length=len(test_password))
                duration = time.time() - start_time
                results['brute_force'].append(duration)
                print(".2f")

            # Dictionary attack simulation
            start_time = time.time()
            # Simulate dictionary lookup
            found = test_password in self.common_passwords
            duration = time.time() - start_time + (0.001 if found else 0.1)  # Simulate lookup time
            results['dictionary'].append(duration)

            # Hybrid attack simulation
            start_time = time.time()
            # Simulate hybrid attack
            found = any(test_password in self.generate_mutations(word, 2)
                       for word in self.common_passwords)
            duration = time.time() - start_time + (0.01 if found else 0.5)
            results['hybrid'].append(duration)

        return results

# Usage examples
cracker = AdvancedPasswordCracker('sha256')

# Test with known password
test_password = "pass123"
test_hash = cracker.hash_password(test_password)

print(f"Testing with hash: {test_hash}")

# Brute force (limited length for demo)
print("\\n=== BRUTE FORCE ATTACK ===")
cracked = cracker.brute_force_parallel(test_hash, max_length=6)

# Hybrid attack simulation
print("\\n=== HYBRID ATTACK SIMULATION ===")
# Create a simple test wordlist
test_wordlist = "/tmp/test_words.txt"
with open(test_wordlist, 'w') as f:
    for word in ['password', 'admin', 'test', 'user', 'pass']:
        f.write(word + '\\n')

cracked = cracker.hybrid_attack(test_hash, test_wordlist)

# Benchmark different methods
print("\\n=== BENCHMARKING ===")
test_cases = ["abc", "pass", "test123"]
benchmarks = cracker.benchmark_cracking_methods(test_cases)

for method, times in benchmarks.items():
    if times:
        avg_time = sum(times) / len(times)
        print(".3f")

import logging
import sys
import getpass
import socket
from typing import Optional, List, Dict, Any
import argparse
import hashlib
import hmac
import secrets
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64

# Configure comprehensive logging with security considerations
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('security_tool.log'),
        logging.StreamHandler(sys.stdout)
    ]
)

class SecureTool:
    def __init__(self, debug_mode: bool = False, encryption_key: Optional[bytes] = None):
        self.debug_mode = debug_mode
        self.logger = logging.getLogger(__name__)

        # Input validation patterns with security considerations
        self.ip_pattern = r'^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$'
        self.domain_pattern = r'^(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}$'

        # Secure random key generation for encryption
        if encryption_key:
            self.encryption_key = encryption_key
        else:
            # Generate a secure key using PBKDF2
            password = secrets.token_bytes(32)
            salt = secrets.token_bytes(16)
            kdf = PBKDF2HMAC(
                algorithm=hashes.SHA256(),
                length=32,
                salt=salt,
                iterations=100000,
            )
            self.encryption_key = base64.urlsafe_b64encode(kdf.derive(password))

        self.cipher = Fernet(self.encryption_key)

    def validate_input(self, target: str, input_type: str = 'auto') -> bool:
        """Comprehensive input validation with security checks"""
        import re

        # Length validation to prevent buffer overflow attempts
        if not target or len(target) > 253:  # Max domain length per RFC
            self.logger.warning(f"Input validation failed: invalid length for {target}")
            return False

        # Prevent common injection attacks
        dangerous_chars = [';', '&', '|', '`', '$', '(', ')', '<', '>', '\\']
        if any(char in target for char in dangerous_chars):
            self.logger.warning(f"Input validation failed: dangerous characters in {target}")
            return False

        # Type-specific validation
        if input_type == 'ip' or (input_type == 'auto' and re.match(self.ip_pattern, target)):
            if re.match(self.ip_pattern, target):
                # Additional security: reject localhost/private IPs in production
                if not self.debug_mode and target.startswith(('127.', '192.168.', '10.', '172.')):
                    self.logger.warning(f"Rejected private/local IP in production mode: {target}")
                    return False
                return True
        elif input_type == 'domain' or (input_type == 'auto' and re.match(self.domain_pattern, target)):
            if re.match(self.domain_pattern, target):
                # Additional security: prevent DNS rebinding attacks
                if 'localhost' in target.lower() or target.startswith('127.'):
                    self.logger.warning(f"Rejected localhost domain: {target}")
                    return False
                return True

        return False

    def safe_execute(self, command: str, timeout: int = 30, allowed_commands: Optional[List[str]] = None) -> tuple:
        """Safe command execution with comprehensive security measures"""
        import subprocess
        import shlex

        # Default allowed commands for security tools
        if allowed_commands is None:
            allowed_commands = ['nmap', 'ping', 'dig', 'curl', 'wget', 'traceroute']

        try:
            # Validate command to prevent injection
            if not self._validate_command(command, allowed_commands):
                raise ValueError("Command validation failed - potential security risk")

            # Log command execution attempt
            self.logger.info(f"Executing command: {command[:100]}...")

            # Execute with timeout and restricted privileges
            result = subprocess.run(
                shlex.split(command),
                capture_output=True,
                text=True,
                timeout=timeout,
                # Security: prevent shell injection
                shell=False,
                # Additional security: limit environment
                env={'PATH': '/usr/bin:/bin', 'LANG': 'C'}
            )

            # Log result
            if result.returncode == 0:
                self.logger.info("Command executed successfully")
            else:
                self.logger.warning(f"Command failed with return code {result.returncode}")

            return result.returncode, result.stdout, result.stderr

        except subprocess.TimeoutExpired:
            self.logger.error(f"Command timed out after {timeout} seconds")
            return -1, "", f"Timeout after {timeout} seconds"
        except Exception as e:
            self.logger.error(f"Command execution failed: {e}")
            return -1, "", str(e)

    def _validate_command(self, command: str, allowed_commands: List[str]) -> bool:
        """Validate command safety with whitelist approach"""
        import shlex

        try:
            cmd_parts = shlex.split(command)
            if not cmd_parts:
                return False

            base_cmd = cmd_parts[0]

            # Exact match whitelist
            if base_cmd not in allowed_commands:
                self.logger.warning(f"Command not in whitelist: {base_cmd}")
                return False

            # Additional validation: prevent dangerous arguments
            dangerous_args = ['--rm', '-rf /', 'rm -rf', 'format', 'fdisk']
            for arg in cmd_parts[1:]:
                if any(dangerous in arg for dangerous in dangerous_args):
                    self.logger.warning(f"Dangerous argument detected: {arg}")
                    return False

            return True

        except Exception as e:
            self.logger.error(f"Command validation error: {e}")
            return False

    def encrypt_sensitive_data(self, data: str) -> str:
        """Encrypt sensitive data like credentials or results"""
        try:
            encrypted = self.cipher.encrypt(data.encode())
            return base64.urlsafe_b64encode(encrypted).decode()
        except Exception as e:
            self.logger.error(f"Encryption failed: {e}")
            raise

    def decrypt_sensitive_data(self, encrypted_data: str) -> str:
        """Decrypt sensitive data"""
        try:
            decoded = base64.urlsafe_b64decode(encrypted_data)
            decrypted = self.cipher.decrypt(decoded)
            return decrypted.decode()
        except Exception as e:
            self.logger.error(f"Decryption failed: {e}")
            raise

    def log_operation(self, operation: str, target: str, result: Any, sensitive: bool = False):
        """Comprehensive operation logging with privacy considerations"""
        # Sanitize sensitive data
        if sensitive and isinstance(result, str):
            result = self.mask_sensitive_data(result)

        log_entry = {
            'timestamp': time.time(),
            'operation': operation,
            'target': target,
            'result': str(result)[:500],  # Truncate long results
            'user': getpass.getuser(),
            'hostname': socket.gethostname(),
            'sensitive': sensitive
        }

        self.logger.info(f"Operation: {operation} on {target} - Result: {'[REDACTED]' if sensitive else result}")

        # For high-security environments, encrypt logs
        if sensitive:
            encrypted_log = self.encrypt_sensitive_data(str(log_entry))
            with open('encrypted_logs.enc', 'a') as f:
                f.write(encrypted_log + '\\n')

    def mask_sensitive_data(self, data: str) -> str:
        """Mask sensitive information in logs"""
        import re

        # Mask IP addresses (keep first two octets)
        data = re.sub(r'(\\d+)\\.(\\d+)\\.(\\d+)\\.(\\d+)', r'\\1.\\2.***.***', data)

        # Mask potential passwords/hashes
        data = re.sub(r'password[=:][^\\s]+', 'password=***', data, flags=re.IGNORECASE)
        data = re.sub(r'[a-f0-9]{32,}', '***HASH_REDACTED***', data)  # MD5+ hashes

        return data

    def rate_limit_check(self, operation: str, target: str, max_per_minute: int = 60) -> bool:
        """Implement rate limiting to prevent DoS and detection"""
        # Simple in-memory rate limiting (use Redis/external storage for production)
        current_time = time.time()
        rate_key = f"{operation}:{target}"

        # This is a simplified implementation - real implementation would use persistent storage
        if hasattr(self, '_rate_limits'):
            last_requests = self._rate_limits.get(rate_key, [])
            # Remove requests older than 1 minute
            recent_requests = [t for t in last_requests if current_time - t < 60]

            if len(recent_requests) >= max_per_minute:
                self.logger.warning(f"Rate limit exceeded for {operation} on {target}")
                return False

            recent_requests.append(current_time)
            self._rate_limits[rate_key] = recent_requests
        else:
            self._rate_limits = {rate_key: [current_time]}

        return True

def main():
    parser = argparse.ArgumentParser(description='Secure Python Security Tool')
    parser.add_argument('--target', required=True, help='Target IP or domain')
    parser.add_argument('--debug', action='store_true', help='Enable debug mode')
    parser.add_argument('--timeout', type=int, default=30, help='Operation timeout')
    parser.add_argument('--encrypt-logs', action='store_true', help='Encrypt sensitive log data')

    args = parser.parse_args()

    # Generate encryption key for logs if requested
    encryption_key = None
    if args.encrypt_logs:
        encryption_key = Fernet.generate_key()

    tool = SecureTool(debug_mode=args.debug, encryption_key=encryption_key)

    if not tool.validate_input(args.target):
        print(f"Invalid target: {args.target}")
        sys.exit(1)

    # Rate limiting check
    if not tool.rate_limit_check('scan', args.target):
        print("Rate limit exceeded. Please wait before retrying.")
        sys.exit(1)

    tool.logger.info(f"Starting operation on {args.target}")

    # Example operation with security considerations
    returncode, stdout, stderr = tool.safe_execute(f"ping -c 3 {args.target}", args.timeout)

    if returncode == 0:
        print(f"Target {args.target} is reachable")
        tool.log_operation('ping', args.target, 'success')
    else:
        print(f"Target {args.target} is not reachable: {stderr}")
        tool.log_operation('ping', args.target, f'failed: {stderr}')

    # Demonstrate encryption/decryption
    if args.encrypt_logs:
        test_data = f"Scan results for {args.target}: {stdout[:100]}"
        encrypted = tool.encrypt_sensitive_data(test_data)
        decrypted = tool.decrypt_sensitive_data(encrypted)
        print(f"Encryption test: {'PASSED' if decrypted == test_data else 'FAILED'}")

if __name__ == "__main__":
    main()