#!/usr/bin/env python3 """ NinjaOne CVE Comprehensive Report Generates detailed breakdown by category, organization, and device. """ import requests from collections import defaultdict from datetime import datetime import csv import json try: from config import NINJAONE_CONFIG except ImportError: print("Error: Copy config.example.py to config.py and add your credentials") exit(1) class NinjaOneReporter: """Generate comprehensive CVE reports from NinjaOne""" def __init__(self): self.config = NINJAONE_CONFIG self.access_token = None self.organizations = {} self.devices = {} def authenticate(self): """Get OAuth token""" response = requests.post( self.config['token_url'], headers={'Content-Type': 'application/x-www-form-urlencoded'}, data={ 'grant_type': 'client_credentials', 'client_id': self.config['client_id'], 'client_secret': self.config['client_secret'], 'scope': self.config['scope'] } ) response.raise_for_status() self.access_token = response.json()['access_token'] print("Authenticated successfully") def api_get(self, endpoint, params=None): """Make API GET request""" headers = {'Authorization': f'Bearer {self.access_token}'} url = f"{self.config['base_url']}/api/v2{endpoint}" response = requests.get(url, headers=headers, params=params) response.raise_for_status() return response.json() def load_organizations(self): """Load all organizations into lookup dict""" print("Loading organizations...") orgs = self.api_get('/organizations') for org in orgs: self.organizations[org['id']] = org['name'] print(f" Loaded {len(self.organizations)} organizations") def load_devices(self): """Load all devices into lookup dict""" print("Loading devices...") devices = self.api_get('/devices-detailed') for device in devices: self.devices[device['id']] = { 'name': device.get('systemName') or device.get('dnsName') or f"Device-{device['id']}", 'org_id': device.get('organizationId'), 'org_name': self.organizations.get(device.get('organizationId'), 'Unknown'), 'os': device.get('os', {}).get('name', 'Unknown') if isinstance(device.get('os'), dict) else 'Unknown' } print(f" Loaded {len(self.devices)} devices") def get_all_patches(self): """Get all OS and software patches""" print("Loading OS patches...") os_result = self.api_get('/queries/os-patches') os_patches = os_result.get('results', []) if isinstance(os_result, dict) else os_result print("Loading software patches...") sw_result = self.api_get('/queries/software-patches') sw_patches = sw_result.get('results', []) if isinstance(sw_result, dict) else sw_result print(f" Found {len(os_patches)} OS patches, {len(sw_patches)} software patches") return os_patches, sw_patches def categorize_patch(self, patch, patch_type): """Categorize a patch by its source/type""" title = patch.get('title', '').lower() if patch_type == 'os': return 'Windows/OS Updates' # Application categories if any(x in title for x in ['chrome', 'firefox', 'edge', 'opera', 'safari']): return 'Web Browsers' elif any(x in title for x in ['adobe', 'acrobat', 'reader', 'flash']): return 'Adobe Products' elif any(x in title for x in ['java', 'jre', 'jdk', 'corretto']): return 'Java/Runtime' elif any(x in title for x in ['office', '365', 'word', 'excel', 'outlook']): return 'Microsoft Office' elif any(x in title for x in ['zoom', 'teams', 'webex', 'slack', 'gotomeeting']): return 'Communication Apps' elif any(x in title for x in ['vnc', 'tightvnc', 'ultravnc', 'teamviewer', 'anydesk']): return 'Remote Access Tools' elif any(x in title for x in ['7-zip', 'winrar', 'winzip']): return 'Compression Tools' elif any(x in title for x in ['vlc', 'media', 'player']): return 'Media Players' elif any(x in title for x in ['notepad', 'sublime', 'vscode', 'visual studio code']): return 'Text Editors/IDEs' elif any(x in title for x in ['putty', 'filezilla', 'winscp']): return 'Network/Transfer Tools' elif any(x in title for x in ['keepass', 'password']): return 'Password Managers' elif any(x in title for x in ['foxit', 'pdf']): return 'PDF Tools' elif any(x in title for x in ['libre', 'open office', 'openoffice']): return 'Office Alternatives' else: return 'Other Applications' def generate_report(self): """Generate comprehensive report""" self.authenticate() self.load_organizations() self.load_devices() os_patches, sw_patches = self.get_all_patches() # Data structures for analysis by_category = defaultdict(lambda: {'total': 0, 'critical': 0, 'devices': set()}) by_org = defaultdict(lambda: {'total': 0, 'critical': 0, 'os': 0, 'software': 0, 'devices': set()}) critical_items = [] device_details = [] # Process OS patches # OS patches use 'name' and 'severity' fields for patch in os_patches: device_id = patch.get('deviceId') device = self.devices.get(device_id, {}) org_name = device.get('org_name', 'Unknown') # OS patches use 'severity' not 'impact' impact = patch.get('severity', patch.get('impact', 'unknown')).lower() # OS patches use 'name' not 'title' title = patch.get('name', patch.get('title', 'Unknown')) category = 'Windows/OS Updates' by_category[category]['total'] += 1 by_category[category]['devices'].add(device_id) by_org[org_name]['total'] += 1 by_org[org_name]['os'] += 1 by_org[org_name]['devices'].add(device_id) if impact == 'critical': by_category[category]['critical'] += 1 by_org[org_name]['critical'] += 1 critical_items.append({ 'type': 'OS', 'title': title, 'device_id': device_id, 'device_name': device.get('name', 'Unknown'), 'org': org_name, 'status': patch.get('status', 'Unknown') }) device_details.append({ 'type': 'OS', 'category': category, 'title': title, 'impact': impact, 'status': patch.get('status', 'Unknown'), 'device_id': device_id, 'device_name': device.get('name', 'Unknown'), 'organization': org_name }) # Process software patches for patch in sw_patches: device_id = patch.get('deviceId') device = self.devices.get(device_id, {}) org_name = device.get('org_name', 'Unknown') impact = patch.get('impact', 'unknown').lower() category = self.categorize_patch(patch, 'software') by_category[category]['total'] += 1 by_category[category]['devices'].add(device_id) by_org[org_name]['total'] += 1 by_org[org_name]['software'] += 1 by_org[org_name]['devices'].add(device_id) if impact == 'critical': by_category[category]['critical'] += 1 by_org[org_name]['critical'] += 1 critical_items.append({ 'type': 'Software', 'title': patch.get('title', 'Unknown'), 'device_id': device_id, 'device_name': device.get('name', 'Unknown'), 'org': org_name, 'status': patch.get('status', 'Unknown') }) device_details.append({ 'type': 'Software', 'category': category, 'title': patch.get('title', 'Unknown'), 'impact': impact, 'status': patch.get('status', 'Unknown'), 'device_id': device_id, 'device_name': device.get('name', 'Unknown'), 'organization': org_name }) # Generate text report timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') report = [] report.append("=" * 70) report.append("NINJAONE CVE COMPREHENSIVE REPORT") report.append(f"Generated: {timestamp}") report.append("=" * 70) report.append(f"\nTOTAL ORGANIZATIONS: {len(self.organizations)}") report.append(f"TOTAL DEVICES: {len(self.devices)}") report.append(f"TOTAL OS PATCHES: {len(os_patches)}") report.append(f"TOTAL SOFTWARE PATCHES: {len(sw_patches)}") report.append(f"TOTAL CRITICAL ITEMS: {len(critical_items)}") # Section 1: By Category report.append("\n" + "=" * 70) report.append("SECTION 1: PATCHES BY CATEGORY (Most Common CVE Causes)") report.append("=" * 70) sorted_categories = sorted(by_category.items(), key=lambda x: x[1]['total'], reverse=True) report.append(f"\n{'Category':<30} {'Total':>8} {'Critical':>10} {'Devices':>10}") report.append("-" * 60) for category, data in sorted_categories: report.append(f"{category:<30} {data['total']:>8} {data['critical']:>10} {len(data['devices']):>10}") # Section 2: By Organization report.append("\n" + "=" * 70) report.append("SECTION 2: PATCHES BY ORGANIZATION") report.append("=" * 70) sorted_orgs = sorted(by_org.items(), key=lambda x: x[1]['total'], reverse=True) report.append(f"\n{'Organization':<35} {'Total':>7} {'Crit':>6} {'OS':>6} {'SW':>6} {'Devices':>8}") report.append("-" * 70) for org, data in sorted_orgs: org_display = org[:33] + '..' if len(org) > 35 else org report.append(f"{org_display:<35} {data['total']:>7} {data['critical']:>6} {data['os']:>6} {data['software']:>6} {len(data['devices']):>8}") # Section 3: Critical Items with Device Details report.append("\n" + "=" * 70) report.append("SECTION 3: CRITICAL ITEMS (Requires Immediate Attention)") report.append("=" * 70) if critical_items: # Group by title critical_by_title = defaultdict(list) for item in critical_items: critical_by_title[item['title']].append(item) for title, items in sorted(critical_by_title.items(), key=lambda x: len(x[1]), reverse=True): report.append(f"\n {title} ({len(items)} devices)") for item in items[:10]: # Show first 10 devices per title report.append(f" - {item['device_name']} ({item['org']}) - {item['status']}") if len(items) > 10: report.append(f" ... and {len(items) - 10} more devices") else: report.append("\n No critical items found") # Section 4: Patches Needing Approval (MANUAL status) report.append("\n" + "=" * 70) report.append("SECTION 4: PATCHES NEEDING APPROVAL (Manual Policy Review)") report.append("=" * 70) manual_patches = [d for d in device_details if d['status'] == 'MANUAL'] if manual_patches: # Group by organization, then device manual_by_org = defaultdict(lambda: defaultdict(list)) for patch in manual_patches: manual_by_org[patch['organization']][patch['device_name']].append(patch) report.append(f"\n Total patches needing approval: {len(manual_patches)}") report.append(f" Across {len(manual_by_org)} organizations\n") for org in sorted(manual_by_org.keys()): devices = manual_by_org[org] total_patches = sum(len(patches) for patches in devices.values()) report.append(f"\n {org} ({total_patches} patches, {len(devices)} devices)") report.append(" " + "-" * 50) for device_name in sorted(devices.keys()): patches = devices[device_name] report.append(f" {device_name}:") for patch in patches: impact_marker = "[CRIT]" if patch['impact'] == 'critical' else "[" + patch['impact'][:4].upper() + "]" report.append(f" {impact_marker} {patch['title']}") else: report.append("\n No patches pending manual approval") # Section 5: Devices needing attention (most patches) report.append("\n" + "=" * 70) report.append("SECTION 5: DEVICES NEEDING MOST ATTENTION") report.append("=" * 70) device_patch_count = defaultdict(lambda: {'total': 0, 'critical': 0, 'name': '', 'org': ''}) for detail in device_details: did = detail['device_id'] device_patch_count[did]['total'] += 1 device_patch_count[did]['name'] = detail['device_name'] device_patch_count[did]['org'] = detail['organization'] if detail['impact'] == 'critical': device_patch_count[did]['critical'] += 1 sorted_devices = sorted(device_patch_count.items(), key=lambda x: (x[1]['critical'], x[1]['total']), reverse=True) report.append(f"\n{'Device Name':<30} {'Organization':<25} {'Total':>7} {'Critical':>10}") report.append("-" * 75) for device_id, data in sorted_devices[:30]: # Top 30 devices name = data['name'][:28] + '..' if len(data['name']) > 30 else data['name'] org = data['org'][:23] + '..' if len(data['org']) > 25 else data['org'] report.append(f"{name:<30} {org:<25} {data['total']:>7} {data['critical']:>10}") report_text = "\n".join(report) # Save text report report_file = f"cve_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" with open(report_file, 'w') as f: f.write(report_text) print(f"\nText report saved to: {report_file}") # Save CSV for detailed analysis csv_file = f"cve_details_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" with open(csv_file, 'w', newline='') as f: writer = csv.DictWriter(f, fieldnames=['type', 'category', 'title', 'impact', 'status', 'device_id', 'device_name', 'organization']) writer.writeheader() writer.writerows(device_details) print(f"CSV details saved to: {csv_file}") # Generate HTML report html_file = f"cve_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.html" html_content = self.generate_html_report( timestamp, sorted_categories, sorted_orgs, critical_by_title, sorted_devices, os_patches, sw_patches, critical_items, manual_by_org ) with open(html_file, 'w') as f: f.write(html_content) print(f"HTML report saved to: {html_file}") print(report_text) return report_text def generate_html_report(self, timestamp, sorted_categories, sorted_orgs, critical_by_title, sorted_devices, os_patches, sw_patches, critical_items, manual_by_org=None): """Generate HTML report""" html = f''' CVE Report - {timestamp}

