#!/usr/bin/env python3 """ Firmware Hunter Pro v4.0 A safer offline firmware triage framework for extracted router / IoT firmware. What it does: - Scans extracted firmware directories safely without executing firmware binaries. - Optionally extracts a firmware image with binwalk if binwalk is installed. - Finds credentials, keys, certs, IPs, domains, URLs, MACs, JWTs, API keys, hashes, CVEs. - Maps web UI files, CGI handlers, forms, scripts, endpoints, and admin-looking routes. - Detects ELF binaries, architecture hints, BusyBox, Linux kernel, OpenSSL, Dropbear, dnsmasq, uClibc/musl/glibc. - Scores findings with severity and confidence. - Produces TXT, JSON, HTML, Markdown, CSV, and separate evidence files. - Supports simple external plugins from a plugins directory. Safe by default: - Does not execute firmware binaries. - Only reads files and optionally runs host tools like strings, file, binwalk, yara. Usage: python3 firmware_hunter_pro.py /path/to/squashfs-root python3 firmware_hunter_pro.py firmware.bin --extract python3 firmware_hunter_pro.py /path/to/rootfs --quick python3 firmware_hunter_pro.py /path/to/rootfs --yara rules.yar python3 firmware_hunter_pro.py /path/to/rootfs --plugins plugins/ """ import os import re import csv import sys import json import math import time import html import shutil import hashlib import argparse import tempfile import subprocess import importlib.util from pathlib import Path from datetime import datetime from collections import Counter, defaultdict from concurrent.futures import ThreadPoolExecutor, as_completed from threading import Lock VERSION = "4.0" REPORT = defaultdict(list) REPORT_LOCK = Lock() # ============================================================================= # CONFIG # ============================================================================= TEXT_EXTENSIONS = { ".txt", ".log", ".conf", ".cfg", ".ini", ".json", ".xml", ".html", ".htm", ".js", ".php", ".asp", ".cgi", ".lua", ".sh", ".service", ".default", ".profile", ".passwd", ".shadow", ".pem", ".key", ".crt", ".pub", ".yaml", ".yml", ".env", ".properties", ".rc", ".rules" } INTERESTING_NAMES = { "passwd", "shadow", "group", "hosts", "resolv.conf", "inittab", "fstab", "rcS", "rc.local", "profile", "authorized_keys", "known_hosts", "motd", "issue", "services", "inetd", "inetd.conf", "udhcpd.conf", "dnsmasq.conf", "dropbear", "telnetd", "lighttpd.conf", "boa.conf", "httpd.conf", "mini_httpd.conf", "uhttpd.conf", "nginx.conf", "config.xml", "nvram", "default.cfg", "wpa_supplicant.conf", "os-release", "version", "release", "syslog.conf", "crontab" } EXECUTABLE_NAMES = { "busybox", "telnetd", "dropbear", "sshd", "httpd", "boa", "lighttpd", "mini_httpd", "uhttpd", "nginx", "nc", "netcat", "wget", "curl", "tftp", "ftpget", "ftpput", "iptables", "ip6tables", "dnsmasq", "udhcpd", "pppd", "openvpn", "openssl", "sqlite3", "ash", "sh" } SUSPICIOUS_KEYWORDS = { "telnetd", "dropbear", "busybox telnet", "nc -l", "netcat", "/bin/sh", "/bin/ash", "reverse shell", "backdoor", "wget http", "curl http", "tftp", "ftpget", "ftpput", "chmod 777", "0.0.0.0", "admin:admin", "root:root", "password", "passwd", "shadow", "debug", "test", "factory", "developer", "enable telnet", "remote shell", "hardcoded", "debug shell", "diagnostic", "support account", "superuser", "hidden", "maintenance" } MALWARE_IOC_KEYWORDS = { "mirai": ["mirai", "busybox MIRAI", "/bin/busybox", "report.%s", "scanListen"], "gafgyt/bashlite": ["gafgyt", "bashlite", "gayfgt", "loligang", "telnet scanner"], "mozi": ["mozi", "Mozi.m", "dht.transmissionbt.com", "router.bittorrent.com"], "xorddos": ["xorddos", "x0r", "/tmp/.x", "BB2FA36AAA9541F0"], "miner": ["stratum+tcp", "xmrig", "minerd", "cryptonight", "monero"], "generic_bot": ["CNC", "C2", "botnet", "udp flood", "syn flood", "http flood"] } REGEX_PATTERNS = { "ipv4": r"\b(?:\d{1,3}\.){3}\d{1,3}\b", "mac_address": r"\b(?:[0-9a-fA-F]{2}[:-]){5}[0-9a-fA-F]{2}\b", "url": r"https?://[^\s'\"<>]{4,}", "email": r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+", "domain": r"\b(?:[a-zA-Z0-9-]+\.)+(?:com|net|org|io|cn|ru|info|biz|us|uk|co|dev|cloud|local|lan)\b", "md5": r"\b[a-fA-F0-9]{32}\b", "sha1": r"\b[a-fA-F0-9]{40}\b", "sha256": r"\b[a-fA-F0-9]{64}\b", "jwt": r"eyJ[a-zA-Z0-9_\-]+\.[a-zA-Z0-9_\-]+\.[a-zA-Z0-9_\-]+", "api_key": r"(?i)(api[_-]?key|apikey|access[_-]?key|secret[_-]?key|client_secret)\s*[:=]\s*['\"]?([a-zA-Z0-9_\-]{12,})", "wifi_psk": r"(?i)(passphrase|psk|wpa_pass|wifi_pass|wifi_password|wireless_key|wpakey)\s*[:=]\s*['\"]?([^\s'\";&]{8,63})", "possible_password": r"(?i)(password|passwd|pwd|pass|admin_pass|root_pass|web_pass|wifi_pass|secret|token)\s*[:=]\s*['\"]?([^\s'\";&]{4,})", "possible_username": r"(?i)(username|user|login|admin_user|root_user)\s*[:=]\s*['\"]?([^\s'\";&]{3,})", "mqtt_credential": r"(?i)(mqtt_(?:user|username|pass|password)|mqttUser|mqttPass)\s*[:=]\s*['\"]?([^\s'\";&]{3,})", "basic_auth": r"(?i)authorization:\s*basic\s+([a-zA-Z0-9+/=]{8,})", } PRIVATE_KEY_MARKERS = { "-----BEGIN RSA PRIVATE KEY-----", "-----BEGIN DSA PRIVATE KEY-----", "-----BEGIN EC PRIVATE KEY-----", "-----BEGIN OPENSSH PRIVATE KEY-----", "-----BEGIN PRIVATE KEY-----" } CERT_MARKERS = {"-----BEGIN CERTIFICATE-----"} COMPONENT_PATTERNS = { "busybox": [ r"BusyBox v([\w.\-]+)", r"busybox\s+v([\w.\-]+)" ], "linux_kernel": [ r"Linux version ([\w.\-]+)", r"kernel version[:= ]+([\w.\-]+)" ], "openssl": [ r"OpenSSL\s+([0-9][\w.\-]+[a-z]?)", r"openssl-([0-9][\w.\-]+)" ], "dropbear": [ r"Dropbear sshd v?([0-9][\w.\-]+)", r"dropbear[_ -]?([0-9]{4}\.[0-9]{2})" ], "dnsmasq": [ r"dnsmasq-?([0-9][\w.\-]+)", r"dnsmasq version ([0-9][\w.\-]+)" ], "uClibc": [ r"uClibc-?([0-9][\w.\-]+)" ], "musl": [ r"musl-?([0-9][\w.\-]+)" ], "glibc": [ r"GNU C Library.*?release version ([0-9][\w.\-]+)", r"GLIBC_([0-9.]+)" ], "lighttpd": [ r"lighttpd/([0-9][\w.\-]+)" ], "boa": [ r"Boa/([0-9][\w.\-]+)" ], "uhttpd": [ r"uhttpd[-/ ]([0-9][\w.\-]+)" ] } CVE_HINTS = { "busybox": { "note": "BusyBox is frequently old in embedded firmware. Verify exact version against NVD/vendor advisories.", "keywords": ["busybox", "ash", "udhcp", "telnetd"] }, "openssl": { "note": "OpenSSL version detected. Check for old TLS/crypto CVEs and weak certificate/key usage.", "keywords": ["openssl", "libssl", "libcrypto"] }, "dropbear": { "note": "Dropbear SSH detected. Check version against Dropbear security advisories.", "keywords": ["dropbear", "sshd"] }, "dnsmasq": { "note": "dnsmasq detected. Check version for DNS/DHCP vulnerabilities.", "keywords": ["dnsmasq"] }, "boa": { "note": "Boa web server is commonly outdated in IoT firmware.", "keywords": ["boa"] } } IMPORTANT_SECTIONS = [ ("Credential Findings", "credential_findings", "credential_findings.txt"), ("Possible Passwords", "possible_password", "possible_passwords.txt"), ("Wi-Fi PSKs", "wifi_psk", "wifi_psks.txt"), ("Possible Usernames", "possible_username", "possible_usernames.txt"), ("API Keys / Secrets", "api_key", "api_keys.txt"), ("MQTT Credentials", "mqtt_credential", "mqtt_credentials.txt"), ("JWT Tokens", "jwt", "jwt_tokens.txt"), ("Private Keys", "private_keys", "private_keys.txt"), ("Certificates", "certificates", "certificates.txt"), ("IP Addresses", "ipv4", "ip_addresses.txt"), ("MAC Addresses", "mac_address", "mac_addresses.txt"), ("URLs", "url", "urls.txt"), ("Domains", "domain", "domains.txt"), ("Emails", "email", "emails.txt"), ("Component Versions", "components", "components.txt"), ("Version Strings", "version_strings", "version_strings.txt"), ("Firmware Identity", "firmware_identity", "firmware_identity.txt"), ("Startup Scripts", "startup_scripts", "startup_scripts.txt"), ("Interesting Files", "interesting_files", "interesting_files.txt"), ("Interesting Binaries", "interesting_binaries", "interesting_binaries.txt"), ("ELF Binaries", "elf_binaries", "elf_binaries.txt"), ("Architecture Summary", "architecture_summary", "architecture_summary.txt"), ("Web Files", "web_files", "web_files.txt"), ("Web Routes", "web_routes", "web_routes.txt"), ("Web Endpoints", "web_endpoints", "web_endpoints.txt"), ("Cron Jobs", "cron_jobs", "cron_jobs.txt"), ("Users / Groups", "users_groups", "users_groups.txt"), ("SSH Related Files", "ssh_related", "ssh_related.txt"), ("Suspicious Keywords", "suspicious_keywords", "suspicious_keywords.txt"), ("Malware IOC Matches", "malware_iocs", "malware_iocs.txt"), ("High Entropy Files", "high_entropy_files", "high_entropy_files.txt"), ("Largest Files", "largest_files", "largest_files.txt"), ("BusyBox Findings", "busybox", "busybox.txt"), ("CVE References", "cve_references", "cve_references.txt"), ("CVE Hints", "cve_hints", "cve_hints.txt"), ("YARA Matches", "yara_matches", "yara_matches.txt"), ("Plugin Findings", "plugin_findings", "plugin_findings.txt"), ("Skipped Large Files", "skipped_large_files", "skipped_large_files.txt"), ("Tool Warnings", "tool_warnings", "tool_warnings.txt"), ("Scan Errors", "scan_errors", "scan_errors.txt"), ] # ============================================================================= # BASIC HELPERS # ============================================================================= def add(category, data): """Thread-safe de-duplicated report append.""" with REPORT_LOCK: if data not in REPORT[category]: REPORT[category].append(data) def safe_rel(path, root): try: return str(Path(path).resolve().relative_to(Path(root).resolve())) except Exception: return str(path) def now_iso(): return datetime.now().isoformat(timespec="seconds") def check_dependencies(yara_requested=False, extract_requested=False): dependencies = ["strings"] if yara_requested: dependencies.append("yara") if extract_requested: dependencies.append("binwalk") for tool in dependencies: if shutil.which(tool) is None: add("tool_warnings", { "tool": tool, "warning": f"'{tool}' was not found in PATH. Some features may be limited." }) print(f"[!] Warning: '{tool}' not found. Some features may be limited.") def sha256_file(path): try: h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() except Exception: return None def entropy_file(path, max_bytes=1024 * 1024): try: with open(path, "rb") as f: data = f.read(max_bytes) if not data: return 0.0 counts = Counter(data) length = len(data) entropy = -sum((count / length) * math.log2(count / length) for count in counts.values()) return round(entropy, 4) except Exception: return None def is_text_file(path): if path.suffix.lower() in TEXT_EXTENSIONS: return True try: with open(path, "rb") as f: chunk = f.read(4096) if not chunk: return False return chunk.count(b"\x00") / len(chunk) < 0.08 except Exception: return False def read_text(path, max_chars=None): try: text = path.read_text(errors="ignore") if max_chars: return text[:max_chars] return text except Exception: return "" def is_elf(path): try: with open(path, "rb") as f: return f.read(4) == b"\x7fELF" except Exception: return False def elf_info(path): try: with open(path, "rb") as f: header = f.read(20) if len(header) < 20: return {} elf_class = header[4] endian = header[5] byte_order = "little" if endian == 1 else "big" if endian == 2 else "little" machine = int.from_bytes(header[18:20], byte_order) arch_map = { 3: "x86", 8: "MIPS", 20: "PowerPC", 40: "ARM", 62: "x86_64", 183: "AArch64", 243: "RISC-V" } return { "class": "32-bit" if elf_class == 1 else "64-bit" if elf_class == 2 else "unknown", "endian": "little" if endian == 1 else "big" if endian == 2 else "unknown", "machine": arch_map.get(machine, f"unknown-{machine}") } except Exception: return {} def run_strings(path, limit=1200): try: result = subprocess.run( ["strings", str(path)], capture_output=True, text=True, timeout=20, errors="ignore" ) return result.stdout.splitlines()[:limit] except Exception: return [] def run_file_cmd(path): if shutil.which("file") is None: return None try: result = subprocess.run( ["file", "-b", str(path)], capture_output=True, text=True, timeout=5, errors="ignore" ) return result.stdout.strip() except Exception: return None # ============================================================================= # EXTRACTION # ============================================================================= def extract_with_binwalk(firmware_image, output_base): firmware_image = Path(firmware_image).resolve() extract_dir = Path(output_base).resolve() / f"binwalk_extract_{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}" extract_dir.mkdir(parents=True, exist_ok=True) if shutil.which("binwalk") is None: raise RuntimeError("binwalk not found. Install it or scan an already extracted rootfs directory.") print(f"[+] Extracting with binwalk into: {extract_dir}") try: subprocess.run( ["binwalk", "-Me", str(firmware_image), "--directory", str(extract_dir)], check=False, timeout=1800 ) except subprocess.TimeoutExpired: add("tool_warnings", {"tool": "binwalk", "warning": "binwalk extraction timed out"}) except Exception as e: add("tool_warnings", {"tool": "binwalk", "warning": f"binwalk extraction failed: {e}"}) candidates = find_rootfs_candidates(extract_dir) if candidates: print("[+] Possible rootfs candidates:") for idx, cand in enumerate(candidates[:10], 1): print(f" {idx}. {cand}") return candidates[0] print("[!] No obvious rootfs found. Scanning extraction directory instead.") return extract_dir def find_rootfs_candidates(base_dir): base_dir = Path(base_dir) candidates = [] for dirpath, dirnames, filenames in os.walk(base_dir): d = Path(dirpath) names = set(filenames) | set(dirnames) score = 0 if "etc" in names: score += 3 if "bin" in names: score += 2 if "sbin" in names: score += 2 if "www" in names or "web" in names or "htdocs" in names: score += 2 if "passwd" in names or "shadow" in names: score += 2 if "init.d" in names: score += 2 if score >= 5: candidates.append((score, d)) candidates.sort(key=lambda x: x[0], reverse=True) return [c[1] for c in candidates] # ============================================================================= # SCANNERS # ============================================================================= def extract_version_strings(text): patterns = [ r"(?i)(version|build|release|fw_ver|firmware|software|sw_ver|hardware|hw_ver|model|vendor|product)\s*[:=]\s*['\"]?([\w.\-/ ]{3,80})", r"(?i)(busybox\s+v\d{1,3}\.\d{1,3}(?:\.\d{1,5})?)", r"(?i)(linux\s+version\s+[\w.\-]+)", r"\bV(\d{1,3}\.\d{1,3}\.\d{1,5})\b", r"\b(\d{4}-\d{2}-\d{2})\b", ] versions = set() for pattern in patterns: for match in re.findall(pattern, text): if isinstance(match, tuple): value = match[-1] else: value = match value = " ".join(value.strip().split()) if 3 <= len(value) <= 120: versions.add(value) return sorted(versions)[:40] def detect_components(text, rel_path): for component, patterns in COMPONENT_PATTERNS.items(): for pattern in patterns: for match in re.findall(pattern, text, re.I): version = match[-1] if isinstance(match, tuple) else match version = str(version).strip() if version: add("components", { "component": component, "version": version, "file": rel_path }) if component in CVE_HINTS: add("cve_hints", { "component": component, "version": version, "file": rel_path, "note": CVE_HINTS[component]["note"] }) def detect_malware_iocs(text, rel_path): low = text.lower() for family, indicators in MALWARE_IOC_KEYWORDS.items(): hits = [] for indicator in indicators: if indicator.lower() in low: hits.append(indicator) if hits: add("malware_iocs", { "family_or_category": family, "file": rel_path, "indicators": sorted(set(hits)), "confidence": "medium" if len(hits) == 1 else "high" }) def scan_text_content(path, content, rel_path): for name, pattern in REGEX_PATTERNS.items(): for match in re.findall(pattern, content): if isinstance(match, tuple): finding = {"file": rel_path, "key": match[0], "value": match[1]} value_for_validation = match[1] else: finding = {"file": rel_path, "value": match} value_for_validation = match if name == "ipv4": parts = value_for_validation.split(".") if not all(p.isdigit() and 0 <= int(p) <= 255 for p in parts): continue add(name, finding) if name in {"possible_password", "wifi_psk", "api_key", "mqtt_credential"}: add_credential_finding(name, finding) for marker in PRIVATE_KEY_MARKERS: if marker in content: add("private_keys", {"file": rel_path, "marker": marker}) add("credential_findings", { "file": rel_path, "type": "private_key", "value": marker, "confidence": "high", "severity": "critical", "reason": "Private key material marker found" }) for marker in CERT_MARKERS: if marker in content: add("certificates", {"file": rel_path, "marker": marker}) lowered = content.lower() for keyword in SUSPICIOUS_KEYWORDS: if keyword.lower() in lowered: add("suspicious_keywords", {"file": rel_path, "keyword": keyword}) versions = extract_version_strings(content) if versions: add("version_strings", {"file": rel_path, "versions": versions}) detect_components(content, rel_path) detect_malware_iocs(content, rel_path) def add_credential_finding(kind, finding): value = str(finding.get("value", "")) key = str(finding.get("key", "")) file = finding.get("file", "") confidence = "medium" severity = "medium" reasons = [] if kind in {"api_key", "wifi_psk"}: confidence = "high" severity = "high" reasons.append(f"{kind} pattern matched") if key.lower() in {"password", "passwd", "pwd", "admin_pass", "root_pass", "secret", "token"}: confidence = "high" severity = "high" reasons.append("credential-like key name") if value.lower() in {"admin", "root", "password", "123456", "12345678", "admin123", "root123"}: confidence = "high" severity = "high" reasons.append("common/default credential value") if any(x in file.lower() for x in ["shadow", "passwd", "default", "config", "nvram", "wpa"]): reasons.append("sensitive config path") if confidence == "medium": confidence = "high" add("credential_findings", { "file": file, "type": kind, "key": key, "value": value, "confidence": confidence, "severity": severity, "reason": "; ".join(reasons) if reasons else "credential-like pattern matched" }) def classify_path(path, root, size, rel_path): lower = str(path).lower() name = path.name.lower() info = {"file": rel_path, "sha256": sha256_file(path), "size": size} if name in INTERESTING_NAMES: add("interesting_files", info) if any(p in lower for p in ["/etc/init.d", "/etc/rc", "/etc/inittab", "/etc/services", "/lib/systemd"]): add("startup_scripts", info) if any(p in lower for p in ["/www", "/web", "/htdocs", "/cgi-bin", "/var/www"]): add("web_files", info) if any(x in name for x in ["config", ".conf", ".cfg", ".ini", ".json", ".xml", ".yaml", ".yml"]): add("config_files", info) if "cron" in lower or "crontab" in lower: add("cron_jobs", info) if name in {"passwd", "shadow", "group"}: add("users_groups", info) if any(x in lower for x in ["ssh", "dropbear", "authorized_keys", "host_key"]): add("ssh_related", info) if name in EXECUTABLE_NAMES: add("interesting_binaries", {**info, "reason": "Common embedded Linux service or utility"}) def scan_busybox(path, text, rel_path): if "busybox" not in str(path).lower() and "busybox" not in text.lower(): return banner = re.search(r"BusyBox v[\w.\-]+", text) applet_hits = [] applet_keywords = ["telnet", "wget", "tftp", "httpd", "ash", "sh", "nc", "ftpget", "ftpput"] for line in text.splitlines(): low = line.lower() if any(x in low for x in applet_keywords) and len(line.strip()) < 250: applet_hits.append(line.strip()) risky_applets = [] for app in applet_keywords: if app in text.lower(): risky_applets.append(app) add("busybox", { "file": rel_path, "banner": banner.group(0) if banner else None, "risky_applet_hints": sorted(set(risky_applets)), "possible_applets_or_strings": sorted(set(applet_hits))[:100] }) def scan_web_endpoints(path, content, rel_path): lower_path = str(path).lower() if not any(x in lower_path for x in ["/www", "/web", "/htdocs", "/cgi-bin", ".html", ".js", ".php", ".cgi", ".asp"]): return endpoints = re.findall(r"[\"'](/[^\"'\s<>]{2,})[\"']", content) forms = re.findall(r"(?i)]+action=[\"']?([^\"'>\s]+)", content) scripts = re.findall(r"(?i)]+src=[\"']?([^\"'>\s]+)", content) ajax = re.findall(r"(?i)(?:url|href|src)\s*[:=]\s*[\"']([^\"']+)[\"']", content) cgi = re.findall(r"[\w./-]+\.cgi(?:\?[^\"'\s<>]*)?", content) admin_hits = [] for route in set(endpoints + forms + scripts + ajax + cgi): if any(k in route.lower() for k in ["admin", "login", "password", "upgrade", "firmware", "reboot", "debug", "shell", "system", "config"]): admin_hits.append(route) if endpoints or forms or scripts or ajax or cgi: data = { "file": rel_path, "endpoints": sorted(set(endpoints))[:150], "forms": sorted(set(forms))[:80], "scripts": sorted(set(scripts))[:80], "ajax_or_refs": sorted(set(ajax))[:120], "cgi_refs": sorted(set(cgi))[:120], "admin_like_routes": sorted(set(admin_hits))[:120] } add("web_endpoints", data) for route in sorted(set(endpoints + forms + scripts + ajax + cgi + admin_hits)): if len(route) < 300: add("web_routes", { "route": route, "file": rel_path, "admin_like": route in admin_hits }) def scan_possible_cves(content, rel_path): for cve in re.findall(r"CVE-\d{4}-\d{4,7}", content, re.I): add("cve_references", {"file": rel_path, "cve": cve.upper()}) def yara_scan_file(path, rel_path, yara_rule_path): if not yara_rule_path: return try: result = subprocess.run( ["yara", "-r", str(yara_rule_path), str(path)], capture_output=True, text=True, timeout=25 ) if result.stdout.strip(): add("yara_matches", { "file": rel_path, "matches": result.stdout.strip().splitlines() }) except FileNotFoundError: add("tool_warnings", {"tool": "yara", "warning": "YARA is not installed or not in PATH"}) except Exception as e: add("scan_errors", {"file": rel_path, "error": f"YARA scan failed: {e}"}) def scan_file(path, root, yara_rules=None, quick=False): try: if not path.is_file(): return size = path.stat().st_size rel_path = safe_rel(path, root) classify_path(path, root, size, rel_path) if quick and size > 10_000_000: add("skipped_large_files", { "file": rel_path, "size": size, "reason": "Skipped because --quick mode is enabled" }) return ent = entropy_file(path) if ent is not None and ent >= 7.5 and size > 1024: add("high_entropy_files", { "file": rel_path, "entropy": ent, "size": size, "sha256": sha256_file(path), "note": "May indicate compression, encryption, packed data, or a binary blob" }) if is_elf(path): info = elf_info(path) file_desc = run_file_cmd(path) add("elf_binaries", { "file": rel_path, "sha256": sha256_file(path), "size": size, "elf": info, "file_cmd": file_desc }) strings_text = "\n".join(run_strings(path)) scan_text_content(path, strings_text, rel_path) scan_busybox(path, strings_text, rel_path) scan_possible_cves(strings_text, rel_path) if is_text_file(path): content = read_text(path) scan_text_content(path, content, rel_path) scan_web_endpoints(path, content, rel_path) scan_possible_cves(content, rel_path) scan_busybox(path, content, rel_path) if yara_rules: yara_scan_file(path, rel_path, yara_rules) except Exception as e: add("scan_errors", {"file": str(path), "error": str(e)}) # ============================================================================= # POST PROCESSING # ============================================================================= def detect_firmware_identity(root): identity_files = [ "etc/os-release", "etc/openwrt_release", "etc/openwrt_version", "etc/version", "etc/banner", "etc/issue", "version", "release" ] for rel in identity_files: path = Path(root) / rel if path.exists() and path.is_file(): text = read_text(path, max_chars=5000) if text: add("firmware_identity", { "file": rel, "content_preview": text[:1000] }) # Heuristic identity from config/version strings vendor_model_keywords = ["model", "vendor", "product", "device", "board", "firmware", "version"] for item in REPORT.get("version_strings", [])[:200]: versions = item.get("versions", []) hits = [v for v in versions if any(k in v.lower() for k in vendor_model_keywords)] if hits: add("firmware_identity", { "file": item.get("file"), "identity_hints": hits[:20] }) def summarize_architecture(): machines = Counter() endians = Counter() classes = Counter() for item in REPORT.get("elf_binaries", []): elf = item.get("elf", {}) if elf.get("machine"): machines[elf["machine"]] += 1 if elf.get("endian"): endians[elf["endian"]] += 1 if elf.get("class"): classes[elf["class"]] += 1 if machines or endians or classes: add("architecture_summary", { "machines": dict(machines.most_common()), "endian": dict(endians.most_common()), "class": dict(classes.most_common()) }) def summarize_largest_files(root, limit=30): files = [] for dirpath, _, filenames in os.walk(root): for filename in filenames: p = Path(dirpath) / filename try: if p.is_file(): files.append((p.stat().st_size, safe_rel(p, root), sha256_file(p))) except Exception: pass for size, rel, digest in sorted(files, reverse=True)[:limit]: add("largest_files", {"file": rel, "size": size, "sha256": digest}) def run_plugins(plugin_dir, root): if not plugin_dir: return plugin_dir = Path(plugin_dir) if not plugin_dir.exists(): add("tool_warnings", {"tool": "plugins", "warning": f"Plugin directory not found: {plugin_dir}"}) return for plugin_file in plugin_dir.glob("*.py"): try: spec = importlib.util.spec_from_file_location(plugin_file.stem, plugin_file) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) if hasattr(module, "scan"): result = module.scan(str(root)) add("plugin_findings", { "plugin": plugin_file.name, "result": result }) else: add("tool_warnings", { "tool": "plugins", "warning": f"{plugin_file.name} has no scan(root) function" }) except Exception as e: add("scan_errors", { "file": str(plugin_file), "error": f"Plugin failed: {e}" }) def score_findings(): weights = { "private_keys": 35, "credential_findings": 20, "possible_password": 15, "wifi_psk": 15, "api_key": 15, "jwt": 15, "users_groups": 12, "yara_matches": 35, "malware_iocs": 30, "interesting_binaries": 8, "suspicious_keywords": 6, "web_routes": 5, "web_endpoints": 5, "ssh_related": 5, "startup_scripts": 5, "cve_references": 5, "cve_hints": 3, } score = 0 reasons = [] for category, weight in weights.items(): count = len(REPORT.get(category, [])) if count: added = min(count * weight, weight * 6) score += added reasons.append(f"{category}: {count} finding(s), +{added}") critical = any(x.get("severity") == "critical" for x in REPORT.get("credential_findings", []) if isinstance(x, dict)) malware = len(REPORT.get("malware_iocs", [])) > 0 if score >= 180 or critical or malware: level = "CRITICAL" if critical or malware else "HIGH" elif score >= 100: level = "HIGH" elif score >= 50: level = "MEDIUM" elif score > 0: level = "LOW" else: level = "NONE" REPORT["risk_summary"] = [{ "score": score, "level": level, "reasons": reasons, "notes": [ "Risk score is a triage aid, not proof of compromise.", "Manually validate credentials, CVEs, and malware indicators." ] }] # ============================================================================= # REPORTING # ============================================================================= def write_list_file(output_dir, filename, title, items): path = output_dir / filename with open(path, "w", encoding="utf-8") as f: f.write(title + "\n") f.write("=" * 100 + "\n\n") if not items: f.write("No findings.\n") return path for item in items: if isinstance(item, dict): for k, v in item.items(): f.write(f"{k}: {v}\n") f.write("-" * 100 + "\n") else: f.write(str(item) + "\n") return path def write_csv(output_dir): csv_file = output_dir / "findings.csv" rows = [] for title, key, _filename in IMPORTANT_SECTIONS: for item in REPORT.get(key, []): if isinstance(item, dict): rows.append({ "category": key, "title": title, "file": item.get("file", ""), "value": item.get("value", item.get("route", item.get("component", ""))), "severity": item.get("severity", ""), "confidence": item.get("confidence", ""), "details": json.dumps(item, ensure_ascii=False) }) else: rows.append({ "category": key, "title": title, "file": "", "value": str(item), "severity": "", "confidence": "", "details": str(item) }) with open(csv_file, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=["category", "title", "file", "value", "severity", "confidence", "details"]) writer.writeheader() writer.writerows(rows) return csv_file def write_markdown(output_dir): md_file = output_dir / "firmware_report.md" risk = REPORT.get("risk_summary", [{}])[0] totals = REPORT.get("scan_totals", [{}])[0] meta = REPORT.get("scan_metadata", [{}])[0] with open(md_file, "w", encoding="utf-8") as f: f.write(f"# Firmware Hunter Pro v{VERSION} Report\n\n") f.write("## Scan Info\n\n") f.write(f"- Firmware root: `{meta.get('root', 'Unknown')}`\n") f.write(f"- Started: `{meta.get('started', 'Unknown')}`\n") f.write(f"- Finished: `{meta.get('finished', 'Unknown')}`\n") f.write(f"- Total files: `{totals.get('total_files', 0)}`\n") f.write(f"- Total size: `{totals.get('total_size_bytes', 0)}` bytes\n") f.write(f"- Quick mode: `{meta.get('quick_mode', False)}`\n\n") f.write("## Risk Summary\n\n") f.write(f"- Level: **{risk.get('level', 'UNKNOWN')}**\n") f.write(f"- Score: **{risk.get('score', 0)}**\n\n") f.write("## Findings Overview\n\n") f.write("| Finding | Count |\n|---|---:|\n") for title, key, _ in IMPORTANT_SECTIONS: count = len(REPORT.get(key, [])) if count: f.write(f"| {title} | {count} |\n") f.write("\n## Recommended Next Steps\n\n") for step in recommended_steps(): f.write(f"- {step}\n") return md_file def write_html_report(output_dir): html_file = output_dir / "firmware_report.html" risk = REPORT.get("risk_summary", [{}])[0] totals = REPORT.get("scan_totals", [{}])[0] meta = REPORT.get("scan_metadata", [{}])[0] severity_color = { "CRITICAL": "#ef4444", "HIGH": "#f97316", "MEDIUM": "#eab308", "LOW": "#22c55e", "NONE": "#94a3b8" }.get(risk.get("level", "UNKNOWN"), "#94a3b8") with open(html_file, "w", encoding="utf-8") as f: f.write(f""" Firmware Hunter Pro Report

