Semgrep SAST Monitoring และ Alerting — คู่มือฉบับสมบูรณ์ 2026

Semgrep SAST Monitoring และ Alerting คืออะไร

Semgrep เป็น open-source Static Application Security Testing (SAST) tool ที่ใช้วิเคราะห์ source code หาช่องโหว่ด้านความปลอดภัย bugs และ code quality issues โดยไม่ต้องรันโปรแกรม ใช้ pattern-based approach ที่เขียน rules ง่ายและเร็วกว่า traditional SAST tools การ setup monitoring และ alerting สำหรับ Semgrep ช่วยให้ทีม security ติดตามผลการสแกนอย่างต่อเนื่อง แจ้งเตือนเมื่อพบ vulnerabilities ใหม่ และ track trends ของ security posture ขององค์กร

Semgrep Fundamentals

# semgrep_basics.py — Semgrep fundamentals
import json

class SemgrepBasics:
 FEATURES = {
 "pattern_matching": {
 "name": "Pattern Matching",
 "description": "เขียน rules ด้วย code patterns — ง่ายกว่า regex, เข้าใจ AST",
 },
 "multi_language": {
 "name": "Multi-Language",
 "description": "รองรับ 30+ ภาษา: Python, JavaScript, Java, Go, C#, Ruby, PHP",
 },
 "fast": {
 "name": "Fast Scanning",
 "description": "สแกนเร็วมาก — 10,000+ files ใน < 1 นาที",
 },
 "cicd": {
 "name": "CI/CD Integration",
 "description": "GitHub Actions, GitLab CI, Jenkins — scan ทุก PR",
 },
 "custom_rules": {
 "name": "Custom Rules",
 "description": "เขียน rules เอง ด้วย YAML — ตรวจจับ patterns เฉพาะองค์กร",
 },
 }

 COMMANDS = """
# Semgrep basic commands

# Install
pip install semgrep

# Scan with default rules
semgrep --config auto .

# Scan with specific ruleset
semgrep --config p/owasp-top-ten .

# Scan with custom rule
semgrep --config my-rules.yaml .

# JSON output
semgrep --config auto --json -o results.json .

# Scan specific language
semgrep --config auto --lang python .

# CI mode (fail on findings)
semgrep ci --config auto
"""

 def show_features(self):
 print("=== Semgrep Features ===\n")
 for key, feat in self.FEATURES.items():
 print(f"[{feat['name']}] {feat['description']}")

 def show_commands(self):
 print(f"\n=== Basic Commands ===")
 print(self.COMMANDS[:400])

basics = SemgrepBasics()
basics.show_features()
basics.show_commands()

Custom Rules

# custom_rules.py — Semgrep custom rules
import json

class SemgrepRules:
 SQL_INJECTION = """
# sql-injection.yaml — Detect SQL injection
rules:
 - id: sql-injection-format-string
 patterns:
 - pattern: |
 cursor.execute(f"... {$VAR} ...")
 - pattern-not: |
 cursor.execute(f"... {$CONST} ...", ...)
 message: "SQL injection via f-string: use parameterized queries"
 languages: [python]
 severity: ERROR
 metadata:
 category: security
 cwe: ["CWE-89"]
 owasp: ["A03:2021"]
"""

 HARDCODED_SECRET = """
# hardcoded-secret.yaml — Detect hardcoded secrets
rules:
 - id: hardcoded-api-key
 patterns:
 - pattern: |
 $KEY = "..."
 - metavariable-regex:
 metavariable: $KEY
 regex: "(api_key|apikey|secret|password|token|auth)"
 - metavariable-regex:
 metavariable: $...EXPR
 regex: "[a-zA-Z0-9]{20,}"
 message: "Hardcoded secret detected: use environment variables"
 languages: [python, javascript, java]
 severity: WARNING
 metadata:
 category: security
 cwe: ["CWE-798"]
"""

 INSECURE_DESERIALIZE = """
# insecure-deserialize.yaml
rules:
 - id: insecure-pickle-load
 pattern: pickle.load(...)
 message: "pickle.load() is vulnerable to arbitrary code execution"
 languages: [python]
 severity: ERROR
 metadata:
 cwe: ["CWE-502"]
 fix: "Use json.load() or yaml.safe_load() instead"
"""

 def show_rules(self):
 print("=== Custom Rules ===\n")
 print("[SQL Injection]")
 print(self.SQL_INJECTION[:300])
 print("\n[Hardcoded Secret]")
 print(self.HARDCODED_SECRET[:300])

