| 87 | clean = 0 |
| 88 | |
| 89 | class AdvancedAnalyzer: |
| 90 | def __init__(self, filepath): |
| 91 | self.filepath = filepath |
| 92 | self.filename = os.path.basename(filepath) |
| 93 | self.size = os.path.getsize(filepath) |
| 94 | self.risk_score = 0 |
| 95 | self.warnings = [] |
| 96 | self.hidden_data = [] |
| 97 | self.indicators = [] |
| 98 | self.data = b"" # Analysis Buffer (Partial) |
| 99 | self.vt_result = "N/A" |
| 100 | self.is_quarantined = False |
| 101 | self.is_partial_read = False |
| 102 | |
| 103 | # --- SMART LOADING (Optimized for 50GB files) --- |
| 104 | try: |
| 105 | with open(filepath, 'rb') as f: |
| 106 | if self.size <= ANALYSIS_RAM_LIMIT: |
| 107 | # Small file: Read all |
| 108 | self.data = f.read() |
| 109 | else: |
| 110 | # Large file: Read Head + Tail only |
| 111 | self.is_partial_read = True |
| 112 | head_size = ANALYSIS_RAM_LIMIT - (10 * 1024 * 1024) # Reserve 10MB for tail |
| 113 | self.data = f.read(head_size) |
| 114 | |
| 115 | try: |
| 116 | f.seek(- (10 * 1024 * 1024), 2) # Go to end minus 10MB |
| 117 | self.data += f.read() |
| 118 | except: pass # File might be smaller than we thought if seek fails |
| 119 | |
| 120 | self.warnings.append(f"Large File ({self.size/1024/1024/1024:.2f} GB). Analyzed Head/Tail only to save RAM.") |
| 121 | except Exception as e: |
| 122 | self.warnings.append(f"Read Error: {str(e)}") |
| 123 | |
| 124 | def get_hashes(self): |
| 125 | # STREAMING HASH (Does not load file into RAM) |
| 126 | s = hashlib.sha256() |
| 127 | try: |
| 128 | with open(self.filepath, 'rb') as f: |
| 129 | while chunk := f.read(8192 * 1024): # Read in 8MB chunks |
| 130 | s.update(chunk) |
| 131 | return s.hexdigest() |
| 132 | except: |
| 133 | return "HASH_ERROR" |
| 134 | |
| 135 | def check_virustotal(self, sha256): |
| 136 | if not VIRUSTOTAL_API_KEY: return |
| 137 | try: |
| 138 | url = f"https://www.virustotal.com/api/v3/files/{sha256}" |
| 139 | headers = {"x-apikey": VIRUSTOTAL_API_KEY} |
| 140 | resp = requests.get(url, headers=headers, timeout=5) |
| 141 | if resp.status_code == 200: |
| 142 | stats = resp.json()['data']['attributes']['last_analysis_stats'] |
| 143 | mal = stats['malicious'] |
| 144 | if mal > 0: |
| 145 | self.risk_score += 15 |
| 146 | self.vt_result = f"[bold red]MALICIOUS ({mal} engines)[/]" |