#!/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 try: if self._eth_iface and self._original_eth_mac: set_mac(self._eth_iface, self._original_eth_mac) if self._wifi_iface and self._original_wifi_mac: 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 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 # Set MAC on Ethernet if self._eth_iface: 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 (different suffix for uniqueness) if self._wifi_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