Nmap and ChatGPT Security Auditing with Wazuh#

1
#!/opt/wazuh-nmap-ai/bin/python
2
import os
3
import sys
4
import json
5
import nmap
6
import openai
7
import xmltodict
8
from datetime import datetime
9
import logging
10

11
# Configure logging
12
logging.basicConfig(
13
    level=logging.INFO,
14
    format='%(asctime)s - %(levelname)s - %(message)s',
15
    filename='/var/ossec/logs/nmap-chatgpt.log'
16
)
17

18
# Configuration
19
OPENAI_API_KEY = os.environ.get('OPENAI_API_KEY', 'your-api-key-here')
20
SCAN_OUTPUT_DIR = '/var/ossec/logs/nmap-scans'
21
ALERT_FILE = '/var/ossec/logs/nmap-ai-alerts.json'
22

23
# Initialize OpenAI
24
openai.api_key = OPENAI_API_KEY
25

26
class NmapChatGPTIntegration:
27
    def __init__(self):
28
        self.nm = nmap.PortScanner()
29

30
    def perform_scan(self, target, scan_type='comprehensive'):
31
        """Perform Nmap scan based on type"""
32
        scan_args = {
33
            'basic': '-sV -sC',
34
            'comprehensive': '-sV -sC -O -A',
35
            'vulnerability': '-sV --script vuln',
36
            'stealth': '-sS -sV -T2 -f',
37
            'quick': '-F -T4'
38
        }
39

40
        try:
41
            logging.info(f"Starting {scan_type} scan on {target}")
42
            self.nm.scan(hosts=target, arguments=scan_args.get(scan_type, '-sV'))
43
            return self.nm
44
        except Exception as e:
45
            logging.error(f"Scan failed: {str(e)}")
46
            return None
47

48
    def parse_scan_results(self, scan_result):
49
        """Parse Nmap results into structured format"""
50
        findings = []
51

52
        for host in scan_result.all_hosts():
53
            host_info = {
54
                'ip': host,
55
                'hostname': scan_result[host].hostname(),
56
                'state': scan_result[host].state(),
57
                'os': self._extract_os_info(scan_result[host]),
58
                'ports': [],
59
                'vulnerabilities': []
60
            }
61

62
            # Extract port information
63
            for proto in scan_result[host].all_protocols():
64
                ports = scan_result[host][proto].keys()
65
                for port in ports:
66
                    port_info = scan_result[host][proto][port]
67
                    host_info['ports'].append({
68
                        'port': port,
69
                        'protocol': proto,
70
                        'state': port_info['state'],
71
                        'service': port_info.get('name', 'unknown'),
72
                        'version': port_info.get('version', ''),
73
                        'product': port_info.get('product', '')
74
                    })
75

76
            # Extract vulnerability scripts results
77
            if 'script' in scan_result[host]:
78
                for script_name, script_output in scan_result[host]['script'].items():
79
                    if 'vuln' in script_name:
80
                        host_info['vulnerabilities'].append({
81
                            'script': script_name,
82
                            'output': script_output
83
                        })
84

85
            findings.append(host_info)
86

87
        return findings
88

89
    def _extract_os_info(self, host_data):
90
        """Extract OS detection information"""
91
        if 'osmatch' in host_data:
92
            os_matches = []
93
            for os_match in host_data['osmatch']:
94
                os_matches.append({
95
                    'name': os_match['name'],
96
                    'accuracy': os_match['accuracy']
97
                })
98
            return os_matches
99
        return []
100

101
    def analyze_with_chatgpt(self, findings):
102
        """Send findings to ChatGPT for analysis"""
103
        # Prepare context for ChatGPT
104
        context = self._prepare_context(findings)
105

106
        try:
107
            response = openai.ChatCompletion.create(
108
                model="gpt-4",
109
                messages=[
110
                    {
111
                        "role": "system",
112
                        "content": """You are a cybersecurity expert analyzing Nmap scan results.
113
                        Provide detailed analysis including:
114
                        1. Security risk assessment (Critical/High/Medium/Low)
115
                        2. Potential vulnerabilities identified
116
                        3. Specific remediation recommendations
117
                        4. Priority order for addressing issues
118
                        5. Additional security considerations
119
                        Format your response as structured JSON."""
120
                    },
121
                    {
122
                        "role": "user",
123
                        "content": f"Analyze these Nmap scan findings:\n{json.dumps(context, indent=2)}"
124
                    }
125
                ],
126
                temperature=0.3,
127
                max_tokens=2000
128
            )