rules = SemgrepRules()
rules.show_rules()

Monitoring Pipeline

# monitoring.py — Semgrep monitoring pipeline
import json

class SemgrepMonitoring:
 CODE = """
# semgrep_monitor.py — Monitor Semgrep scan results
import subprocess
import json
from datetime import datetime
from pathlib import Path

class SemgrepMonitor:
 def __init__(self, project_dir, config="auto"):
 self.project_dir = project_dir
 self.config = config
 self.history_file = Path("semgrep_history.json")
 
 def scan(self):
 '''Run Semgrep scan'''
 cmd = [
 "semgrep", "--config", self.config,
 "--json", "--quiet",
 self.project_dir,
 ]
 
 result = subprocess.run(cmd, capture_output=True, text=True)
 
 if result.returncode in [0, 1]: # 0=no findings, 1=findings
 return json.loads(result.stdout)
 return {"error": result.stderr}
 
 def parse_results(self, scan_output):
 '''Parse scan results into summary'''
 results = scan_output.get('results', [])
 
 by_severity = {}
 by_category = {}
 by_file = {}
 
 for r in results:
 sev = r.get('extra', {}).get('severity', 'unknown')
 by_severity[sev] = by_severity.get(sev, 0) + 1
 
 cat = r.get('extra', {}).get('metadata', {}).get('category', 'unknown')
 by_category[cat] = by_category.get(cat, 0) + 1
 
 file = r.get('path', 'unknown')
 by_file[file] = by_file.get(file, 0) + 1
 
 return {
 'timestamp': datetime.utcnow().isoformat(),
 'total_findings': len(results),
 'by_severity': by_severity,
 'by_category': by_category,
 'top_files': dict(sorted(by_file.items(), key=lambda x: -x[1])[:10]),
 'critical_findings': [
 {
 'rule': r.get('check_id', ''),
 'file': r.get('path', ''),
 'line': r.get('start', {}).get('line', 0),
 'message': r.get('extra', {}).get('message', ''),
 }
 for r in results
 if r.get('extra', {}).get('severity') == 'ERROR'
 ],
 }
 
 def compare_with_previous(self, current):
 '''Compare with previous scan'''
 if self.history_file.exists():
 history = json.loads(self.history_file.read_text())
 previous = history[-1] if history else None
 else:
 previous = None
 
 if previous:
 return {
 'previous_total': previous['total_findings'],
 'current_total': current['total_findings'],
 'delta': current['total_findings'] - previous['total_findings'],
 'trend': 'improving' if current['total_findings'] < previous['total_findings'] else 'worsening',
 }
 return {'note': 'No previous scan to compare'}
 
 def save_history(self, summary):
 '''Save scan to history'''
 history = []
 if self.history_file.exists():
 history = json.loads(self.history_file.read_text())
 
 history.append(summary)
 
 # Keep last 100 scans
 if len(history) > 100:
 history = history[-100:]
 
 self.history_file.write_text(json.dumps(history, indent=2))
 
 def should_alert(self, summary, thresholds=None):
 '''Determine if alert should fire'''
 if thresholds is None:
 thresholds = {
 'critical_max': 0,
 'high_max': 5,
 'total_max': 50,
 }
 
 alerts = []
 
 critical = summary['by_severity'].get('ERROR', 0)
 if critical > thresholds['critical_max']:
 alerts.append({
 'level': 'critical',
 'message': f"Critical findings: {critical} (threshold: {thresholds['critical_max']})",
 })
 
 high = summary['by_severity'].get('WARNING', 0)
 if high > thresholds['high_max']:
 alerts.append({
 'level': 'high',
 'message': f"High findings: {high} (threshold: {thresholds['high_max']})",
 })
 
 if summary['total_findings'] > thresholds['total_max']:
 alerts.append({
 'level': 'medium',
 'message': f"Total findings: {summary['total_findings']} (threshold: {thresholds['total_max']})",
 })
 
 return alerts

# monitor = SemgrepMonitor("./src")
# output = monitor.scan()
# summary = monitor.parse_results(output)
# alerts = monitor.should_alert(summary)
"""

 def show_code(self):
 print("=== Semgrep Monitor ===")
 print(self.CODE[:600])

