Query VirusTotal for target.
(self, target: Target)
| 21 | |
| 22 | async def run(self, target: Target) -> ScanResult: |
| 23 | """Query VirusTotal for target.""" |
| 24 | result = self.create_result(target) |
| 25 | |
| 26 | if not self.api_key: |
| 27 | result.errors.append("VirusTotal API key not configured. Set SECSUITE_VIRUSTOTAL_API_KEY") |
| 28 | result.success = False |
| 29 | result.complete() |
| 30 | return result |
| 31 | |
| 32 | self.logger.info(f"Querying VirusTotal for {target.value}") |
| 33 | |
| 34 | try: |
| 35 | headers = {"x-apikey": self.api_key} |
| 36 | |
| 37 | async with HTTPClient(headers=headers) as client: |
| 38 | if target.target_type == "domain": |
| 39 | await self._check_domain(client, target.value, result) |
| 40 | elif target.target_type == "ip": |
| 41 | await self._check_ip(client, target.value, result) |
| 42 | elif target.target_type == "url": |
| 43 | await self._check_url(client, target.value, result) |
| 44 | else: |
| 45 | # Try domain lookup for unknown types |
| 46 | await self._check_domain(client, target.value, result) |
| 47 | |
| 48 | except Exception as e: |
| 49 | result.errors.append(f"VirusTotal query failed: {str(e)}") |
| 50 | result.success = False |
| 51 | |
| 52 | result.complete() |
| 53 | return result |
| 54 | |
| 55 | async def _check_domain(self, client: HTTPClient, domain: str, result: ScanResult) -> None: |
| 56 | """Check domain reputation.""" |
| 57 | response = await client.get(f"{self.VT_API_BASE}/domains/{domain}") |
no test coverage detected