129

130
            analysis = json.loads(response.choices[0].message.content)
131
            return analysis
132

133
        except Exception as e:
134
            logging.error(f"ChatGPT analysis failed: {str(e)}")
135
            return None
136

137
    def _prepare_context(self, findings):
138
        """Prepare findings for ChatGPT analysis"""
139
        context = {
140
            'scan_timestamp': datetime.now().isoformat(),
141
            'total_hosts': len(findings),
142
            'findings': []
143
        }
144

145
        for host in findings:
146
            host_context = {
147
                'ip': host['ip'],
148
                'hostname': host['hostname'],
149
                'open_ports': len(host['ports']),
150
                'services': [],
151
                'potential_issues': []
152
            }
153

154
            # Summarize services
155
            for port in host['ports']:
156
                service_info = f"{port['service']} ({port['version']})" if port['version'] else port['service']
157
                host_context['services'].append({
158
                    'port': port['port'],
159
                    'service': service_info,
160
                    'state': port['state']
161
                })
162

163
            # Add vulnerabilities if found
164
            if host['vulnerabilities']:
165
                host_context['vulnerabilities'] = host['vulnerabilities']
166

167
            # Flag potential issues
168
            for port in host['ports']:
169
                if port['port'] in [21, 23, 445, 3389]:  # Known risky ports
170
                    host_context['potential_issues'].append(
171
                        f"Risky service {port['service']} on port {port['port']}"
172
                    )
173

174
            context['findings'].append(host_context)
175

176
        return context
177

178
    def generate_wazuh_alert(self, findings, ai_analysis):
179
        """Generate Wazuh-compatible alert"""
180
        alert = {
181
            'timestamp': datetime.now().isoformat(),
182
            'rule': {
183
                'level': self._determine_alert_level(ai_analysis),
184
                'description': 'Nmap scan with AI analysis completed',
185
                'id': '100100',
186
                'groups': ['network_scan', 'ai_analysis']
187
            },
188
            'data': {
189
                'scan_summary': {
190
                    'hosts_scanned': len(findings),
191
                    'open_ports_total': sum(len(h['ports']) for h in findings),
192
                    'vulnerabilities_found': sum(len(h['vulnerabilities']) for h in findings)
193
                },
194
                'ai_analysis': ai_analysis,
195
                'detailed_findings': findings
196
            }
197
        }
198

199
        # Write alert to file for Wazuh to pick up
200
        with open(ALERT_FILE, 'a') as f:
201
            f.write(json.dumps(alert) + '\n')
202

203
        return alert
204

205
    def _determine_alert_level(self, ai_analysis):
206
        """Determine Wazuh alert level based on AI analysis"""
207
        if not ai_analysis:
208
            return 5
209

210
        risk_level = ai_analysis.get('overall_risk', 'medium').lower()
211
        level_mapping = {
212
            'critical': 12,
213
            'high': 10,
214
            'medium': 7,
215
            'low': 5
216
        }
217

218
        return level_mapping.get(risk_level, 7)
219

220
def main():
221
    """Main execution function"""
222
    if len(sys.argv) < 2:
223
        print("Usage: nmap-chatgpt.py <target> [scan_type]")
224
        sys.exit(1)
225

226
    target = sys.argv[1]
227
    scan_type = sys.argv[2] if len(sys.argv) > 2 else 'comprehensive'
228

229
    # Initialize integration
230
    integration = NmapChatGPTIntegration()
231

232
    # Perform scan
233
    scan_result = integration.perform_scan(target, scan_type)
234
    if not scan_result:
235
        logging.error("Scan failed to complete")
236
        sys.exit(1)
237

238
    # Parse results
239
    findings = integration.parse_scan_results(scan_result)
240
    logging.info(f"Scan completed. Found {len(findings)} hosts")
241

242
    # Analyze with ChatGPT
243
    ai_analysis = integration.analyze_with_chatgpt(findings)
244
    if ai_analysis:
245
        logging.info("AI analysis completed successfully")
