איסוף יומנים של Druva Backup

גרסת כלי הניתוח: 2.0

נתמך ב:

במאמר הזה מוסבר איך לאסוף יומנים של Druva Backup על ידי הגדרת פונקציה של Google Cloud Run שמאחזרת אירועים מ-Druva REST API וכותבת אותם לדלי של Google Cloud Storage, ולאחר מכן הגדרת פיד של Google Security Operations באמצעות Google Cloud Storage V2.

‫Druva היא פלטפורמה מבוססת-ענן להגנה על נתונים ולניהול שלהם, שמספקת שירותי גיבוי, התאוששות מאסון וארכיון לנקודות קצה, לאפליקציות SaaS ולעומסי עבודה ארגוניים. הפלטפורמה יוצרת נתוני ביקורת מקיפים, אירועי גיבוי, פעולות שחזור והתראות אבטחה שאפשר לשלב עם פתרונות SIEM לצורך מעקב ותאימות.

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

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

  • מופע של Google SecOps
  • פרויקט ב-Google Cloud שהחיוב בו מופעל
  • ממשקי ה-API הבאים של Google Cloud הופעלו:
    • Cloud Run functions API
    • Cloud Scheduler API
    • Cloud Storage API
    • Pub/Sub API
    • IAM API
  • גישת אדמין ב-Druva Cloud אל מסוף Druva Cloud Platform
  • גישה למרכז השילוב של Druva ליצירת פרטי כניסה ל-API

יצירת קטגוריה של Google Cloud Storage

  1. נכנסים למסוף Google Cloud.
  2. בוחרים פרויקט או יוצרים פרויקט חדש.
  3. בתפריט הניווט, עוברים אל Cloud Storage > Buckets.
  4. לוחצים על Create bucket.

  5. מספקים את פרטי ההגדרה הבאים:

    הגדרה ערך
    Name your bucket (שם הקטגוריה) מזינים שם ייחודי גלובלית (לדוגמה, druva-backup-logs)
    סוג המיקום בוחרים לפי הצרכים (אזור, שני אזורים, מספר אזורים)
    Location בוחרים את המיקום הכי קרוב למופע של Google SecOps (לדוגמה, us-central1)
    סוג אחסון (storage class) רגיל (מומלץ ליומנים שמתבצעת אליהם גישה לעיתים קרובות)
    בקרת גישה אחיד (מומלץ)
    כלים להגנה אופציונלי: הפעלת ניהול גרסאות של אובייקטים או מדיניות שמירת נתונים

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

איסוף פרטי הכניסה ל-Druva API

כדי לאפשר לפונקציית Cloud Run לאחזר אירועים מ-Druva, צריך ליצור פרטי כניסה ל-API עם אימות OAuth 2.0.

יצירת פרטי כניסה ל-API

  1. נכנסים לDruva Cloud Platform Console.
  2. בתפריט Global Navigation בוחרים באפשרות מרכז השילובים.
  3. בחלונית הימנית, לוחצים על פרטי כניסה ל-API.
  4. לוחצים על New Credentials (פרטי כניסה חדשים).
  5. בחלון New Credentials (פרטי כניסה חדשים), מזינים את הפרטים הבאים: Name (שם): מזינים שם תיאורי (לדוגמה, Google SecOps Cloud Storage Integration).
  6. כדי להחיל הגבלות על הרשאות:
    1. בוחרים באפשרות Druva Cloud Administrator כדי לאפשר גישה מלאה לאחזור ולשינוי נתונים.
    2. לחלופין, בוחרים באפשרות אדמין מוצר ובוחרים באחת מהאפשרויות הבאות: תפקיד אדמין בענן (קריאה בלבד): כדי להגביל את הגישה לאחזור נתונים בלבד, ללא זכויות שינוי (מומלץ לשילוב עם SIEM)
  7. לוחצים על Save.

תיעוד פרטי הכניסה ל-API

