איסוף יומנים של Qualys Continuous Monitoring

נתמך ב:

קוד מנתח ה-Logstash הזה מחלץ קודם שדות כמו כתובת IP של המקור, משתמש, שיטה ופרוטוקול אפליקציה מהודעות יומן גולמיות באמצעות דפוסי grok. לאחר מכן, המערכת ממפה שדות ספציפיים מנתוני היומן הגולמיים לשדות התואמים שלהם במודל הנתונים המאוחד (UDM), מבצעת המרות של סוגי נתונים ומעשירה את הנתונים באמצעות תוויות ומטא-נתונים נוספים, לפני שהיא יוצרת את הפלט בפורמט ה-UDM הרצוי.

לפני שמתחילים

חשוב לוודא שמתקיימות דרישות הסף הבאות:

  • מופע של Google Security Operations.
  • גישה עם הרשאות ל- Google Cloud.
  • גישה עם הרשאות ל-Qualys.

מפעילים את ממשקי ה-API הנדרשים:

  1. נכנסים למסוף Google Cloud .
  2. עוברים אל APIs & Services (ממשקי API ושירותים) > Library (ספרייה).
  3. מחפשים את ממשקי ה-API הבאים ומפעילים אותם:
    • Cloud Functions API
    • Cloud Scheduler API
    • ‫Cloud Pub/Sub (נדרש כדי ש-Cloud Scheduler יפעיל פונקציות)

יצירת Google Cloud קטגוריית אחסון

  1. נכנסים למסוף Google Cloud .

  2. נכנסים לדף Cloud Storage Buckets.

    כניסה לדף Buckets

  3. לוחצים על יצירה.

  4. מגדירים את הקטגוריה:

    • שם: מזינים שם ייחודי שעומד בדרישות לשמות של קטגוריות (לדוגמה, qualys-asset-bucket).
    • Choose where to store your data (בחירת המיקום לאחסון הנתונים): בוחרים מיקום.
    • Choose a storage class for your data: בוחרים default storage class לקטגוריה, או בוחרים ב-Autoclass לניהול אוטומטי של סוג האחסון (storage class).
    • בקטע Choose how to control access to objects, בוחרים באפשרות not כדי לא לאכוף public access prevention, ובוחרים באפשרות access control model לאובייקטים של הקטגוריה.
    • סוג אחסון (storage class): בוחרים לפי הצרכים (לדוגמה, Standard).

  5. לוחצים על יצירה.

יצירת חשבון שירות ב-Google Cloud

  1. נכנסים למסוף Google Cloud .
  2. עוברים אל IAM & Admin > Service Accounts.
  3. יוצרים חשבון שירות חדש.
  4. נותנים לו שם תיאורי (לדוגמה, qualys-user).
  5. מקצים לחשבון השירות את התפקיד אדמין של אובייקט אחסון בקטגוריית GCS שיצרתם בשלב הקודם.
  6. מקצים לחשבון השירות את התפקיד Cloud Functions Invoker.
  7. יוצרים מפתח SSH לחשבון השירות.
  8. מורידים קובץ מפתח JSON לחשבון השירות. חשוב לשמור על הקובץ הזה בצורה מאובטחת.

אופציונלי: יוצרים משתמש API ייעודי ב-Qualys

  1. נכנסים למסוף Qualys.
  2. מעבר לדף משתמשים
  3. לוחצים על חדש > משתמש.
  4. מזינים את המידע הכללי הנדרש לגבי המשתמש.
  5. לוחצים על הכרטיסייה תפקיד משתמש.
  6. מוודאים שהתיבה API Access (גישה ל-API) מסומנת בתפקיד.
  7. לוחצים על Save.

זיהוי כתובת ה-URL הספציפית של Qualys API

אפשרות 1

מזהים את כתובות ה-URL כמו שמתואר בזיהוי הפלטפורמה.

אפשרות 2

  1. נכנסים למסוף Qualys.
  2. עוברים אל עזרה > אודות.
  3. גוללים כדי לראות את המידע הזה בקטע Security Operations Center (SOC) [מרכז אבטחה].
  4. מעתיקים את כתובת ה-URL של Qualys API.

