Nuclei Scanner Monitoring และ Alerting
Nuclei Scanner Monitoring และ Alerting คืออะไร
Nuclei เป็น open source vulnerability scanner จาก ProjectDiscovery ที่ใช้ template-based scanning สำหรับตรวจหา vulnerabilities, misconfigurations และ security issues ในเว็บแอปพลิเคชัน APIs และ infrastructure รองรับ templates กว่า 8,000+ รายการ ครอบคลุม CVEs, OWASP Top 10, default credentials และ exposures Monitoring และ Alerting คือการตั้งระบบสแกนอัตโนมัติและแจ้งเตือนเมื่อพบ vulnerabilities ใหม่ เพื่อให้ทีม security ตอบสนองได้รวดเร็ว
Nuclei พื้นฐาน
# nuclei_basics.py — Nuclei scanner fundamentals
import json
class NucleiBasics:
FEATURES = {
"template_based": {
"name": "Template-Based Scanning",
"description": "ใช้ YAML templates กำหนด scan logic (ไม่ใช่ signature)",
"templates": "8,000+ community templates + custom templates",
},
"fast": {
"name": "High Performance",
"description": "เขียนด้วย Go, สแกนเร็วมาก, รองรับ concurrent requests",
"benchmark": "สแกน 1,000+ hosts ใน minutes",
},
"protocols": {
"name": "Multi-Protocol",
"description": "HTTP, DNS, TCP, SSL, File, Code, Headless browser",
"use": "ตรวจได้ทั้ง web, network, DNS misconfig",
},
"severity": {
"name": "Severity Levels",
"description": "critical, high, medium, low, info",
"use": "filter และ prioritize findings",
},
}
INSTALL = """
# Installation
go install -v github.com/projectdiscovery/nuclei/v3/cmd/nuclei@latest
# Or via Docker
docker pull projectdiscovery/nuclei:latest
# Update templates
nuclei -update-templates
# Basic scan
nuclei -u https://example.com -severity critical, high
# Scan multiple targets
nuclei -l targets.txt -severity critical, high, medium -o results.txt
# Scan with specific templates
nuclei -u https://example.com -t cves/ -t misconfigurations/
# JSON output for automation
nuclei -u https://example.com -jsonl -o results.jsonl
"""
TEMPLATE_EXAMPLE = """
# custom-template.yaml — Custom Nuclei template
id: admin-panel-exposed
info:
name: Admin Panel Exposed
author: security-team
severity: high
description: Admin panel is publicly accessible
tags: exposure, admin
http:
- method: GET
path:
- "{{BaseURL}}/admin"
- "{{BaseURL}}/admin/login"
- "{{BaseURL}}/wp-admin"
- "{{BaseURL}}/administrator"
matchers-condition: or
matchers:
- type: word
words:
- "admin"
- "login"
- "dashboard"
condition: or
- type: status
status:
- 200
- 302
"""
def show_features(self):
print("=== Nuclei Features ===\n")
for key, feature in self.FEATURES.items():
print(f"[{feature['name']}]")
print(f" {feature['description']}")
print()
def show_install(self):
print("=== Installation & Usage ===")
print(self.INSTALL[:400])
def show_template(self):
print(f"\n=== Custom Template ===")
print(self.TEMPLATE_EXAMPLE[:400])
nuclei = NucleiBasics()
nuclei.show_features()
nuclei.show_install()
nuclei.show_template()Automated Scanning Pipeline
# pipeline.py — Automated Nuclei scanning pipeline
import json
class ScanPipeline:
PYTHON_SCANNER = """
# nuclei_scanner.py — Automated scanning with Python
import subprocess
import json
import os
from datetime import datetime
class NucleiScanner:
def __init__(self, nuclei_path="nuclei"):
self.nuclei_path = nuclei_path
self.results_dir = "scan_results"
os.makedirs(self.results_dir, exist_ok=True)
def scan(self, targets, severity="critical, high, medium",
templates=None, rate_limit=150):
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_file = f"{self.results_dir}/scan_{timestamp}.jsonl"
cmd = [
self.nuclei_path,
"-severity", severity,
"-rate-limit", str(rate_limit),
"-jsonl",
"-o", output_file,
"-silent",
]
if isinstance(targets, list):
target_file = f"{self.results_dir}/targets_{timestamp}.txt"
with open(target_file, "w") as f:
f.write("\\n".join(targets))
cmd.extend(["-l", target_file])
else:
cmd.extend(["-u", targets])
if templates:
for t in templates:
cmd.extend(["-t", t])
print(f"Scanning {len(targets) if isinstance(targets, list) else 1} targets...")
result = subprocess.run(cmd, capture_output=True, text=True)
findings = self.parse_results(output_file)
return findings
def parse_results(self, output_file):
findings = []
if os.path.exists(output_file):
with open(output_file) as f:
for line in f:
try:
findings.append(json.loads(line.strip()))
except json.JSONDecodeError:
continue
return findings
def summary(self, findings):
severity_count = {}
for f in findings:
sev = f.get("info", {}).get("severity", "unknown")
severity_count[sev] = severity_count.get(sev, 0) + 1
print(f"Total findings: {len(findings)}")
for sev, count in sorted(severity_count.items()):
print(f" {sev}: {count}")
scanner = NucleiScanner()
targets = ["https://example.com", "https://api.example.com"]
findings = scanner.scan(targets)
scanner.summary(findings)
"""
CRON_SCHEDULE = """
# Cron schedule for automated scanning
# Daily scan (critical + high)
0 2 * * * /usr/local/bin/nuclei -l /opt/targets.txt -severity critical, high -jsonl -o /opt/results/daily_$(date +\\%Y\\%m\\%d).jsonl
# Weekly full scan (all severities)
0 3 * * 0 /usr/local/bin/nuclei -l /opt/targets.txt -jsonl -o /opt/results/weekly_$(date +\\%Y\\%m\\%d).jsonl
# Monthly template update
0 1 1 * * /usr/local/bin/nuclei -update-templates
"""
def show_scanner(self):
print("=== Python Scanner ===")
print(self.PYTHON_SCANNER[:600])
def show_cron(self):
print(f"\n=== Cron Schedule ===")
print(self.CRON_SCHEDULE[:400])
pipeline = ScanPipeline()
pipeline.show_scanner()
pipeline.show_cron()Alerting Integration
# alerting.py — Nuclei alerting integration
import json
import random
class NucleiAlerting:
INTEGRATIONS = {
"slack": {
"name": "Slack",
"method": "Webhook",
"config": "nuclei -u target.com -severity critical, high -ms slack -slack-webhook-url https://hooks.slack.com/...",
},
"discord": {
"name": "Discord",
"method": "Webhook",
"config": "nuclei -u target.com -severity critical -ms discord -discord-webhook-url https://discord.com/api/webhooks/...",
},
"telegram": {
"name": "Telegram",
"method": "Bot API",
"config": "nuclei -u target.com -ms telegram -telegram-api-key BOT_TOKEN -telegram-chat-id CHAT_ID",
},
"email": {
"name": "Email (SMTP)",
"method": "SMTP",
"config": "Custom Python script → parse results → send email",
},
"jira": {
"name": "Jira",
"method": "API",
"config": "nuclei -u target.com -ms jira -jira-url https://company.atlassian.net -jira-token TOKEN",
},
}
SLACK_ALERT = """
# slack_alert.py — Custom Slack alerting
import requests
import json
SLACK_WEBHOOK = "https://hooks.slack.com/services/XXX/YYY/ZZZ"
def send_alert(findings):
critical = [f for f in findings if f['info']['severity'] == 'critical']
high = [f for f in findings if f['info']['severity'] == 'high']
if not critical and not high:
return
blocks = [{
"type": "header",
"text": {"type": "plain_text", "text": f"🚨 Nuclei Scan Alert: {len(critical)} Critical, {len(high)} High"}
}]
for finding in (critical + high)[:10]:
blocks.append({
"type": "section",
"text": {"type": "mrkdwn", "text":
f"*[{finding['info']['severity'].upper()}]* {finding['info']['name']}\\n"
f"Host: `{finding.get('host', 'N/A')}`\\n"
f"Template: `{finding.get('template-id', 'N/A')}`"
}
})
requests.post(SLACK_WEBHOOK, json={"blocks": blocks})
print(f"Slack alert sent: {len(critical)} critical, {len(high)} high")
"""
def show_integrations(self):
print("=== Alert Integrations ===\n")
for key, integ in self.INTEGRATIONS.items():
print(f" [{integ['name']}] {integ['method']}")
def show_slack(self):
print(f"\n=== Custom Slack Alert ===")
print(self.SLACK_ALERT[:500])
def simulate(self):
print(f"\n=== Simulated Alert ===")
alerts = [
{"severity": "CRITICAL", "name": "CVE-2024-1234 RCE", "host": "api.example.com"},
{"severity": "HIGH", "name": "SQL Injection", "host": "app.example.com/search"},
{"severity": "HIGH", "name": "Exposed Admin Panel", "host": "admin.example.com"},
]
for a in alerts:
print(f" [{a['severity']:>8}] {a['name']} — {a['host']}")
alert = NucleiAlerting()
alert.show_integrations()
alert.show_slack()
alert.simulate()Dashboard & Reporting
# dashboard.py — Nuclei monitoring dashboard
import json
import random
class NucleiDashboard:
def scan_history(self):
print("=== Scan History (Last 7 Days) ===\n")
for i in range(7, 0, -1):
critical = random.randint(0, 3)
high = random.randint(1, 10)
medium = random.randint(5, 30)
total = critical + high + medium
bar = "█" * (total // 3)
print(f" Day -{i}: C={critical} H={high} M={medium} Total={total:>3} {bar}")
def top_findings(self):
print(f"\n=== Top Findings ===")
findings = [
{"name": "Exposed .git directory", "count": random.randint(3, 15), "severity": "HIGH"},
{"name": "Default credentials", "count": random.randint(2, 8), "severity": "CRITICAL"},
{"name": "Missing security headers", "count": random.randint(10, 50), "severity": "MEDIUM"},
{"name": "SSL certificate issues", "count": random.randint(5, 20), "severity": "MEDIUM"},
{"name": "Open redirect", "count": random.randint(1, 5), "severity": "HIGH"},
]
for f in sorted(findings, key=lambda x: x["count"], reverse=True):
print(f" [{f['severity']:>8}] {f['name']:<30} × {f['count']}")
def asset_coverage(self):
print(f"\n=== Asset Coverage ===")
total = random.randint(50, 200)
scanned = total - random.randint(0, 10)
coverage = (scanned / total) * 100
print(f" Total assets: {total}")
print(f" Scanned: {scanned}")
print(f" Coverage: {coverage:.1f}%")
print(f" Last full scan: {random.randint(1, 7)} days ago")
dash = NucleiDashboard()
dash.scan_history()
dash.top_findings()
dash.asset_coverage()CI/CD Integration
# cicd.py — Nuclei in CI/CD pipeline
import json
class NucleiCICD:
GITHUB_ACTION = """
# .github/workflows/nuclei-scan.yml
name: Nuclei Security Scan
on:
schedule:
- cron: '0 2 * * *' # Daily at 2 AM
workflow_dispatch:
jobs:
nuclei-scan:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Install Nuclei
run: |
go install -v github.com/projectdiscovery/nuclei/v3/cmd/nuclei@latest
nuclei -update-templates
- name: Run Scan
run: |
nuclei -l targets.txt \\
-severity critical, high \\
-jsonl -o results.jsonl \\
-ms slack \\
-slack-webhook-url }
- name: Check Results
run: |
CRITICAL=$(grep -c '"critical"' results.jsonl || true)
if [ "$CRITICAL" -gt 0 ]; then
echo "CRITICAL vulnerabilities found!"
exit 1
fi
- name: Upload Results
if: always()
uses: actions/upload-artifact@v4
with:
name: nuclei-results
path: results.jsonl
"""
def show_action(self):
print("=== GitHub Actions ===")
print(self.GITHUB_ACTION[:600])
def best_practices(self):
print(f"\n=== Best Practices ===")
practices = [
"Rate limit: -rate-limit 100 เพื่อไม่ overload target",
"Exclude: -exclude-tags dos, fuzz สำหรับ production",
"Scope: สแกนเฉพาะ assets ที่ได้รับอนุญาต",
"Schedule: Daily (critical/high), Weekly (all), Monthly (full audit)",
"Triage: review findings ก่อน alert เพื่อลด false positives",
]
for p in practices:
print(f" • {p}")
cicd = NucleiCICD()
cicd.show_action()
cicd.best_practices()FAQ - คำถามที่พบบ่อย
Q: Nuclei กับ Nessus/Qualys อันไหนดี?
A: Nuclei: ฟรี, open source, template-based, เร็วมาก, community-driven Nessus/Qualys: enterprise, compliance-focused, GUI, professional support ใช้ Nuclei: web app scanning, bug bounty, DevSecOps pipeline ใช้ Nessus/Qualys: compliance (PCI DSS, ISO 27001), enterprise audit หลายทีมใช้ทั้งคู่: Nuclei สำหรับ web + Nessus สำหรับ infrastructure
Q: False positives มากไหม?
A: น้อยกว่า traditional scanners เพราะ template-based (specific patterns) Community templates มีคุณภาพดี (reviewed) แต่ยังมีบ้าง โดยเฉพาะ info-level findings แนะนำ: เริ่มด้วย critical+high, triage ก่อน alert, สร้าง exclude list
Q: สแกน production ปลอดภัยไหม?
A: ปลอดภัยถ้าทำถูกวิธี ใช้ -rate-limit (จำกัด requests/sec) ใช้ -exclude-tags dos, fuzz (ไม่ทำ DoS หรือ fuzzing) สแกนช่วง low-traffic แจ้ง ops team ก่อนสแกน ทดสอบใน staging ก่อน production เสมอ
Q: Custom template เขียนยากไหม?
A: ง่ายมาก YAML format อ่านเข้าใจง่าย ดู community templates เป็นตัวอย่าง (8,000+) ใช้ nuclei -validate เพื่อ check template syntax Templates รองรับ: HTTP, DNS, TCP, SSL, file, code, headless เขียน template ใหม่ได้ใน 10-15 นาที