אחרי שיוצרים את פרטי הכניסה ל-API, מופיע החלון פרטי פרטי הכניסה:

  1. לוחצים על סמל ההעתקה לצד מזהה לקוח כדי להעתיק את הערך ללוח.
  2. שומרים את מזהה הלקוח באופן מאובטח (לדוגמה, McNkxxxx4Vicxxxx4Ldpxxxx/09Uxxxx).
  3. לוחצים על סמל ההעתקה לצד Secret Key (מפתח סודי) כדי להעתיק את הערך ללוח.

  4. שומרים את המפתח הסודי באופן מאובטח (לדוגמה, Xmcxxxx8j5xxxx6NxxxxRbRxxxxNNyPt).

יצירה של חשבון שירות

יוצרים חשבון שירות ייעודי לפונקציית Cloud Run כדי לגשת ל-Google Cloud Storage.

  1. במסוף Google Cloud, עוברים אל IAM & Admin > Service Accounts.
  2. לוחצים על יצירת חשבון שירות.
  3. מספקים את פרטי ההגדרה הבאים:
    • שם חשבון השירות: מזינים druva-backup-function (או שם תיאורי)
    • תיאור חשבון השירות: מזינים Service account for Druva Backup Cloud Run function
  4. לוחצים על יצירה והמשך.
  5. בקטע Grant this service account access to project (מתן גישה של חשבון השירות הזה לפרויקט), מוסיפים את התפקידים הבאים:
    1. לוחצים על Select a role (בחירת תפקיד) ובוחרים באפשרות Storage Object Admin (אדמין של אובייקט אחסון).
    2. לוחצים על Add another role ובוחרים באפשרות Cloud Run Invoker.
  6. לוחצים על Continue.
  7. לוחצים על סיום.

  8. רושמים את כתובת האימייל בחשבון השירות (לדוגמה, druva-backup-function@PROJECT_ID.iam.gserviceaccount.com).

יוצרים נושא Pub/Sub

יוצרים נושא Pub/Sub ש-Cloud Scheduler ישתמש בו כדי להפעיל את פונקציית Cloud Run.

  1. במסוף Google Cloud, עוברים אל Pub/Sub > Topics.
  2. לוחצים על יצירת נושא.
  3. מספקים את פרטי ההגדרה הבאים:
    • מזהה הנושא: מזינים druva-backup-trigger
  4. מבטלים את הסימון של Add a default subscription (הוספת מינוי ברירת מחדל).

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

יצירת פונקציית Cloud Run

הכנת קוד הפונקציה

יוצרים פונקציה ב-Cloud Run שמבצעת אימות באמצעות פרטי כניסה של לקוח OAuth 2.0 ל-Druva API, מאחזרת אירועים דרך נקודת הקצה של האירועים עם חלוקה לעמודים, וכותבת את התוצאות כ-NDJSON לדלי GCS.

פריסת הפונקציה של Cloud Run

  1. במסוף Google Cloud, עוברים אל Cloud Run functions.
  2. לוחצים על Create Function (יצירת פונקציה).

  3. מספקים את פרטי ההגדרה הבאים:

    • סביבה: בוחרים באפשרות דור שני.
    • שם הפונקציה: מזינים druva-backup-to-gcs
    • אזור: בוחרים את האזור הכי קרוב לקטגוריית GCS (לדוגמה, us-central1)
    • Trigger type: בוחרים באפשרות Cloud Pub/Sub.
    • נושא ב-Cloud Pub/Sub: בוחרים באפשרות druva-backup-trigger
    • חשבון שירות: בוחרים באפשרות druva-backup-function@PROJECT_ID.iam.gserviceaccount.com
    • הזיכרון שהוקצה: 512 MiB
    • ‫Timeout: 540 seconds
    • מספר המופעים המקסימלי: 1

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

  5. בוחרים באפשרות Python 3.11 בתור סביבת זמן הריצה.

  6. מגדירים את Entry point ל-main.

  7. בעורך קוד המקור, מחליפים את התוכן של main.py בתוכן הבא:

    import base64
import json
import os
import time
from datetime import datetime, timezone, timedelta

import requests
from google.cloud import storage

GCS_BUCKET = os.environ["GCS_BUCKET"]
GCS_PREFIX = os.environ.get("GCS_PREFIX", "druva_backup")
STATE_KEY = os.environ.get("STATE_KEY", "druva_state.json")
DRUVA_BASE_URL = os.environ.get("DRUVA_BASE_URL", "apis.druva.com")
CLIENT_ID = os.environ["CLIENT_ID"]
CLIENT_SECRET = os.environ["CLIENT_SECRET"]
MAX_RECORDS = int(os.environ.get("MAX_RECORDS", "10000"))
PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "500"))
LOOKBACK_HOURS = int(os.environ.get("LOOKBACK_HOURS", "24"))