הגדרת הפונקציה של Cloud Functions

  1. נכנסים אל Cloud Functions במסוף Google Cloud .
  2. לוחצים על יצירת פונקציה.

  3. מגדירים את הפונקציה:

    • שם: מזינים שם לפונקציה (לדוגמה, fetch-qualys-cm-alerts).
    • אזור: בוחרים אזור שקרוב ל-Bucket.
    • זמן ריצה: Python 3.10 (או זמן הריצה המועדף).
    • Trigger (טריגר): בוחרים באפשרות HTTP trigger (טריגר HTTP) אם צריך, או באפשרות Cloud Pub/Sub (Cloud Pub/Sub) להפעלה מתוזמנת.
    • אימות: מאובטח באמצעות אימות.
    • כתיבת הקוד באמצעות עורך ישיר:
    ```python
from google.cloud import storage
import requests
import base64
import json

# Google Cloud Storage Configuration
BUCKET_NAME = "<bucket-name>"
FILE_NAME = "qualys_cm_alerts.json"

# Qualys API Credentials
QUALYS_USERNAME = "<qualys-username>"
QUALYS_PASSWORD = "<qualys-password>"
QUALYS_BASE_URL = "https://<qualys_base_url>"

def fetch_cm_alerts():
    """Fetch alerts from Qualys Continuous Monitoring."""
    auth = base64.b64encode(f"{QUALYS_USERNAME}:{QUALYS_PASSWORD}".encode()).decode()
    headers = {
        "Authorization": f"Basic {auth}",
        "Content-Type": "application/xml"
    }
    payload = """
    <ServiceRequest>
        <filters>
            <Criteria field="alert.date" operator="GREATER">2024-01-01</Criteria>
        </filters>
    </ServiceRequest>
    """
    response = requests.post(f"{QUALYS_BASE_URL}/qps/rest/2.0/search/cm/alert", headers=headers, data=payload)
    response.raise_for_status()
    return response.json()

def upload_to_gcs(data):
    """Upload data to Google Cloud Storage."""
    client = storage.Client()
    bucket = client.get_bucket(BUCKET_NAME)
    blob = bucket.blob(FILE_NAME)
    blob.upload_from_string(json.dumps(data, indent=2), content_type="application/json")

def main(request):
    """Cloud Function entry point."""
    try:
        alerts = fetch_cm_alerts()
        upload_to_gcs(alerts)
        return "Qualys CM alerts uploaded to Cloud Storage successfully!"
    except Exception as e:
        return f"An error occurred: {e}", 500
```

  4. אחרי שמסיימים את ההגדרה, לוחצים על Deploy (פריסה).

הגדרת Cloud Scheduler

  1. נכנסים אל Cloud Scheduler במסוף Google Cloud .
  2. לוחצים על יצירת משימה.

  3. מגדירים את המשימה:

    • שם: מזינים שם למשימה (לדוגמה, trigger-fetch-qualys-cm-alerts).
    • תדירות: משתמשים בתחביר של cron כדי לציין את לוח הזמנים (לדוגמה, 0 * * * * כדי להפעיל כל שעה).
    • אזור זמן: מגדירים את אזור הזמן המועדף.
    • סוג הטריגר: בוחרים באפשרות HTTP.
    • כתובת URL של טריגר: מזינים את כתובת ה-URL של Cloud Functions (מופיעה בפרטי הפונקציה אחרי הפריסה).
    • Method (שיטה): בוחרים באפשרות POST.

  4. יוצרים את המשרה.

הגדרת פידים

כדי להגדיר פיד, מבצעים את השלבים הבאים:

  1. עוברים אל SIEM Settings > Feeds.
  2. לוחצים על הוספת פיד חדש.
  3. בדף הבא, לוחצים על הגדרת פיד יחיד.
  4. בשדה Feed name, מזינים שם לפיד, למשל Qualys Continuous Monitoring Logs.
  5. בוחרים באפשרות Google Cloud Storage V2 בתור סוג המקור.
  6. בוחרים באפשרות Qualys Continuous Monitoring בתור סוג היומן.
  7. לוחצים על הבא.

  8. מציינים ערכים לפרמטרים הבאים של הקלט:

    • ‫Storage Bucket URI: ‏ Google Cloud URI של קטגוריית המקור של אחסון.
    • אפשרויות למחיקת מקורות: בוחרים את אפשרות המחיקה לפי ההעדפה.

  9. לוחצים על הבא.

  10. בודקים את ההגדרות של הפיד החדש במסך סיום ולוחצים על שליחה.

טבלת מיפוי UDM

