Create or update vulnerability scan results file for discovered host
(ip, hostname, mac, is_alive)
| 2995 | from advanced_vuln_scanner import ScanType |
| 2996 | try: |
| 2997 | scan_type = ScanType(data.get('scan_type', 'nuclei')) |
| 2998 | except ValueError: |
| 2999 | return jsonify({'success': False, 'error': f"invalid scan_type: {data.get('scan_type')}"}), 400 |
| 3000 | options = data.get('options') or {} |
| 3001 | scan_id = scanner.start_scan(target, scan_type, options) |
| 3002 | logger.info(f"[MESH-SCAN] Accepted delegated {scan_type.value} scan {scan_id} for {target}") |
| 3003 | return jsonify({'success': True, 'scan_id': scan_id}) |
| 3004 | except RuntimeError as e: |
| 3005 | return jsonify({'success': False, 'error': str(e)}), 400 |
| 3006 | except Exception as e: |
| 3007 | logger.error(f"[MESH-SCAN] start failed: {e}") |
| 3008 | return jsonify({'success': False, 'error': str(e)}), 500 |
| 3009 | |
| 3010 | |
| 3011 | @app.route('/api/mesh/scan/status/<scan_id>', methods=['GET']) |
| 3012 | def mesh_scan_status(scan_id): |
| 3013 | """Progress of a delegated scan, for the delegating unit to relay.""" |
| 3014 | scanner = get_advanced_vuln_scanner() |
| 3015 | if not scanner: |
| 3016 | return jsonify({'success': False, 'error': 'Scanner not available'}), 503 |
| 3017 | scan = scanner.get_scan_status(scan_id) |
| 3018 | if not scan: |
| 3019 | return jsonify({'success': False, 'error': 'scan not found'}), 404 |
| 3020 | return jsonify({'success': True, 'scan': scan}) |
| 3021 | |
| 3022 | |
| 3023 | @app.route('/api/mesh/scan/findings/<scan_id>', methods=['GET']) |
| 3024 | def mesh_scan_findings(scan_id): |
| 3025 | """Findings for a delegated scan, pulled home on completion.""" |
| 3026 | scanner = get_advanced_vuln_scanner() |
| 3027 | if not scanner: |
| 3028 | return jsonify({'success': False, 'error': 'Scanner not available'}), 503 |
| 3029 | try: |
| 3030 | with scanner._lock: |
| 3031 | findings = [f.to_dict() for f in scanner.scan_results.get(scan_id, [])] |
no test coverage detected