def get_oauth_token():
    """Obtain OAuth 2.0 access token using client credentials grant."""
    token_url = f"https://{DRUVA_BASE_URL}/token"
    payload = {
        "grant_type": "client_credentials",
        "scope": "read",
    }
    resp = requests.post(
        token_url,
        data=payload,
        auth=(CLIENT_ID, CLIENT_SECRET),
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["access_token"]

def load_state(storage_client):
    """Load the persisted state (last event time and tracker) from GCS."""
    bucket = storage_client.bucket(GCS_BUCKET)
    blob = bucket.blob(f"{GCS_PREFIX}/{STATE_KEY}")
    if blob.exists():
        return json.loads(blob.download_as_text())
    return {}

def save_state(storage_client, state):
    """Persist state to GCS."""
    bucket = storage_client.bucket(GCS_BUCKET)
    blob = bucket.blob(f"{GCS_PREFIX}/{STATE_KEY}")
    blob.upload_from_string(
        json.dumps(state),
        content_type="application/json",
    )

def fetch_events(token, state):
    """Fetch events from Druva API with pagination via nextPageToken."""
    events_url = f"https://{DRUVA_BASE_URL}/insync/eventmanagement/v2/events"
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json",
    }

    params = {"pageSize": PAGE_SIZE}

    tracker = state.get("tracker")
    last_event_time = state.get("last_event_time")

    if tracker:
        params["tracker"] = tracker
    elif last_event_time:
        params["fromTime"] = last_event_time
    else:
        lookback = datetime.now(timezone.utc) - timedelta(hours=LOOKBACK_HOURS)
        params["fromTime"] = lookback.strftime("%Y-%m-%dT%H:%M:%SZ")

    all_events = []
    total_fetched = 0

    while total_fetched < MAX_RECORDS:
        resp = requests.get(
            events_url,
            headers=headers,
            params=params,
            timeout=60,
        )
        resp.raise_for_status()
        data = resp.json()

        events = data.get("events", [])
        all_events.extend(events)
        total_fetched += len(events)

        new_tracker = data.get("tracker")
        next_page_token = data.get("nextPageToken")

        if new_tracker:
            state["tracker"] = new_tracker

        if next_page_token:
            params["nextPageToken"] = next_page_token
            params.pop("tracker", None)
            params.pop("fromTime", None)
        else:
            break

    if all_events:
        last_ts = all_events[-1].get("eventTime", "")
        if last_ts:
            state["last_event_time"] = last_ts

    return all_events, state

def write_events_to_gcs(storage_client, events):
    """Write events as NDJSON to GCS."""
    if not events:
        return

    now = datetime.now(timezone.utc)
    filename = now.strftime("%Y%m%d_%H%M%S") + ".ndjson"
    blob_path = f"{GCS_PREFIX}/{now.strftime('%Y/%m/%d')}/{filename}"

    ndjson_lines = "\n".join(json.dumps(event) for event in events)

    bucket = storage_client.bucket(GCS_BUCKET)
    blob = bucket.blob(blob_path)
    blob.upload_from_string(
        ndjson_lines,
        content_type="application/x-ndjson",
    )
    print(f"Wrote {len(events)} events to gs://{GCS_BUCKET}/{blob_path}")

def main(event, context):
    """Cloud Run function entry point triggered by Pub/Sub."""
    storage_client = storage.Client()

    token = get_oauth_token()

    state = load_state(storage_client)

    events, updated_state = fetch_events(token, state)

    write_events_to_gcs(storage_client, events)

    save_state(storage_client, updated_state)

    print(f"Completed: fetched {len(events)} events")
    return f"OK: {len(events)} events"

  8. מחליפים את התוכן של requirements.txt בתוכן הבא:

    requests>=2.31.0
google-cloud-storage>=2.14.0

