import subprocess import os import re import math import logging from IoTAnalyzer.models import BinwalkRaw, StringsRaw, RawDataset, Tool from IoTAnalyzer.db_create import ( create_binwalk_raw, create_strings_raw, create_analyzed_dataset, create_binwalk_analyzed, create_strings_analyzed, ) import json logger = logging.getLogger(__name__) def analyze_firmware_file(firmware_file, raw_dataset_id): """ Main function for firmware analysis using binwalk and strings. """ print(f"Start analyzing firmware: {firmware_file}") logger.info(f"Firmware-Analyse gestartet: {firmware_file} (RawDataset ID: {raw_dataset_id})") # Robust path resolution for the firmware file (multiple candidates) # repo_root = project root (3 levels up from analyze_firmware.py) repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..')) candidates = [ # preferred: central files directory in the repo root os.path.join(repo_root, 'files', firmware_file), # Fallback: frontend public folder (Next.js) os.path.join(repo_root, 'Frontend', 'iot-pre-tester-frontend', 'public', 'filesAttachment', firmware_file), # Relative (in case the working directory is already the repo root) os.path.join('files', firmware_file), os.path.join('Frontend', 'iot-pre-tester-frontend', 'public', 'filesAttachment', firmware_file), ] filepath = None for cand in candidates: if os.path.exists(cand): filepath = cand break if filepath is None: logger.error( "Firmware-Datei nicht gefunden. Geprüfte Pfade: " + ", ".join(candidates) ) print("ERROR: File not found. Checked paths:\n" + "\n".join(candidates)) return try: # 1. Run binwalk analysis print("Running binwalk analysis...") binwalk_results = run_binwalk_analysis(filepath) save_binwalk_to_database(binwalk_results, raw_dataset_id) # 2. Run strings extraction (EXTENDED with key-value pairing) print("Running strings extraction with key-value pairing...") strings_results = run_strings_extraction_with_pairing(filepath) save_strings_to_database(strings_results, raw_dataset_id) # 3. Create analyzed dataset for further analysis create_analyzed_dataset_for_firmware(raw_dataset_id, binwalk_results, strings_results) print("Firmware analysis completed successfully.") logger.info(f"Firmware-Analyse erfolgreich abgeschlossen: {firmware_file}") except Exception as e: logger.error(f"Fehler bei Firmware-Analyse: {e}") print(f"ERROR during firmware analysis: {e}") def run_binwalk_analysis(filepath): """ Runs binwalk on the firmware. """ results = [] try: cmd = ['binwalk', filepath] output = subprocess.check_output(cmd, stderr=subprocess.STDOUT, text=True) lines = output.split('\n') for line in lines: if line.startswith('DECIMAL') or line.startswith('----') or not line.strip(): continue parts = line.split(None, 2) if len(parts) >= 3: decimal = parts[0] hexadecimal = parts[1] description = parts[2] results.append({ 'offset': hexadecimal, 'description': description, 'file_path': None }) logger.info(f"Binwalk fand {len(results)} Signaturen") except subprocess.CalledProcessError as e: logger.error(f"Binwalk-Fehler: {e.output}") except FileNotFoundError: logger.error("Binwalk nicht installiert. Bitte installieren: sudo apt-get install binwalk") print("ERROR: binwalk not found. Install with: sudo apt-get install binwalk") return results def calculate_entropy(text): """Calculates the Shannon entropy of a string.""" if not text: return 0 entropy = 0 for x in range(256): p_x = float(text.count(chr(x))) / len(text) if p_x > 0: entropy += - p_x * math.log(p_x, 2) return entropy def looks_like_secret(text): """ Checks whether a string looks like a secret (high entropy, mixed case, digits). """ if len(text) < 16 or len(text) > 256: return False # Must contain letters AND digits if not (re.search(r'[a-zA-Z]', text) and re.search(r'[0-9]', text)): return False entropy = calculate_entropy(text) return entropy > 3.5 def is_blacklisted(text): """ Checks whether a string is on the blacklist. """ text = text.strip() # Too short if len(text) < 3: return True # Pure numbers if re.match(r'^\d+$', text): return True # Boolean / Null if text.lower() in ['true', 'false', 'null', 'none', 'undefined']: return True # Versions (standalone) if re.match(r'^v?\d+(\.\d+)+$', text): return True # Only special characters if re.match(r'^[^a-zA-Z0-9]+$', text): return True return False def run_strings_extraction_with_pairing(filepath, min_length=4): """ EXTENDED strings extraction with KEY-VALUE PAIRING and FILTERING. """ results = [] # CRITICAL KEYWORDS that expect a value in the next string CRITICAL_KEYS = { # WiFi Credentials 'wifi-ssid': 'ssid', 'ssid': 'ssid', 'network-name': 'ssid', 'ap-name': 'ssid', 'wifi_ssid': 'ssid', # WiFi Passwords 'wifi-pass': 'password', 'wifi-password': 'password', 'wifi_pass': 'password', 'wpa-psk': 'password', 'wpa_psk': 'password', 'pre-shared-key': 'password', 'network-password': 'password', 'password': 'password', 'passwd': 'password', # Location 'location': 'location', 'geo-location': 'location', 'coordinates': 'location', 'latitude': 'location', 'longitude': 'location', # Cloud Tokens 'cloud-token': 'cloud_token', 'cloud_token': 'cloud_token', 'auth-token': 'cloud_token', 'access-token': 'cloud_token', 'bearer-token': 'cloud_token', 'refresh-token': 'cloud_token', # API Keys 'api-key': 'api_key', 'api_key': 'api_key', 'apikey': 'api_key', 'secret-key': 'api_key', 'secret_key': 'api_key', 'access-key': 'api_key', 'private-key': 'api_key', 'auth': 'api_key', 'bearer': 'cloud_token', # Device Identifiers 'device-id': 'device_id', 'device_id': 'device_id', 'serial-num': 'serial', 'serial_num': 'serial', 'mac-address': 'mac', 'mac_address': 'mac', # Certificates 'cert-dclrn': 'certificate', 'certificate': 'certificate', } # Trigger keywords for context search (simplified) TRIGGER_KEYWORDS = set(CRITICAL_KEYS.keys()) try: # Extract strings with offset cmd = ['strings', '-a', '-t', 'x', '-n', str(min_length), filepath] output = subprocess.check_output(cmd, stderr=subprocess.STDOUT, text=True) lines = output.split('\n') # Statistics stats_total = 0 stats_kept = 0 stats_filtered = 0 i = 0 while i < len(lines): line = lines[i].strip() if not line: i += 1 continue # Parse: "offset string_value" parts = line.split(None, 1) if len(parts) < 2: i += 1 continue offset = parts[0] string_value = parts[1].strip() stats_total += 1 # 1. Blacklist Check if is_blacklisted(string_value): stats_filtered += 1 i += 1 continue # 2. Context / key-value check string_lower = string_value.lower().replace('-', '_').replace(' ', '_') found_key_type = None # Check whether the current string is a key for key in TRIGGER_KEYWORDS: if key.replace('-', '_') in string_lower: found_key_type = CRITICAL_KEYS.get(key) break if found_key_type: # KEY found! Store key and try to find value results.append({ 'string_value': f"[KEY] {string_value}", 'string_type': found_key_type }) stats_kept += 1 # Get NEXT string as value (look ahead up to 2 strings) lookahead_count = 0 j = i + 1 while j < len(lines) and lookahead_count < 2: next_line = lines[j].strip() if next_line: next_parts = next_line.split(None, 1) if len(next_parts) >= 2: next_value = next_parts[1].strip() # If the next value is NOT on the blacklist, we take it if not is_blacklisted(next_value): results.append({ 'string_value': f"[VALUE] {next_value}", 'string_type': found_key_type }) print(f" ✓ Found {found_key_type}: {string_value} → {next_value}") stats_kept += 1 j += 1 lookahead_count += 1 else: # No key -> check whether it is a secret or an interesting category string_type = categorize_string(string_value) is_secret = looks_like_secret(string_value) keep = False if string_type != 'generic': keep = True elif is_secret: string_type = 'potential_secret' keep = True if keep: results.append({ 'string_value': string_value, 'string_type': string_type }) stats_kept += 1 else: stats_filtered += 1 i += 1 logger.info(f"Strings Analyse: {stats_total} total, {stats_filtered} gefiltert, {stats_kept} behalten") print(f"Strings Analysis: Total {stats_total} | Filtered {stats_filtered} | Kept {stats_kept}") # Output statistics stats = {} for r in results: stats[r['string_type']] = stats.get(r['string_type'], 0) + 1 print("\nString-Kategorien gefunden:") for category, count in sorted(stats.items(), key=lambda x: x[1], reverse=True): print(f" {category}: {count}") except subprocess.CalledProcessError as e: logger.error(f"Strings-Fehler: {e.output}") except FileNotFoundError: logger.error("strings nicht installiert. Bitte installieren: sudo apt-get install binutils") print("ERROR: strings not found. Install with: sudo apt-get install binutils") return results def categorize_string(string_value): """ Categorizes a string based on its content. """ def save_binwalk_to_database(binwalk_results, raw_dataset_id): """ Stores binwalk results in the BinwalkRaw table. """ for result in binwalk_results: try: response = create_binwalk_raw( raw_dataset_id=raw_dataset_id, offset=result['offset'], description=result['description'], file_path=result.get('file_path') ) if hasattr(response, 'content'): response_data = json.loads(response.content.decode('utf-8')) if response_data['status'] != 'success': logger.warning(f"Binwalk-Eintrag konnte nicht gespeichert werden: {result}") except Exception as e: logger.error(f"Fehler beim Speichern von Binwalk-Daten: {e}") def save_strings_to_database(strings_results, raw_dataset_id): """ Stores ALL strings results in the StringsRaw table. NO LIMIT ANYMORE - all strings are stored! """ # Prioritize interesting strings for the ordering priority_order = ['password', 'ssid', 'cloud_token', 'api_key', 'location', 'credentials', 'certificate', 'email', 'url', 'ip', 'mac', 'path', 'device_id', 'serial', 'generic'] sorted_results = sorted( strings_results, key=lambda x: priority_order.index(x['string_type']) if x['string_type'] in priority_order else 999 ) saved_count = 0 for result in sorted_results: try: response = create_strings_raw( raw_dataset_id=raw_dataset_id, string_value=result['string_value'], string_type=result['string_type'] ) if hasattr(response, 'content'): response_data = json.loads(response.content.decode('utf-8')) if response_data['status'] == 'success': saved_count += 1 except Exception as e: logger.error(f"Fehler beim Speichern von Strings-Daten: {e}") logger.info(f"Strings gespeichert: {saved_count} von {len(strings_results)}") print(f"✓ Saved {saved_count} strings to database") def create_analyzed_dataset_for_firmware(raw_dataset_id, binwalk_results, strings_results): """ Creates an AnalyzedDataset and performs extended analysis. """ try: raw_dataset = RawDataset.objects.get(id=raw_dataset_id) file_id = raw_dataset.file.id # Create/get tool binwalk_tool = get_or_create_tool("binwalk", "2.3.4") strings_tool = get_or_create_tool("strings", "2.37") # Create AnalyzedDataset analyzed_response = create_analyzed_dataset(file_id) if hasattr(analyzed_response, 'content'): analyzed_data = json.loads(analyzed_response.content.decode('utf-8')) else: analyzed_data = analyzed_response analyzed_dataset_id = analyzed_data['data']['id'] # Binwalk analysis: identify important components analyze_binwalk_components(analyzed_dataset_id, binwalk_tool.id, binwalk_results) # Strings analysis: identify critical strings (EXTENDED) analyze_critical_strings(analyzed_dataset_id, strings_tool.id, strings_results) logger.info(f"Analyzed Dataset für Firmware erstellt: ID {analyzed_dataset_id}") except Exception as e: logger.error(f"Fehler beim Erstellen des Analyzed Datasets: {e}") def get_or_create_tool(tool_name, version): """ Helper function: gets an existing tool or creates a new one. """ try: tool = Tool.objects.get(tool_name=tool_name, version=version) logger.info(f"Tool gefunden: {tool_name} {version}") return tool except Tool.DoesNotExist: tool = Tool.objects.create( tool_name=tool_name, version=version ) logger.info(f"Tool erstellt: {tool_name} {version}") return tool def analyze_binwalk_components(analyzed_dataset_id, tool_id, binwalk_results): """ Analyzes binwalk results and stores important components. """ important_keywords = { 'Linux kernel': 'Linux Kernel', 'U-Boot': 'U-Boot Bootloader', 'filesystem': 'Filesystem', 'certificate': 'TLS Certificate', 'private key': 'Private Key', 'public key': 'Public Key', 'compressed': 'Compressed Data', 'encrypted': 'Encrypted Data', 'gzip': 'GZIP Archive', 'zlib': 'ZLIB Data', 'lzma': 'LZMA Archive', 'squashfs': 'SquashFS Filesystem', 'jffs2': 'JFFS2 Filesystem', 'cramfs': 'CramFS Filesystem', } for result in binwalk_results: description = result['description'].lower() for keyword, component_type in important_keywords.items(): if keyword.lower() in description: try: create_binwalk_analyzed( analyzed_dataset_id=analyzed_dataset_id, tool_id=tool_id, component_type=component_type, offset=result['offset'], size=None, entropy=None ) logger.info(f"Wichtige Komponente gefunden: {component_type} @ {result['offset']}") except Exception as e: logger.error(f"Fehler beim Speichern von Binwalk-Analyse: {e}") break def analyze_critical_strings(analyzed_dataset_id, tool_id, strings_results): """ EXTENDED analysis of critical strings with KEY-VALUE pairs. Severity levels: - critical: passwords, cloud tokens, API keys, SSID+password combos - high: SSIDs, device IDs, certificates - medium: URLs, email addresses, location - low: paths, MAC addresses, serial numbers """ # Extended severity mapping severity_map = { 'password': 'critical', 'cloud_token': 'critical', 'api_key': 'critical', 'ssid': 'high', 'device_id': 'high', 'credentials': 'high', 'certificate': 'high', 'location': 'medium', 'email': 'medium', 'url': 'medium', 'ip': 'medium', 'serial': 'low', 'mac': 'low', 'path': 'low', 'generic': 'info' } for result in strings_results: string_value = result['string_value'] string_type = result['string_type'] # Base severity from type severity = severity_map.get(string_type, 'info') category = string_type # Store ALL critical, high and medium strings if severity in ['critical', 'high', 'medium']: try: create_strings_analyzed( analyzed_dataset_id=analyzed_dataset_id, tool_id=tool_id, string_value=string_value, category=category, severity=severity ) except Exception as e: logger.error(f"Fehler beim Speichern von Strings-Analyse: {e}") def extract_with_binwalk(filepath, output_dir): """ Optional: runs binwalk -e to extract files. """ try: cmd = ['binwalk', '-e', '-C', output_dir, filepath] subprocess.run(cmd, check=True, capture_output=True, text=True) logger.info(f"Binwalk-Extraktion nach {output_dir} erfolgreich") return output_dir except subprocess.CalledProcessError as e: logger.error(f"Binwalk-Extraktion fehlgeschlagen: {e.stderr}") return None def categorize_string(string_value): """ Categorizes a string based on its content. """ # URLs if re.search(r'https?://', string_value, re.IGNORECASE): return 'url' # Email addresses if re.search(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', string_value): return 'email' # AWS Keys if re.match(r'AKIA[0-9A-Z]{16}', string_value): return 'api_key' # JWT Tokens (Basic check: 3 parts separated by dots, starts with eyJ) if re.match(r'eyJ[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+', string_value): return 'cloud_token' # Private Keys if 'PRIVATE KEY' in string_value: return 'api_key' # IP addresses (IPv4) if re.match(r'^(?:\d{1,3}\.){3}\d{1,3}$', string_value.strip()): return 'ip' # MAC addresses if re.match(r'^(?:[0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}$', string_value.strip()): return 'mac' # File paths if re.match(r'/(?:bin|etc|usr|var|tmp|opt|home|dev|proc|sys|lib|mnt)/', string_value): return 'path' # Certificates if re.search(r'-----BEGIN [A-Z ]+-----', string_value, re.IGNORECASE): return 'certificate' # Base64 Heuristic (length multiple of 4, valid chars, ends with = or == optionally) # Only if length > 20 to avoid false positives on short strings if len(string_value) > 20 and len(string_value) % 4 == 0 and re.match(r'^[A-Za-z0-9+/]+={0,2}$', string_value): return 'base64_data' return 'generic'