DaStra3003
626 lines
20 KiB
Python
626 lines
20 KiB
Python
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' |