NinjaOne CVE Report

Generated: {timestamp}

Organizations

{len(self.organizations)}

Total Devices

{len(self.devices)}

OS Patches

{len(os_patches)}

Software Patches

{len(sw_patches)}

Critical Items

{len(critical_items)}

Pending Approval

{sum(len(patches) for devices in (manual_by_org or {}).values() for patches in devices.values())}

Patches by Category

Understanding what types of software are causing the most CVE exposure:

''' # Category chart max_total = max(d['total'] for _, d in sorted_categories) if sorted_categories else 1 for category, data in sorted_categories: pct = (data['total'] / max_total) * 100 has_critical = 'has-critical' if data['critical'] > 0 else '' critical_text = f" ({data['critical']} critical)" if data['critical'] > 0 else "" html += f'''
{category}
{data['total']}{critical_text}
{len(data['devices'])} dev
''' html += '''

Patches by Organization

''' for org, data in sorted_orgs[:20]: critical_badge = f'{data["critical"]}' if data['critical'] > 0 else '0' html += f''' ''' html += '''
Organization Total Critical OS Software Devices
{org} {data['total']} {critical_badge} {data['os']} {data['software']} {len(data['devices'])}

Critical Items Requiring Immediate Attention

''' if critical_by_title: html += '' for title, items in sorted(critical_by_title.items(), key=lambda x: len(x[1]), reverse=True): device_list = ', '.join([f"{i['device_name']}" for i in items[:5]]) if len(items) > 5: device_list += f' ... +{len(items)-5} more' orgs = list(set([i['org'] for i in items])) html += f''' ''' html += '
SoftwareAffected DevicesDetails
{title} {len(items)} devices {device_list}
Orgs: {", ".join(orgs[:3])}
' else: html += '

No critical items found.

' html += '''