monitor = SemgrepMonitoring()
monitor.show_code()

CI/CD Integration

# cicd.py — Semgrep CI/CD with alerting
import json

class SemgrepCICD:
 GITHUB_ACTIONS = """
# .github/workflows/semgrep.yml
name: Semgrep Security Scan
on:
 push:
 branches: [main]
 pull_request:
 schedule:
 - cron: '0 8 * * 1' # Weekly Monday 8am

jobs:
 semgrep:
 runs-on: ubuntu-latest
 container:
 image: semgrep/semgrep
 steps:
 - uses: actions/checkout@v4
 
 - name: Semgrep Scan
 run: semgrep ci --config auto --json -o results.json
 env:
 SEMGREP_APP_TOKEN: }
 
 - name: Parse Results
 if: always()
 run: |
 CRITICAL=$(cat results.json | python3 -c "
 import json, sys
 data = json.load(sys.stdin)
 critical = sum(1 for r in data.get('results', [])
 if r.get('extra', {}).get('severity') == 'ERROR')
 print(critical)
 ")
 echo "Critical findings: $CRITICAL"
 if [ "$CRITICAL" -gt 0 ]; then
 echo "CRITICAL_FOUND=true" >> $GITHUB_ENV
 fi
 
 - name: Upload Results
 if: always()
 uses: actions/upload-artifact@v4
 with:
 name: semgrep-results
 path: results.json
 
 - name: Slack Alert
 if: env.CRITICAL_FOUND == 'true'
 uses: slackapi/slack-github-action@v1
 with:
 payload: |
 {
 "text": "CRITICAL: Semgrep found critical vulnerabilities in }"
 }
 env:
 SLACK_WEBHOOK_URL: }
 
 - name: Fail on Critical
 if: env.CRITICAL_FOUND == 'true'
 run: exit 1
"""

 ALERTING = {
 "slack": "Webhook → ส่ง message เมื่อพบ critical/high findings",
 "jira": "สร้าง Jira ticket อัตโนมัติสำหรับ findings ใหม่",
 "email": "ส่ง daily/weekly summary report ให้ security team",
 "pagerduty": "Alert เฉพาะ critical findings ที่ต้อง fix ด่วน",
 "dashboard": "Grafana dashboard แสดง trends, top findings, fix rate",
 }

 def show_pipeline(self):
 print("=== GitHub Actions ===")
 print(self.GITHUB_ACTIONS[:500])

 def show_alerting(self):
 print(f"\n=== Alerting Channels ===")
 for channel, desc in self.ALERTING.items():
 print(f" [{channel}] {desc}")

cicd = SemgrepCICD()
cicd.show_pipeline()
cicd.show_alerting()

Dashboard & Reporting

# dashboard.py — Semgrep dashboard and reporting
import json
import random