שדה היומן מיפוי UDM לוגיקה
Alert.alertInfo.appVersion metadata.product_version מיפוי ישיר מ-Alert.alertInfo.appVersion
Alert.alertInfo.operatingSystem principal.platform_version מיפוי ישיר מ-Alert.alertInfo.operatingSystem
Alert.alertInfo.port additional.fields.value.string_value מופה ישירות מ-Alert.alertInfo.port ונוסף כצמד מפתח/ערך ב-additional.fields עם המפתח Alert port
Alert.alertInfo.protocol network.ip_protocol מיפוי ישיר מ-Alert.alertInfo.protocol
Alert.alertInfo.sslIssuer network.tls.client.certificate.issuer מיפוי ישיר מ-Alert.alertInfo.sslIssuer
Alert.alertInfo.sslName additional.fields.value.string_value מופה ישירות מ-Alert.alertInfo.sslName ונוסף כצמד מפתח/ערך ב-additional.fields עם המפתח 'שם SSL'
Alert.alertInfo.sslOrg additional.fields.value.string_value ממופה ישירות מ-Alert.alertInfo.sslOrg ומוסף כצמד מפתח/ערך ב-additional.fields עם המפתח SSL Org
Alert.alertInfo.ticketId additional.fields.value.string_value מופה ישירות מ-Alert.alertInfo.ticketId ונוסף כצמד מפתח/ערך ב-additional.fields עם המפתח Ticket Id
Alert.alertInfo.vpeConfidence additional.fields.value.string_value ממופה ישירות מ-Alert.alertInfo.vpeConfidence ונוסף כצמד מפתח/ערך ב-additional.fields עם המפתח VPE Confidence
Alert.alertInfo.vpeStatus additional.fields.value.string_value ממופה ישירות מ-Alert.alertInfo.vpeStatus ונוסף כצמד מפתח/ערך ב-additional.fields עם המפתח VPE Confidence
Alert.eventType additional.fields.value.string_value ממופה ישירות מ-Alert.eventType ונוסף כצמד מפתח/ערך ב-additional.fields עם המפתח Event Type
Alert.hostname principal.hostname מיפוי ישיר מ-Alert.hostname
Alert.id security_result.threat_id מיפוי ישיר מ-Alert.id
Alert.ipAddress principal.ip מיפוי ישיר מ-Alert.ipAddress
Alert.profile.id additional.fields.value.string_value מופה ישירות מ-Alert.profile.id ונוסף כצמד מפתח/ערך ב-additional.fields עם המפתח Profile Id
Alert.profile.title additional.fields.value.string_value מופה ישירות מ-Alert.profile.title ונוסף כצמד מפתח/ערך ב-additional.fields עם המפתח 'שם הפרופיל'
Alert.qid vulnerability.name מופה כ-QID: ‏ מתוך Alert.qid
Alert.source additional.fields.value.string_value מופה ישירות מ-Alert.source ונוסף כצמד מפתח/ערך ב-additional.fields עם המפתח Alert Source
Alert.triggerUuid metadata.product_log_id מיפוי ישיר מ-Alert.triggerUuid
Alert.vulnCategory additional.fields.value.string_value ממופה ישירות מ-Alert.vulnCategory ונוסף כצמד מפתח/ערך ב-additional.fields עם המפתח 'קטגוריית פגיעות'
Alert.vulnSeverity vulnerability.severity המיפוי מתבצע על סמך הערך של Alert.vulnSeverity: 1-3: נמוך, 4-6: בינוני, 7-8: גבוה
Alert.vulnTitle vulnerability.description מיפוי ישיר מ-Alert.vulnTitle
Alert.vulnType additional.fields.value.string_value ממופה ישירות מ-Alert.vulnType ומוסף כצמד מפתח/ערך ב-additional.fields עם המפתח 'סוג הפגיעות'
מארח principal.ip נותח משורת היומן 'מארח: '
edr.client.ip_addresses הועתק ממכשיר principal.ip
edr.client.hostname הועתק ממכשיר principal.hostname
edr.raw_event_name הערך צריך להיות STATUS_UPDATE אם יש Alert.ipAddress, Alert.hostname או src_ip, אחרת הערך צריך להיות GENERIC_EVENT
metadata.event_timestamp הערך מחולץ מהשדות Alert.eventDate או timestamp. אם השדה Alert.eventDate קיים, הוא מקבל עדיפות. אחרת, נעשה שימוש בשדה timestamp. חותמת הזמן מומרת ל-UTC.
metadata.event_type אותה לוגיקה כמו edr.raw_event_name
metadata.log_type ההגדרה היא QUALYS_CONTINUOUS_MONITORING
metadata.product_name ההגדרה היא QUALYS_CONTINUOUS_MONITORING
metadata.vendor_name ההגדרה היא QUALYS_CONTINUOUS_MONITORING
network.application_protocol נותח משורת היומן /user HTTP"
network.http.method הניתוח בוצע משורת היומן /user HTTP
חותמת זמן event.timestamp הערך מחולץ מהשדות Alert.eventDate או timestamp. אם השדה Alert.eventDate קיים, הוא מקבל עדיפות. אחרת, נעשה שימוש בשדה timestamp. חותמת הזמן מומרת ל-UTC.

שנה רישום

צפייה ביומן השינויים של כלי הניתוח הזה

הבעיה עדיין לא נפתרה? קבלת תשובות מחברי הקהילה וממומחי Google SecOps.