#!/usr/bin/env python3 """Linux malware analysis script. Performs ELF binary analysis, persistence mechanism enumeration, shared library hijacking detection, container/cloud targeting analysis, SSH key theft investigation, and known family identification for Linux malware samples. """ from __future__ import annotations import argparse import hashlib import json import math import os import re import struct import subprocess import sys from datetime import datetime from pathlib import Path def compute_file_hashes(filepath) -> dict: """Compute MD5, SHA1, and SHA256 hashes of a file. Args: filepath: Path to the file to hash. Returns: Dictionary with md5, sha1, and sha256 hash strings. """ hashes = {"md5": hashlib.md5(), "sha1": hashlib.sha1(), "sha256": hashlib.sha256()} with open(filepath, "rb") as f: while chunk := f.read(8192): for h in hashes.values(): h.update(chunk) return {k: v.hexdigest() for k, v in hashes.items()} def calculate_entropy(data) -> dict: """Calculate Shannon entropy of a byte sequence. Args: data: Bytes object to analyze. Returns: Float representing the entropy value (0.0 - 8.0). """ if not data: return 0.0 freq = [0] * 256 for byte in data: freq[byte] += 1 length = len(data) return -sum( (count / length) * math.log2(count / length) for count in freq if count > 0 ) def analyze_elf_header(filepath) -> dict: """Parse and analyze the ELF header for suspicious characteristics. Examines the ELF header fields including architecture, type, entry point, and section headers. Flags anomalies like missing sections, unusual entry points, or signs of packing/tampering. Args: filepath: Path to the ELF binary. Returns: Dictionary with ELF header analysis results. """ results = { "file_type": "", "architecture": "", "linking": "", "stripped": False, "compiler": "unknown", "packed": False, "entropy": 0.0, "sections": [], "suspicious_sections": [], "imported_functions": [], "rpath_set": False, } try: with open(filepath, "rb") as f: magic = f.read(4) if magic != b"\x7fELF": results["error"] = "Not a valid ELF binary" return results # Use readelf for header analysis header_out = subprocess.run( ["readelf", "-h", filepath], capture_output=True, text=True, timeout=30 ) if header_out.returncode == 0: for line in header_out.stdout.splitlines(): if "Class:" in line: results["architecture"] = line.split(":")[-1].strip() elif "Type:" in line: results["file_type"] = line.split(":")[-1].strip() elif "Machine:" in line: results["architecture"] = line.split(":")[-1].strip() # Get section headers sections_out = subprocess.run( ["readelf", "-S", filepath], capture_output=True, text=True, timeout=30 ) if sections_out.returncode == 0: for line in sections_out.stdout.splitlines(): match = re.search(r"\]\s+(\.\S+)", line) if match: section_name = match.group(1) results["sections"].append(section_name) if section_name not in { ".text", ".data", ".bss", ".rodata", ".dynstr", ".dynsym", ".symtab", ".strtab", ".shstrtab", ".plt", ".got", ".got.plt", ".init", ".fini", ".comment", ".note.gnu.build-id", ".note.ABI-tag", ".rela.dyn", ".rela.plt", ".interp", ".hash", ".gnu.hash", ".gnu.version", ".gnu.version_r", ".init_array", ".fini_array", ".dynamic", ".eh_frame", ".eh_frame_hdr", ".debug_info", ".debug_str", ".debug_abbrev", ".debug_line", ".debug_ranges", ".tbss", ".tdata", ".ctors", ".dtors", }: results["suspicious_sections"].append(section_name) # Check if stripped symbols_out = subprocess.run( ["readelf", "-s", filepath], capture_output=True, text=True, timeout=30 ) results["stripped"] = "Symbol table '.symtab'" not in symbols_out.stdout # Get dynamic dependencies dynamic_out = subprocess.run( ["readelf", "-d", filepath], capture_output=True, text=True, timeout=30 ) if dynamic_out.returncode == 0: results["linking"] = "dynamically linked" if "NEEDED" in dynamic_out.stdout else "statically linked" results["rpath_set"] = "RPATH" in dynamic_out.stdout or "RUNPATH" in dynamic_out.stdout # Get imported functions of interest dynsym_out = subprocess.run( ["readelf", "--dyn-syms", filepath], capture_output=True, text=True, timeout=30 ) suspicious_funcs = { "connect", "socket", "execve", "execvp", "system", "fork", "dlopen", "dlsym", "ptrace", "mprotect", "mmap", "prctl", "setuid", "setgid", "chroot", "mount", "init_module", "unlink", "rename", "open", "popen", "dup2", "kill", } if dynsym_out.returncode == 0: for line in dynsym_out.stdout.splitlines(): for func in suspicious_funcs: if re.search(rf"\b{func}\b", line): results["imported_functions"].append(func) results["imported_functions"] = sorted(set(results["imported_functions"])) # Calculate entropy with open(filepath, "rb") as f: data = f.read() results["entropy"] = round(calculate_entropy(data), 2) results["packed"] = results["entropy"] > 7.2 # Detect compiler strings_out = subprocess.run( ["strings", "-a", filepath], capture_output=True, text=True, timeout=30 ) all_strings = strings_out.stdout if "GCC:" in all_strings: gcc_match = re.search(r"GCC:.*?(\d+\.\d+\.\d+)", all_strings) results["compiler"] = f"GCC {gcc_match.group(1)}" if gcc_match else "GCC (version unknown)" elif "go build" in all_strings.lower() or "runtime.gopanic" in all_strings: results["compiler"] = "Go" elif "rustc" in all_strings: results["compiler"] = "Rust" elif "clang" in all_strings.lower(): results["compiler"] = "Clang" except Exception as e: results["error"] = str(e) return results def scan_persistence_mechanisms() -> list: """Scan the system for known Linux persistence mechanisms. Checks crontabs, systemd services/timers, shell profiles, init.d scripts, LD_PRELOAD configuration, and SSH authorized_keys for unauthorized entries. Returns: Dictionary with findings for each persistence category. """ findings = { "crontab": {"found": False, "entries": []}, "systemd": {"found": False, "services": []}, "ld_preload": {"found": False, "library": None}, "shell_profiles": {"found": False, "files_modified": []}, "init_d": {"found": False, "scripts": []}, "ssh_keys": {"modified": False, "unauthorized_keys_added": 0}, } # Check /etc/ld.so.preload preload_path = Path("/etc/ld.so.preload") if preload_path.exists(): content = preload_path.read_text().strip() if content: findings["ld_preload"]["found"] = True findings["ld_preload"]["library"] = content # Check LD_PRELOAD environment variable ld_preload_env = os.environ.get("LD_PRELOAD", "") if ld_preload_env: findings["ld_preload"]["found"] = True findings["ld_preload"]["library"] = ld_preload_env # Check system crontab cron_dirs = [ "/etc/cron.d", "/etc/cron.daily", "/etc/cron.hourly", "/etc/cron.weekly", "/etc/cron.monthly", ] try: result = subprocess.run( ["crontab", "-l"], capture_output=True, text=True, timeout=10 ) if result.returncode == 0 and result.stdout.strip(): for line in result.stdout.strip().splitlines(): line = line.strip() if line and not line.startswith("#"): findings["crontab"]["entries"].append(line) findings["crontab"]["found"] = True except (subprocess.TimeoutExpired, FileNotFoundError): pass for cron_dir in cron_dirs: if os.path.isdir(cron_dir): for entry in os.listdir(cron_dir): entry_path = os.path.join(cron_dir, entry) if os.path.isfile(entry_path): try: content = Path(entry_path).read_text() if re.search(r"(curl|wget|python|bash|sh)\b.*\|", content): findings["crontab"]["found"] = True findings["crontab"]["entries"].append( f"[SUSPICIOUS] {entry_path}" ) except (PermissionError, OSError): pass # Check systemd services systemd_dirs = [ "/etc/systemd/system", "/usr/lib/systemd/system", "/run/systemd/system", ] for sd_dir in systemd_dirs: if os.path.isdir(sd_dir): for unit_file in os.listdir(sd_dir): if unit_file.endswith((".service", ".timer")): unit_path = os.path.join(sd_dir, unit_file) try: stat = os.stat(unit_path) mtime = datetime.fromtimestamp(stat.st_mtime) if (datetime.now() - mtime).days < 30: findings["systemd"]["found"] = True findings["systemd"]["services"].append({ "name": unit_file, "path": unit_path, "modified": mtime.isoformat(), }) except (PermissionError, OSError): pass # Check SSH authorized_keys ssh_dirs = list(Path("/home").glob("*/.ssh")) + [Path("/root/.ssh")] for ssh_dir in ssh_dirs: auth_keys = ssh_dir / "authorized_keys" if auth_keys.exists(): try: stat = os.stat(auth_keys) mtime = datetime.fromtimestamp(stat.st_mtime) if (datetime.now() - mtime).days < 30: findings["ssh_keys"]["modified"] = True key_count = sum( 1 for line in auth_keys.read_text().splitlines() if line.strip() and not line.startswith("#") ) findings["ssh_keys"]["unauthorized_keys_added"] = key_count except (PermissionError, OSError): pass return findings def analyze_cloud_targeting(filepath) -> dict: """Detect container escape and cloud infrastructure targeting in a binary. Searches for patterns indicating Docker socket abuse, Kubernetes API access, cloud metadata endpoint queries, credential theft, and cryptomining activity. Args: filepath: Path to the binary to analyze. Returns: Dictionary with cloud/container targeting findings. """ results = { "docker_socket_access": False, "kubernetes_api_access": False, "cloud_metadata_access": False, "cloud_credential_theft": [], "container_escape_indicators": [], "cryptomining": { "detected": False, "miner": None, "pool": None, "wallet": None, }, } try: strings_out = subprocess.run( ["strings", "-a", "-n", "6", filepath], capture_output=True, text=True, timeout=60 ) all_strings = strings_out.stdout except (subprocess.TimeoutExpired, FileNotFoundError): return results # Docker socket patterns docker_patterns = [ r"docker\.sock", r"/var/run/docker", r"containers/json", r"dockerapi", r"docker\s+exec", r"docker\s+run", ] for pattern in docker_patterns: if re.search(pattern, all_strings, re.IGNORECASE): results["docker_socket_access"] = True break # Kubernetes patterns k8s_patterns = [ r"kubernetes", r"kube-system", r"serviceaccount", r"/api/v1", r"kubectl", r"kubelet", r"/var/run/secrets/kubernetes", ] for pattern in k8s_patterns: if re.search(pattern, all_strings, re.IGNORECASE): results["kubernetes_api_access"] = True break # Cloud metadata patterns if re.search(r"169\.254\.169\.254", all_strings): results["cloud_metadata_access"] = True if re.search(r"metadata\.(google|azure)", all_strings, re.IGNORECASE): results["cloud_metadata_access"] = True # Credential theft patterns cred_patterns = { "AWS credentials (~/.aws/credentials)": r"\.aws/credentials|AWS_ACCESS_KEY|AKIA[0-9A-Z]{16}", "GCP service account": r"gcloud\s+auth|service_account\.json|GOOGLE_APPLICATION_CREDENTIALS", "Azure credentials": r"az\s+login|\.azure/|AZURE_CLIENT", "Docker config": r"\.docker/config\.json", } for desc, pattern in cred_patterns.items(): if re.search(pattern, all_strings, re.IGNORECASE): results["cloud_credential_theft"].append(desc) # Container escape indicators escape_patterns = { "nsenter with host PID namespace": r"nsenter.*--mount.*--pid|nsenter.*-t\s*1", "cgroup release_agent abuse": r"release_agent|notify_on_release", "Mounting host filesystem": r"/proc/1/root|/proc/sysrq-trigger", "Privileged container detection": r"capsh|CAP_SYS_ADMIN|SYS_PTRACE", } for desc, pattern in escape_patterns.items(): if re.search(pattern, all_strings, re.IGNORECASE): results["container_escape_indicators"].append(desc) # Cryptomining detection mining_patterns = { "pool": r"(stratum\+tcp://[^\s\"']+)", "xmrig": r"xmrig", "wallet": r"(4[0-9AB][1-9A-HJ-NP-Za-km-z]{93})", # Monero address pattern } if re.search(r"(xmrig|cryptonight|hashrate|stratum\+tcp|monero)", all_strings, re.IGNORECASE): results["cryptomining"]["detected"] = True pool_match = re.search(mining_patterns["pool"], all_strings) if pool_match: results["cryptomining"]["pool"] = pool_match.group(1) if re.search(mining_patterns["xmrig"], all_strings, re.IGNORECASE): results["cryptomining"]["miner"] = "XMRig" wallet_match = re.search(mining_patterns["wallet"], all_strings) if wallet_match: results["cryptomining"]["wallet"] = wallet_match.group(1) return results def identify_malware_family(filepath) -> dict: """Identify known Linux malware families based on string signatures and behavioral markers. Checks against signature patterns for TeamTNT, Kinsing, XorDDoS, Mirai variants, BPFDoor, and other known Linux malware families. Args: filepath: Path to the binary to analyze. Returns: Dictionary with family identification results and confidence level. """ results = { "family": "unknown", "confidence": "low", "variant": None, "indicators": [], } try: strings_out = subprocess.run( ["strings", "-a", "-n", "4", filepath], capture_output=True, text=True, timeout=60 ) all_strings = strings_out.stdout except (subprocess.TimeoutExpired, FileNotFoundError): return results families = { "TeamTNT": { "patterns": [ r"teamtnt", r"hilde@", r"chimaera", r"bioset", r"masscan", r"zgrab", r"pnscan", ], "min_matches": 2, }, "Kinsing": { "patterns": [ r"kinsing", r"kdevtmpfsi", r"libsystem\.so", r"/tmp/kinsing", r"spreading", ], "min_matches": 1, }, "XorDDoS": { "patterns": [ r"/lib/libudev\.so", r"/boot/.*random", r"xorddos", r"BB2FA36AAA9541F0", ], "min_matches": 2, }, "Mirai": { "patterns": [ r"table_init", r"attack_", r"scanner_init", r"/bin/busybox", r"MIRAI", r"killer_init", ], "min_matches": 2, }, "BPFDoor": { "patterns": [ r"haldrund", r"packet_filter", r"socket_filter", r"/var/run/haldrund", r"SOCK_RAW.*AF_PACKET", ], "min_matches": 2, }, } best_match = None best_score = 0 for family_name, family_info in families.items(): matches = [] for pattern in family_info["patterns"]: if re.search(pattern, all_strings, re.IGNORECASE): matches.append(pattern) if len(matches) >= family_info["min_matches"]: score = len(matches) / len(family_info["patterns"]) if score > best_score: best_score = score best_match = family_name results["indicators"] = [ f"Matched pattern: {m}" for m in matches ] if best_match: results["family"] = best_match results["confidence"] = "high" if best_score > 0.5 else "medium" return results def extract_iocs(filepath) -> list: """Extract indicators of compromise from a binary. Searches for IP addresses, domain names, URLs, file paths, and other indicators that can be used for detection and threat intelligence. Args: filepath: Path to the binary to analyze. Returns: Dictionary with categorized IOCs. """ iocs = { "ip_addresses": [], "domains": [], "urls": [], "file_paths": [], "email_addresses": [], } try: strings_out = subprocess.run( ["strings", "-a", "-n", "6", filepath], capture_output=True, text=True, timeout=60 ) all_strings = strings_out.stdout except (subprocess.TimeoutExpired, FileNotFoundError): return iocs # Extract IP addresses (exclude common false positives) ip_pattern = r"\b(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b" for match in re.finditer(ip_pattern, all_strings): ip = match.group() if not ip.startswith(("0.", "127.", "255.")): iocs["ip_addresses"].append(ip) iocs["ip_addresses"] = sorted(set(iocs["ip_addresses"])) # Extract URLs url_pattern = r"https?://[^\s\"'<>]{4,}" for match in re.finditer(url_pattern, all_strings): iocs["urls"].append(match.group()) iocs["urls"] = sorted(set(iocs["urls"])) # Extract suspicious file paths path_pattern = r"(/tmp/[^\s\"']+|/dev/shm/[^\s\"']+|/var/tmp/[^\s\"']+|/root/\.[^\s\"']+)" for match in re.finditer(path_pattern, all_strings): iocs["file_paths"].append(match.group()) iocs["file_paths"] = sorted(set(iocs["file_paths"])) return iocs def run_full_analysis(filepath, audit_log=None) -> dict: """Run all analysis modules and produce a consolidated report. Args: filepath: Path to the malware sample. audit_log: Optional path to an auditd log file for correlation. Returns: Dictionary with complete analysis results. """ file_hashes = compute_file_hashes(filepath) # Get file type try: file_out = subprocess.run( ["file", filepath], capture_output=True, text=True, timeout=10 ) file_type = file_out.stdout.split(":", 1)[-1].strip() except Exception: file_type = "unknown" report = { "sample": { "filename": os.path.basename(filepath), "sha256": file_hashes["sha256"], "md5": file_hashes["md5"], "file_type": file_type, "size_bytes": os.path.getsize(filepath), }, "elf_analysis": analyze_elf_header(filepath), "family_identification": identify_malware_family(filepath), "persistence": scan_persistence_mechanisms(), "cloud_targeting": analyze_cloud_targeting(filepath), "iocs": extract_iocs(filepath), "mitre_attack": [], "analysis_timestamp": datetime.utcnow().isoformat() + "Z", } # Map findings to MITRE ATT&CK techniques attack_mapping = [] if report["persistence"]["crontab"]["found"]: attack_mapping.append("T1053.003") # Cron if report["persistence"]["systemd"]["found"]: attack_mapping.append("T1543.002") # Systemd Service if report["persistence"]["ld_preload"]["found"]: attack_mapping.append("T1574.006") # LD_PRELOAD Hijacking if report["persistence"]["ssh_keys"]["modified"]: attack_mapping.append("T1552.004") # SSH Authorized Keys if report["cloud_targeting"]["cryptomining"]["detected"]: attack_mapping.append("T1496") # Resource Hijacking if report["cloud_targeting"]["docker_socket_access"]: attack_mapping.append("T1610") # Deploy Container if report["cloud_targeting"]["cloud_metadata_access"]: attack_mapping.append("T1552.005") # Cloud Instance Metadata API if "execve" in report["elf_analysis"].get("imported_functions", []): attack_mapping.append("T1059.004") # Unix Shell report["mitre_attack"] = sorted(set(attack_mapping)) return report def main() -> None: parser = argparse.ArgumentParser( description="Linux Malware Analyzer - Analyze ELF binaries and detect Linux-specific threats", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: %(prog)s --sample malware.elf --mode elf-analysis --output elf.json %(prog)s --mode persistence --output persistence.json %(prog)s --sample malware.elf --mode cloud-analysis --output cloud.json %(prog)s --sample malware.elf --mode family-id --output family.json %(prog)s --sample malware.elf --mode full-analysis --output report.json """, ) parser.add_argument( "--input", "--sample", help="Path to the malware sample (ELF binary)", ) parser.add_argument( "--mode", choices=["elf-analysis", "persistence", "cloud-analysis", "family-id", "full-analysis"], default="full-analysis", help="Analysis mode to run (default: full-analysis)", ) parser.add_argument( "--audit-log", help="Path to auditd log file for correlation (optional)", ) parser.add_argument( "--output", default="linux_analysis.json", help="Output file path for JSON results (default: linux_analysis.json)", ) parser.add_argument( "--format", default="json", choices=["json", "text", "csv"], help="Output format (default: json)", ) args = parser.parse_args() # Validate arguments if args.mode != "persistence" and not args.sample: parser.error("--sample is required for all modes except 'persistence'") if args.sample and not os.path.isfile(args.sample): print(f"Error: Sample file not found: {args.sample}", file=sys.stderr) sys.exit(1) # Run the selected analysis mode if args.mode == "elf-analysis": results = {"elf_analysis": analyze_elf_header(args.sample)} elif args.mode == "persistence": results = {"persistence": scan_persistence_mechanisms()} elif args.mode == "cloud-analysis": results = {"cloud_targeting": analyze_cloud_targeting(args.sample)} elif args.mode == "family-id": results = {"family_identification": identify_malware_family(args.sample)} elif args.mode == "full-analysis": results = run_full_analysis(args.sample, args.audit_log) # Write output output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) with open(output_path, "w") as f: json.dump(results, f, indent=2, default=str) print(f"Analysis complete. Results written to: {output_path}") # Print summary to stdout if args.mode == "full-analysis" and "family_identification" in results: family = results["family_identification"] print(f" Family: {family.get('family', 'unknown')} " f"(confidence: {family.get('confidence', 'low')})") if "elf_analysis" in results: elf = results["elf_analysis"] print(f" Architecture: {elf.get('architecture', 'unknown')}") print(f" Entropy: {elf.get('entropy', 0)}") print(f" Packed: {elf.get('packed', False)}") if __name__ == "__main__": main()