| import logging |
| import subprocess |
| import random |
| import time |
| import uuid |
| import ipaddress |
| from typing import Dict, Any |
| import platform |
|
|
| class IdentityManager: |
| def __init__(self, logger: logging.Logger): |
| self.logger = logger |
| self.original_mac = None |
| self.original_ip = None |
| self.interface = self._get_default_interface() |
|
|
| def _get_default_interface(self) -> str: |
| """Detects the default network interface.""" |
| system = platform.system() |
| if system == "Linux": |
| try: |
| result = subprocess.run(["route", "-n"], capture_output=True, text=True, check=True) |
| for line in result.stdout.splitlines(): |
| if "UG" in line and "0.0.0.0" in line: |
| return line.split()[-1] |
| except Exception as e: |
| self.logger.error(f"Error detecting default interface: {e}") |
| return "eth0" |
| elif system == "Darwin": |
| try: |
| result = subprocess.run(["route", "-n", "get", "default"], capture_output=True, text=True, check=True) |
| for line in result.stdout.splitlines(): |
| if "interface:" in line: |
| return line.split("interface:")[1].strip() |
| except Exception as e: |
| self.logger.error(f"Error detecting default interface: {e}") |
| return "en0" |
| else: |
| self.logger.warning(f"Unsupported OS: {system}. Defaulting to eth0.") |
| return "eth0" |
| return "eth0" |
|
|
| def get_current_mac(self, interface: str = None) -> str: |
| interface = interface or self.interface |
| try: |
| result = subprocess.run(["ifconfig", interface], capture_output=True, text=True, check=True) |
| for line in result.stdout.splitlines(): |
| if "ether" in line or "lladdr" in line: |
| parts = line.split("ether" if "ether" in line else "lladdr") |
| if len(parts) > 1: |
| return parts[1].strip().split(" ")[0] |
| return None |
| except Exception as e: |
| self.logger.error(f"Error getting MAC address: {e}") |
| return None |
|
|
| def get_current_ip(self, interface: str = None) -> str: |
| interface = interface or self.interface |
| try: |
| result = subprocess.run(["ifconfig", interface], capture_output=True, text=True, check=True) |
| for line in result.stdout.splitlines(): |
| if "inet " in line: |
| return line.split("inet ")[1].split(" ")[0] |
| return None |
| except Exception as e: |
| self.logger.error(f"Error getting IP address: {e}") |
| return None |
|
|
| def change_mac_address(self, new_mac: str, interface: str = None) -> bool: |
| interface = interface or self.interface |
| try: |
| self.original_mac = self.get_current_mac(interface) |
| subprocess.run(["ifconfig", interface, "down"], check=True) |
| subprocess.run(["ifconfig", interface, "hw", "ether", new_mac], check=True) |
| subprocess.run(["ifconfig", interface, "up"], check=True) |
| self.logger.info(f"MAC address changed to {new_mac} on {interface}") |
| return True |
| except Exception as e: |
| self.logger.error(f"Error changing MAC address: {e}") |
| return False |
|
|
| def restore_mac_address(self, interface: str = None) -> bool: |
| interface = interface or self.interface |
| if not self.original_mac: |
| self.logger.warning("Original MAC address not saved. Cannot restore.") |
| return False |
| try: |
| subprocess.run(["ifconfig", interface, "down"], check=True) |
| subprocess.run(["ifconfig", interface, "hw", "ether", self.original_mac], check=True) |
| subprocess.run(["ifconfig", interface, "up"], check=True) |
| self.logger.info(f"MAC address restored to {self.original_mac} on {interface}") |
| return True |
| except Exception as e: |
| self.logger.error(f"Error restoring MAC address: {e}") |
| return False |
|
|
| def change_ip_address(self, new_ip: str, interface: str = None) -> bool: |
| interface = interface or self.interface |
| try: |
| self.original_ip = self.get_current_ip(interface) |
| subprocess.run(["ifconfig", interface, new_ip], check=True) |
| self.logger.info(f"IP address changed to {new_ip} on {interface}") |
| return True |
| except Exception as e: |
| self.logger.error(f"Error changing IP address: {e}") |
| return False |
|
|
| def restore_ip_address(self, interface: str = None) -> bool: |
| interface = interface or self.interface |
| if not self.original_ip: |
| self.logger.warning("Original IP address not saved. Cannot restore.") |
| return False |
| try: |
| subprocess.run(["ifconfig", interface, self.original_ip], check=True) |
| self.logger.info(f"IP address restored to {self.original_ip} on {interface}") |
| return True |
| except Exception as e: |
| self.logger.error(f"Error restoring IP address: {e}") |
| return False |
|
|
| def start_tor_session(self): |
| self.original_mac = self.get_current_mac() |
| self.original_ip = self.get_current_ip() |
| new_mac = self._generate_random_mac() |
| new_ip = self._generate_random_ip() |
| self.change_mac_address(new_mac) |
| self.change_ip_address(new_ip) |
| self.logger.info("TOR session started with new identity.") |
|
|
| def stop_tor_session(self): |
| self.restore_mac_address() |
| self.restore_ip_address() |
| self.logger.info("TOR session stopped and original identity restored.") |
|
|
| def _generate_random_mac(self) -> str: |
| mac = [0x00, 0x16, 0x3e, |
| random.randint(0x00, 0x7f), |
| random.randint(0x00, 0xff), |
| random.randint(0x00, 0xff)] |
| return ':'.join(map(lambda x: "%02x" % x, mac)) |
|
|
| def _generate_random_ip(self) -> str: |
| ip = ipaddress.IPv4Address(random.randint(0x0B000000, 0xDF000000)) |
| return str(ip) |
|
|