#!/usr/bin/env python3 """Innocuous MAC profile manager — select and apply a believable device identity. Queries data/innocuous_macs.db for consumer device profiles (Fire TV, iPhone, printers, etc.), sets MAC + DHCP hostname + vendor class + TCP stack params to match the selected device. The goal: look like something that belongs on every network. """ import json import logging import os import random import sqlite3 import time from pathlib import Path from typing import Optional from modules.base import BaseModule from utils.networking import ( get_mac, get_primary_interface, get_wifi_interfaces, set_mac, ) from utils.stealth import set_sysctl, set_tcp_timestamps, set_tcp_window, set_ttl logger = logging.getLogger("sensor.stealth.mac_manager") # Category preferences per network type _BUSINESS_CATEGORIES = ("printers", "network", "smart_home") _HOME_CATEGORIES = ("streaming", "smart_home", "phones", "gaming") _DEFAULT_CATEGORIES = ("streaming", "smart_home", "phones") # Map config shorthand to DB device_type values _CATEGORY_MAP = { "streaming": ("smart_tv", "streaming"), "phones": ("phone", "tablet"), "smart_home": ("smart_speaker", "iot"), "printers": ("printer",), "gaming": ("gaming",), "network": ("printer", "iot"), } class MacManager(BaseModule): """Select and apply a consumer-device MAC profile for network blending.""" name = "mac_manager" module_type = "stealth" priority = -400 requires_root = True def __init__(self, bus, state, config, engine=None): super().__init__(bus, state, config, engine) self._profile: Optional[dict] = None self._eth_iface: Optional[str] = None self._wifi_iface: Optional[str] = None self._original_eth_mac: Optional[str] = None self._original_wifi_mac: Optional[str] = None self._db_path = self._resolve_db_path() # ------------------------------------------------------------------ # BaseModule interface # ------------------------------------------------------------------ def start(self) -> None: if self._running: return self._eth_iface = self.config.get("network", {}).get( "primary_interface", "auto" ) if self._eth_iface == "auto": self._eth_iface = get_primary_interface() wifi_ifaces = get_wifi_interfaces() wifi_cfg = self.config.get("network", {}).get("wifi", {}).get("interface", "wlan0") self._wifi_iface = wifi_cfg if wifi_cfg in wifi_ifaces else (wifi_ifaces[0] if wifi_ifaces else None) # Save originals for restore on stop if self._eth_iface: try: self._original_eth_mac = get_mac(self._eth_iface) except Exception: pass if self._wifi_iface: try: self._original_wifi_mac = get_mac(self._wifi_iface) except Exception: pass # Select and apply profile self._profile = self._select_profile() if self._profile is None: logger.warning("No MAC profile found — using random consumer OUI fallback") self._profile = self._fallback_profile() self._apply_profile(self._profile) # Persist for other modules self.state.set(self.name, "profile", json.dumps(self._profile)) self.state.set_module_status(self.name, "running", pid=os.getpid()) self._running = True self._pid = os.getpid() self._start_time = time.time() logger.info( "MAC profile applied: %s (%s) — MAC %s, hostname %s", self._profile.get("device_name", "unknown"), self._profile.get("vendor", "unknown"), self._profile.get("applied_mac", "?"), self._profile.get("dhcp_hostname", "?"), ) def stop(self) -> None: if not self._running: return # Restore original MACs on clean exit — skip WiFi interfaces to avoid # dropping the active association. wifi_ifaces = set(get_wifi_interfaces()) try: if self._eth_iface and self._original_eth_mac and self._eth_iface not in wifi_ifaces: set_mac(self._eth_iface, self._original_eth_mac) if (self._wifi_iface and self._original_wifi_mac and self._wifi_iface != self._eth_iface and self._wifi_iface not in wifi_ifaces): set_mac(self._wifi_iface, self._original_wifi_mac) except Exception as exc: logger.warning("Failed to restore original MAC: %s", exc) self._cleanup_dhclient_conf() self._running = False self.state.set_module_status(self.name, "stopped") logger.info("MacManager stopped — original MACs restored") def status(self) -> dict: base = { "running": self._running, "pid": self._pid, "uptime": time.time() - self._start_time if self._start_time else 0, } if self._profile: base["profile_name"] = self._profile.get("device_name", "unknown") base["applied_mac"] = self._profile.get("applied_mac", "unknown") base["dhcp_hostname"] = self._profile.get("dhcp_hostname", "unknown") if self._eth_iface: try: base["current_eth_mac"] = get_mac(self._eth_iface) except Exception: base["current_eth_mac"] = "error" return base def configure(self, config: dict) -> None: self.config.update(config) if self._running: # Re-select and apply on config change self._profile = self._select_profile() if self._profile: self._apply_profile(self._profile) self.state.set(self.name, "profile", json.dumps(self._profile)) # ------------------------------------------------------------------ # Profile selection # ------------------------------------------------------------------ def _resolve_db_path(self) -> str: """Locate innocuous_macs.db relative to project root.""" candidates = [ os.path.join(os.path.dirname(__file__), "..", "..", "data", "innocuous_macs.db"), "/opt/.cache/bb/data/innocuous_macs.db", ] for p in candidates: resolved = os.path.realpath(p) if os.path.isfile(resolved): return resolved # Default — will be created/populated by setup return os.path.realpath(candidates[0]) def _select_profile(self) -> Optional[dict]: """Select a MAC profile from the database based on config.""" if not os.path.isfile(self._db_path): logger.warning("innocuous_macs.db not found at %s", self._db_path) return None stealth_cfg = self.config.get("stealth", {}) mac_profile = stealth_cfg.get("mac_profile", "auto") network_type = stealth_cfg.get("mac_network_type", "auto") conn = sqlite3.connect(self._db_path) conn.row_factory = sqlite3.Row try: if mac_profile not in ("auto", None) and mac_profile not in _CATEGORY_MAP: # Specific device name requested row = conn.execute( "SELECT * FROM mac_profiles WHERE device_name = ? LIMIT 1", (mac_profile,), ).fetchone() if row: return dict(row) # Category-based selection if mac_profile == "auto" or mac_profile is None: categories = self._pick_categories(network_type) else: categories = _CATEGORY_MAP.get(mac_profile, _DEFAULT_CATEGORIES) placeholders = ",".join("?" for _ in categories) rows = conn.execute( f"SELECT * FROM mac_profiles WHERE device_type IN ({placeholders})", categories, ).fetchall() if rows: return dict(random.choice(rows)) return None except sqlite3.OperationalError: logger.warning("mac_profiles table missing — DB not seeded, using fallback") return None finally: conn.close() def _pick_categories(self, network_type: str) -> tuple: """Choose device categories appropriate for the network type.""" if network_type == "business": cat_keys = _BUSINESS_CATEGORIES elif network_type == "home": cat_keys = _HOME_CATEGORIES else: # Auto: default safe categories cat_keys = _DEFAULT_CATEGORIES types = [] for key in cat_keys: types.extend(_CATEGORY_MAP.get(key, ())) return tuple(types) def _fallback_profile(self) -> dict: """Generate a plausible fallback profile without the database.""" # Amazon Fire TV Stick OUI oui = "FC:65:DE" suffix = ":".join(f"{random.randint(0, 255):02x}" for _ in range(3)) return { "device_type": "streaming", "vendor": "Amazon", "device_name": "Fire TV Stick 4K", "oui": oui, "dhcp_hostname": "amazon-fire-tv", "dhcp_vendor_class": "amazon-fire-tv-stick", "ttl": 64, "tcp_window": 65535, } # ------------------------------------------------------------------ # Apply profile to system # ------------------------------------------------------------------ def _apply_profile(self, profile: dict) -> None: """Set MAC, DHCP config, TCP stack to match the selected profile.""" oui = profile.get("oui", "FC:65:DE") suffix = ":".join(f"{random.randint(0, 255):02x}" for _ in range(3)) new_mac = f"{oui}:{suffix}" profile["applied_mac"] = new_mac wifi_ifaces = set(get_wifi_interfaces()) # Set MAC on Ethernet — skip if it's also a WiFi interface (changing MAC # on an associated WiFi interface drops the connection and causes ENETDOWN # on all AF_PACKET sockets bound to it). if self._eth_iface: if self._eth_iface in wifi_ifaces: logger.info( "Skipping MAC change on %s — active WiFi interface, would break association", self._eth_iface, ) else: try: set_mac(self._eth_iface, new_mac) logger.debug("Set %s MAC to %s", self._eth_iface, new_mac) except Exception as exc: logger.error("Failed to set MAC on %s: %s", self._eth_iface, exc) # Set MAC on WiFi only if it's a separate interface from the primary eth # (e.g., a dedicated monitor card). Same interface = already handled above. if self._wifi_iface and self._wifi_iface != self._eth_iface: wifi_suffix = ":".join(f"{random.randint(0, 255):02x}" for _ in range(3)) wifi_mac = f"{oui}:{wifi_suffix}" try: set_mac(self._wifi_iface, wifi_mac) profile["applied_wifi_mac"] = wifi_mac logger.debug("Set %s MAC to %s", self._wifi_iface, wifi_mac) except Exception as exc: logger.error("Failed to set WiFi MAC on %s: %s", self._wifi_iface, exc) # DHCP configuration self._write_dhclient_conf(profile) # TCP stack tuning ttl = profile.get("ttl", 64) tcp_win = profile.get("tcp_window", 65535) set_ttl(ttl) set_tcp_window(tcp_win) set_tcp_timestamps(True) # Most consumer devices have timestamps enabled logger.debug("TCP stack: TTL=%d, window=%d", ttl, tcp_win) def _write_dhclient_conf(self, profile: dict) -> None: """Write dhclient.conf with hostname + vendor class matching the profile.""" hostname = profile.get("dhcp_hostname", "localhost") vendor_class = profile.get("dhcp_vendor_class", "") conf_path = "/etc/dhcp/dhclient.conf.d" conf_file = os.path.join(conf_path, "bb-profile.conf") try: os.makedirs(conf_path, exist_ok=True) lines = [ f'send host-name "{hostname}";', ] if vendor_class: lines.append(f'send vendor-class-identifier "{vendor_class}";') with open(conf_file, "w") as f: f.write("\n".join(lines) + "\n") logger.debug("Wrote DHCP config: hostname=%s, vendor=%s", hostname, vendor_class) except (IOError, PermissionError) as exc: logger.warning("Failed to write dhclient.conf: %s", exc) def _cleanup_dhclient_conf(self) -> None: """Remove our DHCP config on clean exit.""" conf_file = "/etc/dhcp/dhclient.conf.d/bb-profile.conf" try: if os.path.isfile(conf_file): os.unlink(conf_file) except (IOError, PermissionError): pass