Firmware Hunter Pro v{VERSION}

Offline firmware triage report

Risk{html.escape(str(risk.get('level', 'UNKNOWN')))}
Risk Score{html.escape(str(risk.get('score', 0)))}
Total Files{html.escape(str(totals.get('total_files', 0)))}
Total Size{html.escape(str(totals.get('total_size_bytes', 0)))} bytes

Scan Info

Firmware Root: {html.escape(str(meta.get('root', 'Unknown')))}

Started: {html.escape(str(meta.get('started', 'Unknown')))}

Finished: {html.escape(str(meta.get('finished', 'Unknown')))}

Quick Mode: {html.escape(str(meta.get('quick_mode', False)))}

Worker Jobs: {html.escape(str(meta.get('jobs', 'Unknown')))}

Findings Overview

""") for title, key, filename in IMPORTANT_SECTIONS: count = len(REPORT.get(key, [])) if count: f.write(f"

{html.escape(title)}: {count} — {html.escape(filename)}

\n") f.write("""

Recommended Next Steps

    """) for step in recommended_steps(): f.write(f"
  • {html.escape(step)}
    • \n") f.write("""

Detailed Findings

""") for title, key, _filename in IMPORTANT_SECTIONS: items = REPORT.get(key, []) if not items: continue f.write(f"
{html.escape(title)} ({len(items)})\n") f.write("
")
            f.write(html.escape(json.dumps(items, indent=2, ensure_ascii=False)))
            f.write("
\n") f.write("
") return html_file def recommended_steps(): return [ "Review credential_findings.txt, possible_passwords.txt, and wifi_psks.txt for hardcoded credentials.", "Review startup_scripts.txt to understand what starts at boot.", "Review web_routes.txt and web_endpoints.txt for login panels, CGI handlers, firmware update routes, and debug routes.", "Review interesting_binaries.txt for telnet, dropbear, busybox, wget, tftp, netcat, and web servers.", "Review private_keys.txt and certificates.txt for exposed secrets or reused keys.", "Review components.txt and cve_hints.txt, then verify versions against authoritative vulnerability databases.", "Review malware_iocs.txt and yara_matches.txt, then manually validate before making conclusions.", "Use full_report.json or findings.csv for automation, diffing, or importing into other tools." ] def write_reports(base_output): timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") output_dir = Path(base_output) / f"scan_{timestamp}" output_dir.mkdir(parents=True, exist_ok=True) json_file = output_dir / "full_report.json" main_report = output_dir / "firmware_report.txt" summary_file = output_dir / "summary.txt" normal_report = {k: v for k, v in REPORT.items()} with open(json_file, "w", encoding="utf-8") as f: json.dump(normal_report, f, indent=4, ensure_ascii=False) risk = REPORT.get("risk_summary", [{}])[0] totals = REPORT.get("scan_totals", [{}])[0] meta = REPORT.get("scan_metadata", [{}])[0] for title, key, filename in IMPORTANT_SECTIONS: items = REPORT.get(key, []) if items: write_list_file(output_dir, filename, title, items) csv_file = write_csv(output_dir) md_file = write_markdown(output_dir) html_file = write_html_report(output_dir) with open(summary_file, "w", encoding="utf-8") as f: f.write("Firmware Hunter Pro Summary\n") f.write("=" * 100 + "\n\n") f.write(f"Scan Time: {timestamp}\n") f.write(f"Firmware Root: {meta.get('root', 'Unknown')}\n") f.write(f"Quick Mode: {meta.get('quick_mode', False)}\n") f.write(f"Jobs: {meta.get('jobs', 'Unknown')}\n") f.write(f"Total Files: {totals.get('total_files', 0)}\n") f.write(f"Total Size: {totals.get('total_size_bytes', 0)} bytes\n\n") f.write("Risk Summary\n") f.write("-" * 100 + "\n") f.write(f"Risk Level: {risk.get('level', 'UNKNOWN')}\n") f.write(f"Risk Score: {risk.get('score', 0)}\n\n") f.write("Finding Counts\n") f.write("-" * 100 + "\n") for title, key, _ in IMPORTANT_SECTIONS: count = len(REPORT.get(key, [])) if count: f.write(f"{title:<35} {count}\n") with open(main_report, "w", encoding="utf-8") as f: f.write("Firmware Hunter Pro Report\n") f.write("=" * 100 + "\n") f.write("Offline firmware triage report\n") f.write("=" * 100 + "\n\n") f.write("[SCAN INFO]\n") f.write(f"Firmware Root : {meta.get('root', 'Unknown')}\n") f.write(f"Started : {meta.get('started', 'Unknown')}\n") f.write(f"Finished : {meta.get('finished', 'Unknown')}\n") f.write(f"Quick Mode : {meta.get('quick_mode', False)}\n") f.write(f"Worker Jobs : {meta.get('jobs', 'Unknown')}\n") f.write(f"Total Files : {totals.get('total_files', 0)}\n") f.write(f"Total Size : {totals.get('total_size_bytes', 0)} bytes\n\n") f.write("[RISK SUMMARY]\n") f.write(f"Level : {risk.get('level', 'UNKNOWN')}\n") f.write(f"Score : {risk.get('score', 0)}\n\n") if risk.get("reasons"): f.write("Reasons:\n") for reason in risk.get("reasons", []): f.write(f" - {reason}\n") f.write("\n") f.write("[FINDINGS OVERVIEW]\n") for title, key, filename in IMPORTANT_SECTIONS: count = len(REPORT.get(key, [])) if count: f.write(f"[+] {title:<35} {count:<5} -> {filename}\n") else: f.write(f"[-] {title:<35} 0\n") f.write("\n[RECOMMENDED NEXT STEPS]\n") f.write("-" * 100 + "\n") for idx, step in enumerate(recommended_steps(), 1): f.write(f"{idx}. {step}\n") f.write("\n[GENERATED FILES]\n") for file in sorted(output_dir.iterdir()): f.write(f"- {file.name}\n") print() print("=" * 100) print(" Firmware Hunter Pro Scan Complete") print("=" * 100) print(f" Output Folder : {output_dir}") print(f" Main Report : {main_report}") print(f" Summary : {summary_file}") print(f" HTML Report : {html_file}") print(f" Markdown : {md_file}") print(f" CSV Findings : {csv_file}") print(f" JSON Report : {json_file}") print("=" * 100) # ============================================================================= # MAIN SCAN # ============================================================================= def scan_firmware(root, yara_rules=None, quick=False, jobs=8, plugins=None): root = Path(root).resolve() if not root.exists(): raise FileNotFoundError(f"Firmware path does not exist: {root}") if not root.is_dir(): raise NotADirectoryError(f"Firmware path should be an extracted directory/rootfs: {root}") files = [] total_size = 0 for dirpath, _, filenames in os.walk(root): for filename in filenames: path = Path(dirpath) / filename if path.is_file(): files.append(path) try: total_size += path.stat().st_size except Exception: pass REPORT["scan_metadata"] = [{ "tool": "Firmware Hunter Pro", "version": VERSION, "root": str(root), "started": now_iso(), "quick_mode": quick, "jobs": jobs, "yara_rules": str(yara_rules) if yara_rules else None, "plugins": str(plugins) if plugins else None }] print(f"[+] Files discovered: {len(files)}") print(f"[+] Total size: {total_size} bytes") print(f"[+] Worker threads: {jobs}") completed = 0 start_time = time.time() with ThreadPoolExecutor(max_workers=max(1, jobs)) as executor: futures = [executor.submit(scan_file, f, root, yara_rules, quick) for f in files] for future in as_completed(futures): completed += 1 try: future.result() except Exception as e: add("scan_errors", {"file": "unknown", "error": str(e)}) if completed % 500 == 0: elapsed = max(time.time() - start_time, 1) rate = completed / elapsed print(f"[+] Progress: {completed}/{len(files)} files scanned ({rate:.1f} files/sec)") detect_firmware_identity(root) summarize_architecture() summarize_largest_files(root) run_plugins(plugins, root) score_findings() REPORT["scan_totals"] = [{ "total_files": len(files), "total_size_bytes": total_size, "elapsed_seconds": round(time.time() - start_time, 2) }] REPORT["scan_metadata"][0]["finished"] = now_iso() def main(): parser = argparse.ArgumentParser( description="Firmware Hunter Pro v4.0 - advanced offline firmware triage scanner" ) parser.add_argument("target", help="Path to extracted firmware/rootfs directory, or firmware image with --extract") parser.add_argument("-o", "--output", default="firmware_hunter_output", help="Base output directory") parser.add_argument("--extract", action="store_true", help="Use binwalk to extract target first, then scan extraction") parser.add_argument("--yara", help="Optional path to YARA rule file or directory") parser.add_argument("--quick", action="store_true", help="Quick mode: skip files larger than 10 MB") parser.add_argument("-j", "--jobs", type=int, default=8, help="Number of worker threads, default: 8") parser.add_argument("--plugins", help="Optional plugins directory. Each plugin should expose scan(root).") args = parser.parse_args() print(f"[+] Firmware Hunter Pro v{VERSION}") print("[+] Safe mode: this tool does not execute firmware binaries.") check_dependencies(yara_requested=bool(args.yara), extract_requested=args.extract) target = Path(args.target).resolve() if args.extract: scan_root = extract_with_binwalk(target, args.output) else: scan_root = target print(f"[+] Scanning: {scan_root}") scan_firmware( scan_root, yara_rules=args.yara, quick=args.quick, jobs=args.jobs, plugins=args.plugins ) write_reports(args.output) if __name__ == "__main__": main()