trojan_scanner.py

#!/usr/bin/env python3"""Trojan ScannerLocation: 03-malware-analysis/trojans/detection/trojan_scanner.pyThis script scans for trojan infections using signature-based detection,heuristic analysis, and behavioral indicators.""" import osimport sysimport hashlibimport jsonimport timeimport argparseimport psutilimport reimport shutilimport subprocessfrom pathlib import Pathfrom datetime import datetime, timedeltafrom collections import defaultdictimport pefileimport requestsfrom colorama import init, Fore, Style init(autoreset=True) class TrojanScanner:    """    Trojan Scanner with multiple detection methods    """        def __init__(self, scan_paths=None, signatures_file="trojan_signatures.json",                 quarantine_dir="/quarantine", use_virustotal=False, vt_api_key=None):        """        Initialize Trojan Scanner                Args:            scan_paths: List of paths to scan            signatures_file: File containing trojan signatures            quarantine_dir: Directory for quarantined files            use_virustotal: Enable VirusTotal API checking            vt_api_key: VirusTotal API key        """        self.scan_paths = scan_paths or [            "C:\\" if os.name == 'nt' else "/",            str(Path.home())        ]        self.signatures_file = signatures_file        self.quarantine_dir = quarantine_dir        self.use_virustotal = use_virustotal        self.vt_api_key = vt_api_key                # Create quarantine directory        os.makedirs(quarantine_dir, exist_ok=True)                # Signatures storage        self.signatures = self.load_signatures()                # Known trojan indicators        self.trojan_indicators = {            'file_names': [                'svchost.exe', 'explorer.exe', 'iexplore.exe', 'winlogon.exe',                'services.exe', 'lsass.exe', 'csrss.exe', 'smss.exe'            ],            'suspicious_strings': [                'mIRC', 'P2P', 'backdoor', 'rat', 'remote admin',                'keylogger', 'password stealer', 'inject', 'hook',                'CreateRemoteThread', 'WriteProcessMemory', 'VirtualAllocEx'            ],            'registry_keys': [                r'SOFTWARE\Microsoft\Windows\CurrentVersion\Run',                r'SOFTWARE\Microsoft\Windows\CurrentVersion\RunOnce',                r'SOFTWARE\Microsoft\Windows\CurrentVersion\RunServices',                r'SOFTWARE\Wow6432Node\Microsoft\Windows\CurrentVersion\Run'            ],            'suspicious_ports': [4444, 1337, 6667, 8080, 8443, 31337, 12345, 27374],            'suspicious_domains': [                'no-ip.org', 'dyndns.org', 'servehttp.com', 'redirectme.net',                'hopto.org', 'serveftp.net', 'ddns.net', 'zapto.org'            ]        }                # Statistics        self.stats = {            'files_scanned': 0,            'suspicious_files': 0,            'infected_files': 0,            'quarantined_files': 0,            'errors': 0,            'start_time': time.time()        }                # Scan results        self.results = {            'scan_id': datetime.now().strftime('%Y%m%d_%H%M%S'),            'timestamp': datetime.now().isoformat(),            'findings': [],            'statistics': self.stats        }        def load_signatures(self):        """Load trojan signatures from file"""        if os.path.exists(self.signatures_file):            try:                with open(self.signatures_file, 'r') as f:                    return json.load(f)            except Exception as e:                print(f"{Fore.YELLOW}[!] Error loading signatures: {e}{Style.RESET_ALL}")                # Default signatures        return {            'hashes': {},            'strings': [],            'patterns': [],            'yara_rules': []        }        def save_signatures(self):        """Save signatures to file"""        try:            with open(self.signatures_file, 'w') as f:                json.dump(self.signatures, f, indent=2)            print(f"{Fore.GREEN}[✓] Signatures saved to {self.signatures_file}{Style.RESET_ALL}")        except Exception as e:            print(f"{Fore.RED}[✗] Error saving signatures: {e}{Style.RESET_ALL}")        def calculate_file_hash(self, filepath, hash_type='sha256'):        """Calculate file hash"""        try:            hash_func = hashlib.new(hash_type)            with open(filepath, 'rb') as f:                for chunk in iter(lambda: f.read(8192), b''):                    hash_func.update(chunk)            return hash_func.hexdigest()        except Exception:            return None        def check_virustotal(self, filepath):        """Check file against VirusTotal API"""        if not self.use_virustotal or not self.vt_api_key:            return None                try:            file_hash = self.calculate_file_hash(filepath)            if not file_hash:                return None                        url = f"https://www.virustotal.com/api/v3/files/{file_hash}"            headers = {'x-apikey': self.vt_api_key}                        response = requests.get(url, headers=headers, timeout=10)            if response.status_code == 200:                data = response.json()                stats = data.get('data', {}).get('attributes', {}).get('last_analysis_stats', {})                return {                    'malicious': stats.get('malicious', 0),                    'suspicious': stats.get('suspicious', 0),                    'undetected': stats.get('undetected', 0),                    'total': sum(stats.values())                }            elif response.status_code == 404:                return {'not_found': True}                    except Exception as e:            print(f"{Fore.YELLOW}[!] VirusTotal API error: {e}{Style.RESET_ALL}")                return None        def check_pe_file(self, filepath):        """Analyze PE file for suspicious characteristics"""        try:            pe = pefile.PE(filepath)            suspicious = []                        # Check for suspicious sections            for section in pe.sections:                name = section.Name.decode('utf-8', errors='ignore').strip('\x00')                characteristics = section.Characteristics                                # Suspicious section names                if name in ['.upx', '.packed', '.themida', '.vmp']:                    suspicious.append(f"Packed section: {name}")                                # Check for executable and writable sections                if characteristics & 0x80000000 and characteristics & 0x20000000:  # MEM_WRITE and MEM_EXECUTE                    suspicious.append(f"Executable and writable section: {name}")                        # Check imports for suspicious APIs            suspicious_apis = [                'CreateRemoteThread', 'WriteProcessMemory', 'ReadProcessMemory',                'VirtualAllocEx', 'SetWindowsHookEx', 'GetAsyncKeyState',                'socket', 'connect', 'send', 'recv', 'URLDownloadToFile'            ]                        if hasattr(pe, 'DIRECTORY_ENTRY_IMPORT'):                for entry in pe.DIRECTORY_ENTRY_IMPORT:                    for imp in entry.imports:                        if imp.name and imp.name.decode('utf-8', errors='ignore') in suspicious_apis:                            suspicious.append(f"Suspicious API: {imp.name.decode('utf-8', errors='ignore')}")                        # Check for suspicious resources            if hasattr(pe, 'DIRECTORY_ENTRY_RESOURCE'):                for resource_type in pe.DIRECTORY_ENTRY_RESOURCE.entries:                    if resource_type.name is not None:                        if str(resource_type.name).find('RT_RCDATA') >= 0:                            suspicious.append("Contains embedded resources (possible payload)")                        # Check for digital signature            if not hasattr(pe, 'DIRECTORY_ENTRY_SECURITY'):                suspicious.append("No digital signature")                        # Check entry point            if pe.OPTIONAL_HEADER.AddressOfEntryPoint in [section.VirtualAddress for section in pe.sections]:                suspicious.append("Entry point in suspicious location")                        return suspicious                    except Exception as e:            return [f"PE analysis error: {e}"]        def check_strings(self, filepath):        """Extract and analyze strings from file"""        suspicious_strings = []                try:            # Extract strings using strings command            if os.name == 'posix':                result = subprocess.run(['strings', filepath], capture_output=True, text=True, timeout=10)                strings = result.stdout.split('\n')            else:                # Simple string extraction for Windows                with open(filepath, 'rb') as f:                    data = f.read()                    strings = []                    current = []                    for b in data:                        if 32 <= b <= 126:  # Printable ASCII                            current.append(chr(b))                        elif current:                            if len(current) > 4:  # Minimum string length                                strings.append(''.join(current))                            current = []                        # Check for suspicious strings            for string in strings[:1000]:  # Limit for performance                string_lower = string.lower()                                # Check against known suspicious strings                for indicator in self.trojan_indicators['suspicious_strings']:                    if indicator.lower() in string_lower:                        suspicious_strings.append(string)                        break                                # Check for URLs/IPs                if re.match(r'https?://', string_lower):                    suspicious_strings.append(f"URL: {string}")                                if re.match(r'\d+\.\d+\.\d+\.\d+', string):                    suspicious_strings.append(f"IP: {string}")                                # Check for dynamic DNS domains                for domain in self.trojan_indicators['suspicious_domains']:                    if domain in string_lower:                        suspicious_strings.append(f"Dynamic DNS: {string}")                    except Exception as e:            suspicious_strings.append(f"String extraction error: {e}")                return suspicious_strings[:20]  # Return top 20        def check_entropy(self, filepath):        """Calculate file entropy (high entropy indicates packed/encrypted)"""        try:            with open(filepath, 'rb') as f:                data = f.read()                                if not data:                    return 0                                entropy = 0                for x in range(256):                    p_x = data.count(x) / len(data)                    if p_x > 0:                        entropy += -p_x * (p_x.bit_length())                                return entropy / 8  # Normalize to 0-1                        except Exception:            return 0        def signature_match(self, filepath):        """Check against known trojan signatures"""        file_hash = self.calculate_file_hash(filepath)        if file_hash and file_hash in self.signatures.get('hashes', {}):            return self.signatures['hashes'][file_hash]        return None        def scan_file(self, filepath):        """Scan individual file"""        self.stats['files_scanned'] += 1                findings = {            'file': filepath,            'size': os.path.getsize(filepath) if os.path.exists(filepath) else 0,            'modified': datetime.fromtimestamp(os.path.getmtime(filepath)).isoformat() if os.path.exists(filepath) else None,            'indicators': []        }                # Skip if too large (>100MB)        if findings['size'] > 100 * 1024 * 1024:            return None                # 1. Signature-based detection        signature = self.signature_match(filepath)        if signature:            findings['indicators'].append({                'type': 'signature',                'severity': 'HIGH',                'description': f"Matches known trojan: {signature.get('name', 'Unknown')}"            })            self.stats['infected_files'] += 1                # 2. Check file extension        ext = os.path.splitext(filepath)[1].lower()        if ext in ['.exe', '.dll', '.scr', '.bat', '.cmd', '.vbs', '.ps1']:            # Suspicious file names            filename = os.path.basename(filepath).lower()            for suspicious_name in self.trojan_indicators['file_names']:                if suspicious_name in filename and filename != suspicious_name:                    findings['indicators'].append({                        'type': 'masquerading',                        'severity': 'MEDIUM',                        'description': f"File masquerading as {suspicious_name}"                    })                # 3. PE analysis for executables        if ext in ['.exe', '.dll', '.scr']:            pe_issues = self.check_pe_file(filepath)            for issue in pe_issues:                findings['indicators'].append({                    'type': 'pe_analysis',                    'severity': 'MEDIUM',                    'description': issue                })                # 4. String analysis        suspicious_strings = self.check_strings(filepath)        if suspicious_strings:            findings['indicators'].append({                'type': 'strings',                'severity': 'LOW',                'description': f"Suspicious strings found: {', '.join(suspicious_strings[:3])}"            })                # 5. Entropy check        entropy = self.check_entropy(filepath)        if entropy > 0.7:            findings['indicators'].append({                'type': 'entropy',                'severity': 'MEDIUM',                'description': f"High entropy ({entropy:.2f}) - possible packed/encrypted"            })                # 6. VirusTotal check        if self.use_virustotal:            vt_result = self.check_virustotal(filepath)            if vt_result:                if vt_result.get('malicious', 0) > 0:                    findings['indicators'].append({                        'type': 'virustotal',                        'severity': 'HIGH',                        'description': f"VirusTotal: {vt_result['malicious']}/{vt_result['total']} detections"                    })                    self.stats['infected_files'] += 1                elif vt_result.get('not_found'):                    findings['indicators'].append({                        'type': 'virustotal',                        'severity': 'LOW',                        'description': "Not found in VirusTotal (possibly new malware)"                    })                # Determine overall risk        if findings['indicators']:            self.stats['suspicious_files'] += 1                        # Calculate risk score            risk_score = 0            for indicator in findings['indicators']:                if indicator['severity'] == 'HIGH':                    risk_score += 3                elif indicator['severity'] == 'MEDIUM':                    risk_score += 2                else:                    risk_score += 1                        findings['risk_score'] = risk_score            findings['risk_level'] = 'HIGH' if risk_score > 5 else 'MEDIUM' if risk_score > 2 else 'LOW'                        return findings                return None        def scan_directory(self, directory):        """Scan directory recursively"""        print(f"\n{Fore.CYAN}[*] Scanning directory: {directory}{Style.RESET_ALL}")                for root, dirs, files in os.walk(directory):            # Skip system directories for performance            if any(skip in root.lower() for skip in ['windows\\system32\\config', '$recycle.bin', 'system volume information']):                continue                        for file in files:                filepath = os.path.join(root, file)                                try:                    findings = self.scan_file(filepath)                                        if findings:                        self.results['findings'].append(findings)                                                # Print finding                        risk_color = Fore.RED if findings['risk_level'] == 'HIGH' else Fore.YELLOW if findings['risk_level'] == 'MEDIUM' else Fore.CYAN                        print(f"{risk_color}[!] {findings['risk_level']} risk: {filepath}{Style.RESET_ALL}")                        for indicator in findings['indicators']:                            print(f"      - [{indicator['severity']}] {indicator['description']}")                                        # Progress indicator                    if self.stats['files_scanned'] % 100 == 0:                        print(f"{Fore.CYAN}  [*] Scanned {self.stats['files_scanned']} files...{Style.RESET_ALL}")                                        except Exception as e:                    self.stats['errors'] += 1                    print(f"{Fore.RED}[!] Error scanning {filepath}: {e}{Style.RESET_ALL}")        def scan_processes(self):        """Scan running processes for trojans"""        print(f"\n{Fore.CYAN}[*] Scanning running processes...{Style.RESET_ALL}")                for proc in psutil.process_iter(['pid', 'name', 'exe', 'cmdline']):            try:                proc_info = proc.info                                if proc_info['exe'] and os.path.exists(proc_info['exe']):                    findings = self.scan_file(proc_info['exe'])                                        if findings:                        findings['process'] = {                            'pid': proc_info['pid'],                            'name': proc_info['name'],                            'cmdline': proc_info['cmdline']                        }                        self.results['findings'].append(findings)                                                print(f"{Fore.RED}[!] Suspicious process: {proc_info['name']} (PID: {proc_info['pid']}){Style.RESET_ALL}")                                    except (psutil.NoSuchProcess, psutil.AccessDenied):                pass        def scan_registry(self):        """Scan Windows registry for trojan persistence"""        if os.name != 'nt':            return                print(f"\n{Fore.CYAN}[*] Scanning registry for trojan persistence...{Style.RESET_ALL}")                import winreg                for reg_key in self.trojan_indicators['registry_keys']:            try:                key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, reg_key)                                i = 0                while True:                    try:                        value_name, value_data, _ = winreg.EnumValue(key, i)                                                # Check if value points to suspicious file                        if isinstance(value_data, str) and os.path.exists(value_data):                            findings = self.scan_file(value_data)                                                        if findings:                                findings['registry'] = {                                    'key': reg_key,                                    'value': value_name,                                    'data': value_data                                }                                self.results['findings'].append(findings)                                                                print(f"{Fore.RED}[!] Suspicious registry entry: {reg_key}\\{value_name} -> {value_data}{Style.RESET_ALL}")                                                i += 1                    except WindowsError:                        break                                winreg.CloseKey(key)            except:                pass        def quarantine_file(self, filepath):        """Move suspicious file to quarantine"""        try:            filename = os.path.basename(filepath)            quarantine_path = os.path.join(self.quarantine_dir, f"{int(time.time())}_{filename}")                        # Copy file with metadata            shutil.copy2(filepath, quarantine_path)                        # Add .quarantine extension            os.rename(quarantine_path, quarantine_path + '.quarantine')                        # Delete original            os.remove(filepath)                        self.stats['quarantined_files'] += 1            print(f"{Fore.GREEN}[✓] Quarantined: {filepath}{Style.RESET_ALL}")                        return quarantine_path + '.quarantine'                    except Exception as e:            print(f"{Fore.RED}[✗] Error quarantining {filepath}: {e}{Style.RESET_ALL}")            return None        def generate_report(self):        """Generate scan report"""        self.stats['duration'] = time.time() - self.stats['start_time']        self.results['statistics'] = self.stats                # Save report        report_file = f"trojan_scan_{self.results['scan_id']}.json"        with open(report_file, 'w') as f:            json.dump(self.results, f, indent=2)                # Print summary        print(f"\n{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"{Fore.CYAN}📊 TROJAN SCAN SUMMARY{Style.RESET_ALL}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")        print(f"  Scan ID: {self.results['scan_id']}")        print(f"  Duration: {self.stats['duration']:.2f} seconds")        print(f"  Files Scanned: {self.stats['files_scanned']}")        print(f"  Suspicious Files: {self.stats['suspicious_files']}")        print(f"  Infected Files: {self.stats['infected_files']}")        print(f"  Quarantined Files: {self.stats['quarantined_files']}")        print(f"  Errors: {self.stats['errors']}")        print(f"{Fore.CYAN}{'='*60}{Style.RESET_ALL}")                # List high risk findings        high_risk = [f for f in self.results['findings'] if f.get('risk_level') == 'HIGH']        if high_risk:            print(f"\n{Fore.RED}HIGH RISK FINDINGS:{Style.RESET_ALL}")            for finding in high_risk[:10]:                print(f"  - {finding['file']}")                print(f"\n{Fore.GREEN}[✓] Report saved to {report_file}{Style.RESET_ALL}")                return report_file        def run_scan(self, scan_processes=True, scan_registry=True, auto_quarantine=False):        """Run complete trojan scan"""        print(f"""        {Fore.CYAN}╔═══════════════════════════════════════╗        ║        Trojan Scanner v1.0           ║        ║      FOR EDUCATIONAL USE ONLY        ║        ╚═══════════════════════════════════════╝{Style.RESET_ALL}        """)                print(f"{Fore.YELLOW}[!] WARNING: This tool should only be used on systems you own{Style.RESET_ALL}")        print(f"{Fore.YELLOW}[!] Always have proper backups before scanning{Style.RESET_ALL}\n")                # Scan directories        for path in self.scan_paths:            if os.path.exists(path):                self.scan_directory(path)                # Scan processes        if scan_processes:            self.scan_processes()                # Scan registry        if scan_registry and os.name == 'nt':            self.scan_registry()                # Auto-quarantine if requested        if auto_quarantine:            for finding in self.results['findings']:                if finding.get('risk_level') == 'HIGH':                    response = input(f"\n{Fore.YELLOW}[?] Quarantine {finding['file']}? (yes/no): {Style.RESET_ALL}")                    if response.lower() == 'yes':                        self.quarantine_file(finding['file'])                # Generate report        self.generate_report() def main():    parser = argparse.ArgumentParser(description='Trojan Scanner')    parser.add_argument('-p', '--paths', help='Paths to scan (comma-separated)')    parser.add_argument('--no-processes', action='store_true', help='Skip process scanning')    parser.add_argument('--no-registry', action='store_true', help='Skip registry scanning')    parser.add_argument('--quarantine', action='store_true', help='Auto-quarantine high risk files')    parser.add_argument('--virustotal', action='store_true', help='Enable VirusTotal checking')    parser.add_argument('--vt-api-key', help='VirusTotal API key')    parser.add_argument('--signatures', default='trojan_signatures.json', help='Signatures file')        args = parser.parse_args()        # Parse paths    scan_paths = args.paths.split(',') if args.paths else None        scanner = TrojanScanner(        scan_paths=scan_paths,        signatures_file=args.signatures,        use_virustotal=args.virustotal,        vt_api_key=args.vt_api_key    )        scanner.run_scan(        scan_processes=not args.no_processes,        scan_registry=not args.no_registry,        auto_quarantine=args.quarantine    ) if __name__ == "__main__":    main()