class SemgrepDashboard:
 METRICS = {
 "total_findings": "จำนวน findings ทั้งหมด — target: ลดลงทุกสัปดาห์",
 "mttr": "Mean Time to Remediate — เวลาเฉลี่ยที่ใช้ fix finding",
 "fix_rate": "% ของ findings ที่ fix แล้วต่อสัปดาห์",
 "new_vs_fixed": "จำนวน findings ใหม่ vs fixed — ต้อง fixed > new",
 "coverage": "% ของ repos ที่มี Semgrep scan — target: 100%",
 "false_positive": "% ของ findings ที่เป็น false positive — ต้อง < 10%",
 }

 WEEKLY_REPORT = {
 "period": "Week 24, 2026",
 "repos_scanned": 45,
 "total_findings": 234,
 "critical": 3,
 "high": 28,
 "medium": 103,
 "low": 100,
 "new_findings": 12,
 "fixed_findings": 18,
 "trend": "improving (-6 net findings)",
 "top_categories": [
 ("SQL Injection", 15),
 ("XSS", 12),
 ("Hardcoded Secrets", 8),
 ("Insecure Deserialization", 5),
 ],
 }

 def show_metrics(self):
 print("=== Dashboard Metrics ===\n")
 for metric, desc in self.METRICS.items():
 print(f" [{metric}] {desc}")

 def show_weekly_report(self):
 r = self.WEEKLY_REPORT
 print(f"\n=== Weekly Report ({r['period']}) ===")
 print(f" Repos Scanned: {r['repos_scanned']}")
 print(f" Total: {r['total_findings']} (Critical: {r['critical']}, High: {r['high']})")
 print(f" New: {r['new_findings']}, Fixed: {r['fixed_findings']}")
 print(f" Trend: {r['trend']}")
 print(f"\n Top Categories:")
 for cat, count in r['top_categories']:
 print(f" {cat}: {count}")

dashboard = SemgrepDashboard()
dashboard.show_metrics()
dashboard.show_weekly_report()

FAQ - คำถามที่พบบ่อย

Q: Semgrep กับ SonarQube อันไหนดีกว่า?

อ่านเพิ่ม: Developer Toolkit 2026 รวมเครื่องมือที่ Developer ทุกคนต้องม · อ่านเพิ่ม: Monitoring คืออะไร? สอน Observability ตั้งแต่ Prometheus Gra · อ่านเพิ่ม: Shell Scripting คืออะไร? สอนเขียน Bash Script สำหรับ DevOps

A: Semgrep: เร็วกว่า, rules เขียนง่ายกว่า (YAML), CI/CD friendly, open-source SonarQube: comprehensive กว่า (code quality + security), IDE integration, dashboard ดี เลือก Semgrep: ถ้าเน้น security, ต้องการ custom rules, CI/CD integration เลือก SonarQube: ถ้าต้องการ code quality + security รวมกัน, มี server dedicated ใช้ร่วมกัน: Semgrep ใน CI (fast feedback) + SonarQube สำหรับ deep analysis

Q: False positive เยอะไหม?

A: Semgrep มี false positive rate ต่ำกว่า traditional SAST — เพราะใช้ semantic analysis วิธีลด: ใช้ curated rulesets (p/owasp-top-ten), tune rules ตาม codebase, ใช้ nosemgrep comment suppress False positive management: mark as false positive ใน Semgrep App → ไม่แจ้งซ้ำ ทั่วไป: false positive ~5-15% — ดีกว่า SAST tools อื่นที่อาจสูง 30-50%

Q: Semgrep ฟรีไหม?

A: Semgrep OSS: ฟรี 100% — CLI scanner + community rules + custom rules Semgrep App (Cloud): Free tier สำหรับ small teams, Team/Enterprise plans มีค่าใช้จ่าย สิ่งที่ต้องจ่าย: dashboard, RBAC, advanced features, Semgrep Supply Chain (SCA) สำหรับส่วนใหญ่: Semgrep OSS + GitHub Actions ฟรีเพียงพอ

Q: ใช้เวลาสแกนนานไหม?

A: เร็วมาก — เป็นจุดเด่นของ Semgrep Benchmark: 10,000 files ใน < 60 วินาที, 100,000 files ใน < 5 นาที เปรียบเทียบ: SonarQube อาจใช้ 10-30 นาที สำหรับ codebase เดียวกัน เหตุผล: Semgrep ทำงาน locally ไม่ต้อง compile, ใช้ parallel processing เหมาะสำหรับ PR checks — ไม่ทำให้ CI pipeline ช้า