Patches Needing Approval

These patches require manual policy approval before they can be deployed:

''' if manual_by_org: total_manual = sum( len(patches) for devices in manual_by_org.values() for patches in devices.values() ) html += f'

Total: {total_manual} patches across {len(manual_by_org)} organizations

' html += '' for org in sorted(manual_by_org.keys()): devices = manual_by_org[org] first_org = True for device_name in sorted(devices.keys()): patches = devices[device_name] patch_list = '
'.join([ f'{p["impact"][:4].upper()} {p["title"]}' for p in patches ]) org_cell = f'' if first_org else '' first_org = False html += f''' {org_cell} ''' html += '
OrganizationDevicePatches Needing Approval
{org}
{device_name} {patch_list}
' else: html += '

No patches pending manual approval.

' html += '''

Devices Needing Most Attention

''' for device_id, data in sorted_devices[:25]: critical_badge = f'{data["critical"]}' if data['critical'] > 0 else '0' html += f''' ''' html += '''
Device Organization Total Patches Critical
{data['name']} {data['org']} {data['total']} {critical_badge}
''' return html def apply_patches(self, dry_run=True, patch_type='software'): """Trigger patch apply on devices with critical patches""" self.authenticate() self.load_organizations() self.load_devices() print(f"\nLoading {patch_type} patches...") if patch_type == 'software': result = self.api_get('/queries/software-patches') else: result = self.api_get('/queries/os-patches') patches = result.get('results', []) if isinstance(result, dict) else result # Filter to critical patches critical_patches = [ p for p in patches if p.get('impact', '').lower() == 'critical' ] # Get unique devices with critical patches devices_to_patch = {} for patch in critical_patches: device_id = patch.get('deviceId') if device_id not in devices_to_patch: devices_to_patch[device_id] = { 'name': self.devices.get(device_id, {}).get('name', f"Device-{device_id}"), 'org': self.devices.get(device_id, {}).get('org_name', 'Unknown'), 'patches': [] } devices_to_patch[device_id]['patches'].append(patch.get('title', 'Unknown')) print(f"\nFound {len(critical_patches)} critical patches on {len(devices_to_patch)} devices") # Show status breakdown from collections import Counter statuses = Counter(p.get('status') for p in critical_patches) print("\nPatch status breakdown:") for status, count in statuses.most_common(): print(f" {status}: {count}") if not devices_to_patch: print("No devices need patching.") return print("\nDevices to patch:") print("-" * 60) for device_id, info in sorted(devices_to_patch.items(), key=lambda x: len(x[1]['patches']), reverse=True): print(f" {info['name']} ({info['org']}): {len(info['patches'])} critical patches") for title in info['patches'][:3]: print(f" - {title}") if len(info['patches']) > 3: print(f" ... and {len(info['patches']) - 3} more") if dry_run: print("\n" + "=" * 60) print("DRY RUN - No changes made") print("Run with --apply to trigger patch installation") print("=" * 60) return # Trigger patch apply on each device print("\n" + "=" * 60) print("TRIGGERING PATCH APPLY...") print("=" * 60) success = 0 failed = 0 for device_id, info in devices_to_patch.items(): try: headers = {'Authorization': f'Bearer {self.access_token}', 'Content-Type': 'application/json'} url = f"{self.config['base_url']}/api/v2/device/{device_id}/patch/{patch_type}/apply" response = requests.post(url, headers=headers) if response.status_code in [200, 201, 204]: print(f" Triggered: {info['name']} ({len(info['patches'])} patches)") success += 1 else: print(f" Failed: {info['name']} - {response.status_code}: {response.text[:100]}") failed += 1 except Exception as e: print(f" Error: {info['name']} - {str(e)}") failed += 1 print("\n" + "=" * 60) print(f"COMPLETE: {success} devices triggered, {failed} failed") print("=" * 60) print("\nPatch installation has been triggered on these devices.") print("Check NinjaOne dashboard for progress.") def scan_devices(self, device_ids=None): """Trigger patch scan on devices""" self.authenticate() self.load_organizations() self.load_devices() if device_ids is None: # Get devices with critical patches print("\nLoading software patches...") result = self.api_get('/queries/software-patches') patches = result.get('results', []) if isinstance(result, dict) else result critical = [p for p in patches if p.get('impact', '').lower() == 'critical'] device_ids = list(set(p.get('deviceId') for p in critical)) print(f"\nTriggering scan on {len(device_ids)} devices...") print("=" * 60) success = 0 failed = 0 for device_id in device_ids: info = self.devices.get(device_id, {}) name = info.get('name', f"Device-{device_id}") try: headers = {'Authorization': f'Bearer {self.access_token}', 'Content-Type': 'application/json'} url = f"{self.config['base_url']}/api/v2/device/{device_id}/patch/software/scan" response = requests.post(url, headers=headers) if response.status_code in [200, 201, 204]: print(f" Scan triggered: {name}") success += 1 else: print(f" Failed: {name} - {response.status_code}: {response.text[:80]}") failed += 1 except Exception as e: print(f" Error: {name} - {str(e)}") failed += 1 print("=" * 60) print(f"COMPLETE: {success} scans triggered, {failed} failed") if __name__ == '__main__': import sys reporter = NinjaOneReporter() if '--apply' in sys.argv: # Trigger patch apply on devices reporter.apply_patches(dry_run=False) elif '--scan' in sys.argv: # Trigger patch scan on devices reporter.scan_devices() elif '--remediate' in sys.argv: # Dry run - show what would be patched reporter.apply_patches(dry_run=True) else: # Default: generate report reporter.generate_report()