246
    else:
247
        logging.warning("AI analysis failed, proceeding with basic alert")
248

249
    # Generate Wazuh alert
250
    alert = integration.generate_wazuh_alert(findings, ai_analysis)
251
    logging.info(f"Alert generated with level {alert['rule']['level']}")
252

253
    # Output summary
254
    print(json.dumps({
255
        'status': 'success',
256
        'hosts_scanned': len(findings),
257
        'alert_level': alert['rule']['level'],
258
        'ai_analysis_available': ai_analysis is not None
259
    }, indent=2))
260

261
if __name__ == '__main__':
262
    main()

1
class SecurityPrompts:
2
    """Advanced prompts for different security contexts"""
3

4
    @staticmethod
5
    def vulnerability_assessment():
6
        return """You are a vulnerability assessment expert. Analyze the scan results and provide:
7
        1. CVE identification for discovered services
8
        2. CVSS score estimation
9
        3. Exploitation difficulty assessment
10
        4. Potential attack vectors
11
        5. Detailed remediation steps with commands/configurations
12
        6. Compensating controls if immediate patching isn't possible
13

14
        Structure your response as JSON with these sections:
15
        - vulnerabilities: array of identified vulnerabilities
16
        - risk_matrix: risk assessment for each finding
17
        - remediation_plan: prioritized action items
18
        - quick_wins: immediate actions that can be taken
19
        """
20

21
    @staticmethod
22
    def compliance_check():
23
        return """You are a compliance auditor. Analyze the scan results against common frameworks:
24
        1. PCI DSS requirements
25
        2. HIPAA technical safeguards
26
        3. NIST 800-53 controls
27
        4. CIS benchmarks
28

29
        For each applicable framework, identify:
30
        - Compliance gaps
31
        - Required configurations
32
        - Evidence of controls
33
        - Remediation priorities
34

35
        Format as JSON with framework-specific findings."""
36

37
    @staticmethod
38
    def threat_intelligence():
39
        return """You are a threat intelligence analyst. Based on the scan results:
40
        1. Identify services commonly targeted by threat actors
41
        2. Map findings to MITRE ATT&CK techniques
42
        3. Assess likelihood of compromise
43
        4. Recommend threat hunting queries
44
        5. Suggest defensive measures
45

46
        Include IoCs to monitor and detection rules to implement."""

1
from flask import Flask, jsonify, request
2
import json
3
import subprocess
4

5
app = Flask(__name__)
6

7
@app.route('/api/scan/trigger', methods=['POST'])
8
def trigger_scan():
9
    """API endpoint to trigger on-demand scans"""
10
    data = request.json
11
    target = data.get('target')
12
    scan_type = data.get('scan_type', 'basic')
13

14
    # Validate input
15
    if not target:
16
        return jsonify({'error': 'Target required'}), 400
17

18
    # Execute scan
19
    result = subprocess.run(
20
        ['/opt/wazuh-nmap-ai/bin/python', '/var/ossec/integrations/nmap-chatgpt.py', target, scan_type],
21
        capture_output=True,
22
        text=True
23
    )
24

25
    return jsonify({
26
        'status': 'success' if result.returncode == 0 else 'failed',
27
        'output': result.stdout,
28
        'error': result.stderr
29
    })
30

31
@app.route('/api/scan/history', methods=['GET'])
32
def scan_history():
33
    """Get recent scan results"""
34
    limit = request.args.get('limit', 10, type=int)
35

36
    # Read recent alerts
37
    alerts = []
38
    with open('/var/ossec/logs/nmap-ai-alerts.json', 'r') as f:
39
        for line in f:
40
            alerts.append(json.loads(line))
41
            if len(alerts) >= limit:
42
                break
43

44
    return jsonify(alerts)
45

46
@app.route('/api/scan/analysis/<scan_id>', methods=['GET'])
47
def get_analysis(scan_id):
48
    """Get detailed AI analysis for a specific scan"""
49
    # Implementation to retrieve specific scan analysis
50
    pass
51

52
if __name__ == '__main__':
53
    app.run(host='0.0.0.0', port=5000)