הגדרת משתני סביבה

  1. בהגדרות של פונקציית Cloud Run, עוברים לקטע Runtime, build, connections and security settings (הגדרות של זמן ריצה, build, חיבורים ואבטחה).

  2. בקטע Runtime environment variables (משתני סביבה של זמן הריצה), מוסיפים את המשתנים הבאים:

    • ‫GCS_BUCKET: השם של קטגוריית GCS (לדוגמה, druva-backup-logs)
    • ‫GCS_PREFIX: נתיב הקידומת לקובצי יומן (לדוגמה, druva_backup)
    • ‫STATE_KEY: שם קובץ תמונת המצב (לדוגמה, druva_state.json)
    • ‫DRUVA_BASE_URL: כתובת ה-URL הבסיסית של Druva API:
      • ‫apis.druva.com ל-Druva Cloud (Standard)
      • ‫govcloudapis.druva.com ל-Druva GovCloud
    • ‫CLIENT_ID: מזהה הלקוח מפרטי הכניסה ל-Druva API
    • ‫CLIENT_SECRET: מפתח הסוד מפרטי הכניסה ל-Druva API
    • ‫MAX_RECORDS: המספר המקסימלי של רשומות לאחזור בכל הפעלה (לדוגמה, 10000)
    • ‫PAGE_SIZE: מספר האירועים בכל דף API (מקסימום 500)
    • ‫LOOKBACK_HOURS: מספר השעות שצריך לחזור אחורה בהרצה הראשונה (לדוגמה, 24)

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

  4. מחכים שהפריסה תסתיים בהצלחה.

יצירת משימה ב-Cloud Scheduler

יוצרים משימה של Cloud Scheduler כדי להפעיל את פונקציית Cloud Run במרווחי זמן קבועים.

  1. במסוף Google Cloud, עוברים אל Cloud Scheduler.
  2. לוחצים על Create Job (יצירת משימה).

  3. מספקים את פרטי ההגדרה הבאים:

    • שם: מזינים druva-backup-scheduler
    • אזור: בוחרים את אותו אזור שבו פועלת פונקציית Cloud Run (לדוגמה, us-central1)
    • תיאור: מזינים Triggers Druva Backup log collection every 30 minutes
    • תדירות: מזינים */30 * * * * (כל 30 דקות)
    • אזור זמן: בוחרים את אזור הזמן המועדף (לדוגמה, UTC)

  4. לוחצים על Continue.

  5. מגדירים את היעד:

    • Target type (סוג היעד): בוחרים באפשרות Pub/Sub.
    • נושא ב-Cloud Pub/Sub: בוחרים באפשרות druva-backup-trigger
    • גוף ההודעה: מזינים {"trigger": "scheduled"}

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

בדיקת המשימה ב-Cloud Scheduler

  1. ברשימה Cloud Scheduler, מאתרים את druva-backup-scheduler.
  2. לוחצים על הפעלה בכוח כדי להפעיל את הפונקציה באופן מיידי.
  3. כדי לאמת את ההפעלה, בודקים את:
    • היומנים של פונקציית Cloud Run נמצאים ב-Cloud Run functions > druva-backup-to-gcs > Logs
    • קטגוריית GCS לקובצי NDJSON חדשים ב-Cloud Storage > druva-backup-logs

אחזור חשבון השירות של Google SecOps והגדרת הפיד

קבלת כתובת האימייל בחשבון השירות

  1. עוברים אל SIEM Settings > Feeds (הגדרות SIEM > פידים).
  2. לוחצים על הוספת פיד חדש.
  3. לוחצים על Configure a single feed (הגדרת פיד יחיד).
  4. בשדה שם הפיד, מזינים שם לפיד (לדוגמה, Druva Backup Events).
  5. בוחרים באפשרות Google Cloud Storage V2 בתור סוג המקור.
  6. בוחרים באפשרות Druva Backup בתור סוג היומן.

  7. לוחצים על קבלת חשבון שירות. תוצג כתובת האימייל בחשבון שירות ייחודי, לדוגמה:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  8. מעתיקים את כתובת האימייל הזו לשימוש בשלב הבא.

