Debug endpoint for vulnerability scanner
()
| 17196 | continue |
| 17197 | ports_str = host.get('ports', '') |
| 17198 | ports_list = [p.strip() for p in ports_str.split(',') if p.strip()] if ports_str else [] |
| 17199 | threats = detect_threats( |
| 17200 | vendor=host.get('vendor', ''), |
| 17201 | mac=host.get('mac', ''), |
| 17202 | hostname=host.get('hostname', ''), |
| 17203 | ports=ports_list |
| 17204 | ) |
| 17205 | for threat in threats: |
| 17206 | if host_ip not in grouped: |
| 17207 | grouped[host_ip] = { |
| 17208 | 'ip': host_ip, |
| 17209 | 'total_vulnerabilities': 0, |
| 17210 | 'severity_counts': {'critical': 0, 'high': 0, 'medium': 0, 'low': 0}, |
| 17211 | 'affected_ports': set(), |
| 17212 | 'affected_services': set(), |
| 17213 | 'vulnerabilities': [] |
| 17214 | } |
| 17215 | threat_vuln_id = f"threat_{threat['id']}_{host_ip.replace('.','_')}" |
| 17216 | # Avoid duplicate threat entries |
| 17217 | existing_ids = {v['id'] for v in grouped[host_ip]['vulnerabilities']} |
| 17218 | if threat_vuln_id not in existing_ids: |
| 17219 | grouped[host_ip]['total_vulnerabilities'] += 1 |
| 17220 | sev = threat['severity'] |
| 17221 | if sev in grouped[host_ip]['severity_counts']: |
| 17222 | grouped[host_ip]['severity_counts'][sev] += 1 |
| 17223 | grouped[host_ip]['affected_services'].add(threat['category']) |
| 17224 | grouped[host_ip]['vulnerabilities'].append({ |
| 17225 | 'id': threat_vuln_id, |
| 17226 | 'port': 0, |
| 17227 | 'service': threat['category'], |
| 17228 | 'vulnerability': f"⚠ ROGUE DEVICE: {threat['name']} — {threat['description']}", |
| 17229 | 'severity': sev, |
| 17230 | 'discovered': datetime.now().isoformat(), |
| 17231 | 'status': 'active', |
| 17232 | 'resolved': None |
| 17233 | }) |
| 17234 | except Exception as e: |
| 17235 | logger.warning(f"Threat detection enrichment failed: {e}") |
| 17236 | |
| 17237 | # Convert sets to lists for JSON serialization |
| 17238 | for host_data in grouped.values(): |
| 17239 | host_data['affected_ports'] = sorted(list(host_data['affected_ports'])) |
| 17240 | host_data['affected_services'] = sorted(list(host_data['affected_services'])) |
| 17241 | # Sort vulnerabilities by severity |
| 17242 | severity_order = {'critical': 0, 'high': 1, 'medium': 2, 'low': 3} |
| 17243 | host_data['vulnerabilities'].sort( |
| 17244 | key=lambda v: severity_order.get(v['severity'], 4) |
| 17245 | ) |
| 17246 | |
| 17247 | # Convert to list and sort by total vulnerability count |
| 17248 | grouped_list = sorted( |
nothing calls this directly
no test coverage detected