1
{
2
  "visualization": {
3
    "title": "AI-Enhanced Network Security Overview",
4
    "visState": {
5
      "type": "vega",
6
      "params": {
7
        "spec": {
8
          "$schema": "https://vega.github.io/schema/vega-lite/v5.json",
9
          "data": {
10
            "url": {
11
              "index": "wazuh-alerts-*",
12
              "body": {
13
                "query": {
14
                  "bool": {
15
                    "must": [
16
                      {"term": {"rule.groups": "ai_analysis"}},
17
                      {"range": {"timestamp": {"gte": "now-7d"}}}
18
                    ]
19
                  }
20
                },
21
                "aggs": {
22
                  "risk_distribution": {
23
                    "terms": {
24
                      "field": "data.ai_analysis.overall_risk.keyword"
25
                    }
26
                  },
27
                  "timeline": {
28
                    "date_histogram": {
29
                      "field": "timestamp",
30
                      "interval": "1d"
31
                    },
32
                    "aggs": {
33
                      "risk_levels": {
34
                        "terms": {
35
                          "field": "data.ai_analysis.overall_risk.keyword"
36
                        }
37
                      }
38
                    }
39
                  }
40
                }
41
              }
42
            }
43
          },
44
          "layer": [
45
            {
46
              "mark": "bar",
47
              "encoding": {
48
                "x": {"field": "key", "type": "temporal"},
49
                "y": {"field": "doc_count", "type": "quantitative"},
50
                "color": {"field": "risk_levels.buckets.key", "type": "nominal"}
51
              }
52
            }
53
          ]
54
        }
55
      }
56
    }
57
  }
58
}

1
# Secure API key storage
2
import keyring
3
from cryptography.fernet import Fernet
4

5
class SecureConfig:
6
    def __init__(self):
7
        self.key = self._get_or_create_key()
8
        self.cipher = Fernet(self.key)
9

10
    def _get_or_create_key(self):
11
        """Get or create encryption key"""
12
        key = keyring.get_password("wazuh-nmap-ai", "encryption_key")
13
        if not key:
14
            key = Fernet.generate_key().decode()
15
            keyring.set_password("wazuh-nmap-ai", "encryption_key", key)
16
        return key.encode()
17

18
    def store_api_key(self, api_key):
19
        """Securely store API key"""
20
        encrypted = self.cipher.encrypt(api_key.encode())
21
        keyring.set_password("wazuh-nmap-ai", "openai_api_key", encrypted.decode())
22

23
    def get_api_key(self):
24
        """Retrieve API key"""
25
        encrypted = keyring.get_password("wazuh-nmap-ai", "openai_api_key")
26
        if encrypted:
27
            return self.cipher.decrypt(encrypted.encode()).decode()
28
        return None

1
class RateLimiter:
2
    def __init__(self, max_requests_per_hour=10):
3
        self.max_requests = max_requests_per_hour
4
        self.requests = []
5

6
    def can_make_request(self):
7
        """Check if request can be made within rate limits"""
8
        now = datetime.now()
9
        # Remove requests older than 1 hour
10
        self.requests = [r for r in self.requests if (now - r).seconds < 3600]
11

12
        if len(self.requests) < self.max_requests:
13
            self.requests.append(now)
14
            return True
15
        return False
16

17
    def estimate_cost(self, tokens_used):
18
        """Estimate API cost"""
19
        # GPT-4 pricing (example)
20
        cost_per_1k_tokens = 0.03
21
        return (tokens_used / 1000) * cost_per_1k_tokens

1
import ipaddress
2
import re
3

4
def validate_scan_target(target):
5
    """Validate scan target is safe"""
6
    # Check if it's a valid IP or CIDR
7
    try:
8
        ipaddress.ip_network(target, strict=False)
9
    except ValueError:
10
        # Check if it's a valid hostname
11
        hostname_regex = re.compile(
12
            r'^(?!-)(?:[a-zA-Z0-9-]{1,63}(?<!-)\.)*[a-zA-Z]{2,}$'
13
        )
14
        if not hostname_regex.match(target):
15
            raise ValueError(f"Invalid target: {target}")
16

17
    # Check against blocked ranges
18
    blocked_ranges = [
19
        '127.0.0.0/8',  # Loopback
20
        '169.254.0.0/16',  # Link-local
21
        '224.0.0.0/4'  # Multicast
22
    ]
23

24
    for blocked in blocked_ranges:
25
        if ipaddress.ip_network(target, strict=False).overlaps(
26
            ipaddress.ip_network(blocked)
27
        ):
28
            raise ValueError(f"Target in blocked range: {target}")
29

30
    return True

1
#!/bin/bash
2
# Monitor integration performance
3

4
echo "=== Nmap-ChatGPT Integration Monitor ==="
5
echo "Timestamp: $(date)"
6
echo ""
7

8
# Check API usage
9
echo "API Requests (Last 24h):"
10
grep "ChatGPT analysis completed" /var/ossec/logs/nmap-chatgpt.log | \
11
    grep "$(date -d '1 day ago' +%Y-%m-%d)" | wc -l
12

13
# Average scan time
14
echo -e "\nAverage Scan Duration:"
15
grep "Scan completed" /var/ossec/logs/nmap-chatgpt.log | \
16
    tail -10 | awk '{print $1}' | \
17
    while read timestamp; do
18
        date -d "$timestamp" +%s
19
    done | awk '{if(NR>1)print $1-prev; prev=$1}' | \
20
    awk '{sum+=$1; count++} END {print sum/count " seconds"}'
21

22
# Error rate
23
echo -e "\nError Rate:"
24
total=$(grep -c "Starting" /var/ossec/logs/nmap-chatgpt.log)
25
errors=$(grep -c "ERROR" /var/ossec/logs/nmap-chatgpt.log)
26
echo "$errors errors out of $total scans"
27

28
# Wazuh alerts generated
29
echo -e "\nAlerts Generated:"
30
wc -l /var/ossec/logs/nmap-ai-alerts.json

1
def vulnerability_focused_scan(target):
2
    """Specialized scan for vulnerability assessment"""
3
    scan_args = """
4
    -sV --script=vuln,exploit,auth,default
5
    --script-args=unsafe=1
6
    -p-
7
    """
8

9
    # Custom AI prompt for vulnerability analysis
10
    vuln_prompt = """
11
    Focus on identifying:
12
    1. Known CVEs for detected services
13
    2. Default credentials
14
    3. Misconfigurations
15
    4. Outdated software versions
16
    5. Weak encryption settings
17

18
    Provide specific exploitation steps and patches.
19
    """
20

21
    return enhanced_scan(target, scan_args, vuln_prompt)

1
def compliance_scan(target, framework='pci'):
2
    """Scan for compliance framework requirements"""
3
    frameworks = {
4
        'pci': {
5
            'ports': '21,22,23,25,80,443,445,1433,3306,3389',
6
            'scripts': 'ssl-cert,ssl-enum-ciphers,http-security-headers'
7
        },
8
        'hipaa': {
9
            'ports': '22,443,3389,5432,27017',
10
            'scripts': 'ssl-*,http-security-headers,smb-security-mode'
11
        }
12
    }
13

14
    config = frameworks.get(framework, frameworks['pci'])
15
    scan_args = f"-sV -p{config['ports']} --script={config['scripts']}"
16

17
    return targeted_scan(target, scan_args, framework)

1
def differential_scan(target, baseline_file):
2
    """Compare current state with baseline"""
3
    current_scan = perform_scan(target)
4

5
    # Load baseline
6
    with open(baseline_file, 'r') as f:
7
        baseline = json.load(f)
8

9
    # Identify changes
10
    changes = {
11
        'new_ports': [],
12
        'closed_ports': [],
13
        'service_changes': [],
14
        'new_vulnerabilities': []
15
    }
16

17
    # Compare and analyze changes
18
    for host in current_scan['hosts']:
19
        baseline_host = find_host_in_baseline(host['ip'], baseline)
20
        if baseline_host:
21
            changes['new_ports'].extend(
22
                find_new_ports(host['ports'], baseline_host['ports'])
23
            )
24

25
    # Get AI analysis of changes
26
    change_analysis = analyze_changes_with_ai(changes)
27

28
    return changes, change_analysis

1
import requests
2

3
def send_slack_alert(webhook_url, alert_data):
4
    """Send critical alerts to Slack"""
5
    if alert_data['risk_level'] in ['critical', 'high']:
6
        message = {
7
            "text": f"🚨 Security Alert: {alert_data['risk_level'].upper()}",
8
            "attachments": [{
9
                "color": "danger" if alert_data['risk_level'] == 'critical' else "warning",
10
                "fields": [
11
                    {
12
                        "title": "Target",
13
                        "value": alert_data['target'],
14
                        "short": True
15
                    },
16
                    {
17
                        "title": "Open Ports",
18
                        "value": str(alert_data['open_ports']),
19
                        "short": True
20
                    },
21
                    {
22
                        "title": "AI Summary",
23
                        "value": alert_data['ai_summary'],
24
                        "short": False
25
                    },
26
                    {
27
                        "title": "Top Recommendations",
28
                        "value": "\n".join(alert_data['recommendations'][:3]),
29
                        "short": False
30
                    }
31
                ]
32
            }]
33
        }
34

35
        requests.post(webhook_url, json=message)

1
from jira import JIRA
2

3
def create_jira_issue(findings, ai_analysis):
4
    """Create JIRA tickets for findings"""
5
    jira = JIRA(
6
        server='https://your-instance.atlassian.net',
7
        basic_auth=('email@example.com', 'api_token')
8
    )
9

10
    for finding in findings['critical_issues']:
11
        issue_dict = {
12
            'project': 'SEC',
13
            'summary': f"Security Finding: {finding['title']}",
14
            'description': f"""
15
                *AI Risk Assessment:* {finding['risk_level']}
16

17
                *Description:* {finding['description']}
18

19
                *Affected System:* {finding['target']}
20

21
                *Remediation Steps:*
22
                {finding['remediation']}
23

24
                *AI Recommendations:*
25
                {ai_analysis['recommendations']}
26
            """,
27
            'issuetype': {'name': 'Security'},
28
            'priority': {'name': 'High' if finding['risk_level'] == 'critical' else 'Medium'}
29
        }
30

31
        jira.create_issue(fields=issue_dict)

1
import hashlib
2
import pickle
3

4
class AICache:
5
    def __init__(self, cache_dir='/var/ossec/cache/ai'):
6
        self.cache_dir = cache_dir
7
        os.makedirs(cache_dir, exist_ok=True)
8

9
    def get_cache_key(self, findings):
10
        """Generate cache key from findings"""
11
        # Create hash of relevant findings data
12
        cache_data = {
13
            'services': sorted([p['service'] for h in findings for p in h['ports']]),
14
            'port_count': sum(len(h['ports']) for h in findings),
15
            'os_matches': [os['name'] for h in findings for os in h.get('os', [])]
16
        }
17

18
        return hashlib.sha256(
19
            json.dumps(cache_data, sort_keys=True).encode()
20
        ).hexdigest()
21

22
    def get(self, findings):
23
        """Retrieve cached analysis"""
24
        cache_key = self.get_cache_key(findings)
25
        cache_file = os.path.join(self.cache_dir, f"{cache_key}.pkl")
26

27
        if os.path.exists(cache_file):
28
            # Check if cache is still valid (24 hours)
29
            if time.time() - os.path.getmtime(cache_file) < 86400:
30
                with open(cache_file, 'rb') as f:
31
                    return pickle.load(f)
32

33
        return None
34

35
    def set(self, findings, analysis):
36
        """Cache analysis results"""
37
        cache_key = self.get_cache_key(findings)
38
        cache_file = os.path.join(self.cache_dir, f"{cache_key}.pkl")
39

40
        with open(cache_file, 'wb') as f:
41
            pickle.dump(analysis, f)

1
from concurrent.futures import ThreadPoolExecutor, as_completed
2

3
def parallel_scan(targets, max_workers=5):
4
    """Scan multiple targets in parallel"""
5
    results = []
6

7
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
8
        # Submit scan tasks
9
        future_to_target = {
10
            executor.submit(perform_scan, target): target
11
            for target in targets
12
        }
13

14
        # Process completed scans
15
        for future in as_completed(future_to_target):
16
            target = future_to_target[future]
17
            try:
18
                scan_result = future.result()
19
                # Process with AI in batches to optimize API usage
20
                results.append((target, scan_result))
21
            except Exception as e:
22
                logging.error(f"Scan failed for {target}: {str(e)}")
23

24
    # Batch AI analysis
25
    if results:
26
        batch_analyze_with_ai(results)
27

28
    return results