Scan target(s) and return (services, errors). Args: target_value: IP, CIDR, range, or comma-separated list Returns: Tuple of (discovered_services, error_messages)
(self, target_value: str)
| 208 | |
| 209 | async def scan(self, target_value: str) -> tuple[list[dict], list[str]]: |
| 210 | """ |
| 211 | Scan target(s) and return (services, errors). |
| 212 | |
| 213 | Args: |
| 214 | target_value: IP, CIDR, range, or comma-separated list |
| 215 | |
| 216 | Returns: |
| 217 | Tuple of (discovered_services, error_messages) |
| 218 | """ |
| 219 | ips = expand_target(target_value) |
| 220 | if not ips: |
| 221 | return [], [f"No valid IPs parsed from: {target_value}"] |
| 222 | |
| 223 | self.logger.info(f"Scanning {len(ips)} host(s) with profile '{scrub(self.profile)}'") |
| 224 | |
| 225 | semaphore = asyncio.Semaphore(self.max_parallel) |
| 226 | errors: list[str] = [] |
| 227 | all_services: list[dict] = [] |
| 228 | lock = asyncio.Lock() |
| 229 | |
| 230 | async def scan_one(ip: str) -> None: |
| 231 | async with semaphore: |
| 232 | try: |
| 233 | services = await scan_host(ip, self.profile, self.custom_ports) |
| 234 | if services: |
| 235 | async with lock: |
| 236 | all_services.extend(services) |
| 237 | except Exception as e: |
| 238 | async with lock: |
| 239 | errors.append(f"{ip}: {e}") |
| 240 | |
| 241 | await asyncio.gather(*[scan_one(ip) for ip in ips]) |
| 242 | return all_services, errors |
| 243 | |
| 244 | async def run(self, target: Target) -> ScanResult: |
| 245 | """Security-suite compatible run() interface.""" |
| 246 | result = ScanResult(target=target, module="vulnscan.scanner") |
no test coverage detected