Cobra
1eb35c9050
- Replace BigBrother -> SystemMonitor in display names and docstrings - Replace logger names: bb.* -> sensor.* - Replace process names: bb-* -> sensor-* - Replace home directory: ~/.bigbrother -> ~/.implant - Replace LUKS device: /dev/mapper/bb-* -> /dev/mapper/sensor-* - Updated 76 Python files across all modules - Improves OPSEC by removing obvious tool fingerprints from logs and runtime
320 lines
12 KiB
Python
320 lines
12 KiB
Python
#!/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
|