Validate multiple protocols against port scan results Args: protocols: List of protocol names to validate port_scan_results: Results from port scanning max_threads: Maximum number of threads per validator timeout: Timeout per
(self, protocols: List[str], port_scan_results: Dict,
max_threads: int = 10, timeout: int = 5, **kwargs)
| 96 | return info |
| 97 | |
| 98 | def validate_protocols(self, protocols: List[str], port_scan_results: Dict, |
| 99 | max_threads: int = 10, timeout: int = 5, **kwargs) -> Dict[str, Dict]: |
| 100 | """ |
| 101 | Validate multiple protocols against port scan results |
| 102 | |
| 103 | Args: |
| 104 | protocols: List of protocol names to validate |
| 105 | port_scan_results: Results from port scanning |
| 106 | max_threads: Maximum number of threads per validator |
| 107 | timeout: Timeout per validation |
| 108 | **kwargs: Additional protocol-specific arguments |
| 109 | |
| 110 | Returns: |
| 111 | Dict mapping protocol names to their validation results |
| 112 | """ |
| 113 | if not port_scan_results: |
| 114 | return {} |
| 115 | |
| 116 | results = {} |
| 117 | |
| 118 | for protocol in protocols: |
| 119 | if protocol not in self.validators: |
| 120 | logger.warning(f"Protocol '{protocol}' not available, skipping") |
| 121 | continue |
| 122 | |
| 123 | logger.info(f"🔍 Validating {protocol.upper()} protocol...") |
| 124 | |
| 125 | try: |
| 126 | # Get validator instance |
| 127 | validator = self.get_validator(protocol, timeout=timeout, max_threads=max_threads) |
| 128 | |
| 129 | # Collect relevant IP:port combinations |
| 130 | ip_port_list = self._collect_targets_for_protocol(port_scan_results, validator) |
| 131 | |
| 132 | if not ip_port_list: |
| 133 | logger.info(f"No {protocol.upper()} ports found for validation") |
| 134 | results[protocol] = {} |
| 135 | continue |
| 136 | |
| 137 | # Perform validation |
| 138 | if hasattr(validator, 'validate_targets_threaded'): |
| 139 | # Use the base validator method |
| 140 | validation_results = validator.validate_targets_threaded(ip_port_list, **kwargs) |
| 141 | elif hasattr(validator, 'validate_smb_ports_threaded'): |
| 142 | # Special case for SMB manager |
| 143 | validation_results = validator.validate_smb_ports_threaded( |
| 144 | ip_port_list, |
| 145 | kwargs.get('username', ''), |
| 146 | kwargs.get('password', ''), |
| 147 | kwargs.get('domain', ''), |
| 148 | kwargs.get('ntlm_hash', ''), |
| 149 | kwargs.get('kerberos_ticket', '') |
| 150 | ) |
| 151 | else: |
| 152 | logger.warning(f"Validator for {protocol} doesn't have a threaded validation method") |
| 153 | validation_results = {} |
| 154 | |
| 155 | results[protocol] = validation_results |
nothing calls this directly
no test coverage detected