#!/usr/bin/env python3 """Analyze UEFI firmware images for implants and modifications. Parses UEFI firmware structure and checks against known implant patterns. """ from __future__ import annotations import argparse import hashlib import json import re import struct import sys from datetime import datetime from pathlib import Path # Known UEFI implant signatures KNOWN_IMPLANTS = { "LoJax": { "description": "APT28/Fancy Bear UEFI rootkit (2018)", "indicators": [b"rpcnetp", b"autoche.exe", b"ReWriter_read"], }, "MosaicRegressor": { "description": "Chinese-speaking threat actor UEFI implant (2020)", "indicators": [b"IntelUpdate", b"load.rem"], }, "CosmicStrand": { "description": "Firmware-level rootkit in UEFI images (2022)", "indicators": [b"CosmicStrand", b"rkloader"], }, "BlackLotus": { "description": "First UEFI bootkit bypassing Secure Boot (2023)", "indicators": [b"BLRecovery", b"grubx64.efi"], }, "ESPecter": { "description": "EFI System Partition bootkit (2021)", "indicators": [b"EfiGuard", b"bootmgfw_orig"], }, } # UEFI firmware volume GUID EFI_FV_GUID = b"\x78\xE5\x8C\x8C\x3D\x8A\x1C\x4F\x99\x35\x89\x61\x85\xC3\x2D\xD3" def compute_hash(data) -> dict: return hashlib.sha256(data).hexdigest() def find_pe_headers(data) -> dict: """Find PE headers in firmware image (DXE drivers are PE/COFF).""" pe_locations = [] offset = 0 while True: pos = data.find(b"MZ", offset) if pos == -1: break # Verify PE signature try: pe_offset = struct.unpack_from(" list: """Check firmware against known implant signatures.""" findings = [] for implant_name, info in KNOWN_IMPLANTS.items(): for indicator in info["indicators"]: if indicator in data: findings.append({ "type": "known_implant", "severity": "critical", "implant": implant_name, "description": info["description"], "indicator": indicator.decode("ascii", errors="ignore"), "offset": hex(data.find(indicator)), }) return findings def find_suspicious_strings(data) -> list: """Find strings suspicious in UEFI context.""" findings = [] suspicious_patterns = [ (rb"cmd\.exe", "Command shell reference"), (rb"powershell", "PowerShell reference"), (rb"/bin/sh", "Unix shell reference"), (rb"http://|https://", "URL in firmware"), (rb"\\x00r\\x00o\\x00o\\x00t", "Root string (Unicode)"), (rb"backdoor|rootkit|payload|shellcode", "Malicious keyword"), (rb"CreateRemoteThread|VirtualAlloc", "Windows API in firmware"), ] for pattern, description in suspicious_patterns: matches = list(re.finditer(pattern, data, re.IGNORECASE)) if matches: findings.append({ "type": "suspicious_string", "severity": "high", "description": description, "match_count": len(matches), "first_offset": hex(matches[0].start()), }) return findings def analyze_entropy_regions(data, block_size=4096) -> dict: """Find high-entropy regions that might contain encrypted/compressed payloads.""" import math from collections import Counter high_entropy_regions = [] for offset in range(0, len(data) - block_size, block_size): block = data[offset:offset + block_size] counter = Counter(block) entropy = -sum( (count / block_size) * math.log2(count / block_size) for count in counter.values() ) # Very high entropy might indicate encrypted payload if entropy > 7.8: high_entropy_regions.append({ "offset": hex(offset), "entropy": round(entropy, 3), "size": block_size, }) return high_entropy_regions def analyze_uefi(firmware_path) -> dict: """Main UEFI analysis function.""" path = Path(firmware_path) if not path.exists(): print(f"[!] File not found: {firmware_path}", file=sys.stderr) sys.exit(1) data = path.read_bytes() report = { "tool": "uefi_analyzer", "timestamp": datetime.now().isoformat(), "firmware_file": str(firmware_path), "firmware_size": len(data), "firmware_hash": compute_hash(data), } # Find PE headers (DXE drivers) pe_headers = find_pe_headers(data) report["pe_headers"] = pe_headers report["dxe_driver_count"] = len(pe_headers) # Check known implants implant_findings = check_implant_signatures(data) report["implant_check"] = implant_findings # Check suspicious strings string_findings = find_suspicious_strings(data) report["suspicious_strings"] = string_findings # Check for UEFI firmware volume signatures fv_count = data.count(EFI_FV_GUID) report["firmware_volumes"] = fv_count # Entropy analysis high_entropy = analyze_entropy_regions(data) report["high_entropy_regions"] = len(high_entropy) report["entropy_details"] = high_entropy[:20] # Limit output # Overall risk assessment risk_score = 0 if implant_findings: risk_score = 100 risk_score += len(string_findings) * 15 risk_score += min(len(high_entropy) * 2, 20) report["risk_assessment"] = { "score": min(risk_score, 100), "level": "critical" if risk_score >= 60 else ("high" if risk_score >= 40 else ("medium" if risk_score >= 20 else "low")), } return report def main() -> None: parser = argparse.ArgumentParser(description="Analyze UEFI firmware for implants") parser.add_argument("--input", "--firmware", "-f", required=True, help="UEFI firmware image") parser.add_argument("--output", "-o", help="Output file (JSON)") parser.add_argument("--format", choices=["json", "text"], default="text") args = parser.parse_args() report = analyze_uefi(args.firmware) if args.format == "json": output = json.dumps(report, indent=2) else: output = f"=== UEFI Firmware Analysis ===\n" output += f"File: {report['firmware_file']}\n" output += f"SHA256: {report['firmware_hash']}\n" output += f"Size: {report['firmware_size']} bytes\n" output += f"Firmware Volumes: {report['firmware_volumes']}\n" output += f"DXE Drivers (PE): {report['dxe_driver_count']}\n" output += f"High Entropy Regions: {report['high_entropy_regions']}\n\n" risk = report["risk_assessment"] output += f"Risk: {risk['level'].upper()} (score: {risk['score']}/100)\n\n" if report["implant_check"]: output += "!!! KNOWN IMPLANT SIGNATURES DETECTED !!!\n" for f in report["implant_check"]: output += f" [{f['severity'].upper()}] {f['implant']}: {f['description']}\n" output += "\n" if report["suspicious_strings"]: output += "Suspicious Strings:\n" for f in report["suspicious_strings"]: output += f" [{f['severity'].upper()}] {f['description']} ({f['match_count']} matches)\n" if args.output: Path(args.output).write_text(output) print(f"[+] Report saved to {args.output}") else: print(output) if __name__ == "__main__": main()