הגדרת הפיד

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

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

    • כתובת ה-URL של קטגוריית האחסון: מזינים את ה-URI של קטגוריית GCS עם נתיב הקידומת:

      gs://druva-backup-logs/druva_backup/

    • אפשרות למחיקת המקור: בוחרים את אפשרות המחיקה לפי ההעדפה שלכם:

      • אף פעם: לא מוחקים קבצים אחרי ההעברה (מומלץ לבדיקות)
      • מחיקת קבצים שהועברו: מחיקת קבצים אחרי העברה מוצלחת
      • מחיקת קבצים שהועברו וספריות ריקות: מחיקת קבצים וספריות ריקות אחרי העברה מוצלחת

    • הגיל המקסימלי של הקובץ: כולל קבצים ששונו במספר הימים האחרון (ברירת המחדל היא 180 ימים)

    • מרחב השמות של הנכס: מרחב השמות של הנכס

    • תוויות להוספה: התווית שתתווסף לאירועים מהפיד הזה

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

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

הענקת הרשאות IAM לחשבון השירות של Google SecOps

לחשבון השירות של Google SecOps צריך להיות התפקיד Storage Object Viewer בקטגוריה של GCS כדי לקרוא את קובצי היומן שנכתבו על ידי פונקציית Cloud Run.

  1. עוברים אל Cloud Storage > Buckets (קטגוריות).
  2. לוחצים על שם הקטגוריה (לדוגמה, druva-backup-logs).
  3. עוברים לכרטיסייה Permissions.
  4. לוחצים על הענקת גישה.
  5. מספקים את פרטי ההגדרה הבאים:
    • הוספת ישויות מורשות: מדביקים את כתובת האימייל בחשבון השירות של Google SecOps (לדוגמה, chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com)
    • הקצאת תפקידים: בוחרים באפשרות צפייה באובייקטים באחסון.

  6. לוחצים על Save.

טבלת מיפוי UDM

שדה ביומן מיפוי UDM היגיון
inSyncUserID, eventsGroupId, FilesMissed, FilesBackedup, TotalBackupSize, TotalBytesTransferred, facility, inSyncDataSourceID, initiator, event_type additional.fields אם השדה לא ריק, הוא ימוזג עם התוויות שנוצרו מכל שדה.
יוזם הפעילות extensions.auth.type הערך שמוגדר הוא AUTHTYPE_UNSPECIFIED אם היוזם תואם לביטוי הרגולרי של כתובת האימייל
metadata.event_type הערך הוא USER_LOGIN אם has_target_user הוא true ו-has_principal הוא true; הערך הוא STATUS_UPDATE אם has_principal הוא true ו-has_target הוא false; אחרת הערך הוא GENERIC_EVENT
eventID metadata.product_log_id הומר למחרוזת
metadata.product_name ההגדרה היא DRUVA_BACKUP
clientVersion metadata.product_version הערך הועתק ישירות
inSyncDataSourceName principal.asset.hostname הערך הועתק ישירות
ip principal.asset.ip המיזוג בוצע מכתובת ה-IP
inSyncDataSourceName principal.hostname הערך הועתק ישירות
ip principal.ip המיזוג בוצע מכתובת ה-IP
clientOS principal.platform הערך הוא LINUX אם יש התאמה ל-‎ (?i)Linux;‏ WINDOWS אם יש התאמה ל-‎ (?i)windows;‏ MAC אם יש התאמה ל-‎ (?i)mac
profileName principal.resource.name הערך הועתק ישירות
profileID principal.resource.product_object_id הומר למחרוזת
eventState security_result.action הערך הוא ALLOW אם יש התאמה ל-Success (לא תלוי באותיות קטנות או גדולות) אחרת הערך הוא BLOCK
eventState security_result.action_details הערך הועתק ישירות
severity security_result.severity הערך הוא LOW אם הוא נמצא בטווח [0,1,2,3,LOW];‏ MEDIUM אם הוא נמצא בטווח [4,5,6,MEDIUM,SUBSTANTIAL,INFO];‏ HIGH אם הוא נמצא בטווח [7,8,HIGH,SEVERE];‏ CRITICAL אם הוא נמצא בטווח [9,10,VERY-HIGH,CRITICAL]
inSyncUserEmail, initiator target.user.email_addresses מוזג מ-inSyncUserEmail; גם מהיוזם אם יש התאמה לביטוי הרגולרי של כתובת האימייל
inSyncUserName target.user.userid הערך הועתק ישירות
metadata.vendor_name ההגדרה היא DRUVA_BACKUP

יומן שינויים

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

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