איסוף יומנים של Mandiant Threat Intelligence Custom IOC

נתמך ב:

במאמר הזה מוסבר איך להטמיע נתונים של אינדיקטורים מותאמים אישית של Mandiant Threat Intelligence ב-Google Security Operations באמצעות Google Cloud Storage V2.

‫Mandiant Threat Intelligence מספק אינדיקטורים עשירים בהקשר של פריצות (IOC), כולל כתובות IP, דומיינים, כתובות URL, גיבוב קבצים וכתובות אימייל. הפלטפורמה מספקת מודיעין איומי סייבר שמבוסס על חקירות של תגובה לאירועים בחזית, ומאפשרת לצוותי אבטחה לזהות ולתעדף איומים שמכוונים לארגון שלהם.

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

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

  • מופע של Google SecOps
  • פרויקט ב-GCP עם Cloud Storage API מופעל
  • הרשאות ליצירה ולניהול של קטגוריות GCS
  • הרשאות לניהול מדיניות IAM בקטגוריות GCS
  • הרשאות ליצירת שירותי Cloud Run, נושאי Pub/Sub ומשימות Cloud Scheduler
  • מינוי פעיל ל-Mandiant Advantage Threat Intelligence
  • גישה מורשית ל-Mandiant Advantage portal

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

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

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

איסוף פרטי כניסה ל-Mandiant Advantage

קבלת כתובת ה-URL של אתר Mandiant

  1. נכנסים למופע של Mandiant Advantage.
  2. מעתיקים את כתובת האתר מסרגל הכתובות בדפדפן.
    • כתובת ה-URL היא: https://advantage.mandiant.com

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

  1. נכנסים אל Mandiant Advantage.
  2. עוברים אל הגדרות.
  3. גוללים למטה אל API Access and Keys (גישה לממשקי API ומפתחות) או בוחרים באפשרות הזו בתפריט הניווט.
  4. לוחצים על קבלת מזהה מפתח וסוד.
  5. מעתיקים ושומרים את פרטי הכניסה ל-API בצורה מאובטחת:
    • מזהה מפתח: מעתיקים את הערך הזה (נקרא גם מזהה מפתח API או מפתח ציבורי)
    • Key Secret: מעתיקים את הערך הזה (נקרא גם API Secret או Private Key).

אימות ההרשאות

כדי לוודא שלחשבון יש את ההרשאות הנדרשות:

  1. נכנסים ל-Mandiant Advantage.
  2. עוברים אל הגדרות.
  3. בודקים את הרשאות המשתמש בקטע פרטי משתמש.
  4. מוודאים שיש לכם גישה למודול Threat Intelligence.
  5. אם אתם לא רואים את הנתונים של Threat Intelligence, פנו לאדמין של Mandiant כדי לקבל את ההרשאות המתאימות.

בדיקת הגישה ל-API

  • לפני שממשיכים בשילוב, כדאי לבדוק את פרטי הכניסה:

    # Replace with your actual credentials
    API_KEY_ID="your-key-id"
    API_SECRET="your-key-secret"
    API_BASE="https://api.intelligence.mandiant.com"
    
    # Get OAuth token
    TOKEN=$(curl -s -X POST "${API_BASE}/token" \
        -H "Content-Type: application/x-www-form-urlencoded" \
        -d "grant_type=client_credentials" \
        -u "${API_KEY_ID}:${API_SECRET}" | jq -r '.access_token')
    
    # Test API access
    curl -v -H "Authorization: Bearer ${TOKEN}" \
        "${API_BASE}/v4/indicator?limit=1"
    

אם הבדיקה מצליחה, אמורה להתקבל תגובת JSON עם נתוני אינדיקטורים. אם מופיעה הודעת שגיאה:

  • HTTP 401: צריך לוודא שמזהה מפתח ה-API והסוד נכונים
  • HTTP 403: מוודאים שלחשבון יש הרשאות Threat Intelligence
  • HTTP 404: מוודאים שכתובת הבסיס של ה-API נכונה

יצירת חשבון שירות לפונקציית Cloud Run

פונקציית Cloud Run צריכה חשבון שירות עם הרשאות כתיבה לקטגוריית GCS והרשאה להפעלה על ידי Pub/Sub.

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

  1. במסוף GCP, נכנסים אל IAM & Admin > Service Accounts.
  2. לוחצים על יצירת חשבון שירות.
  3. מספקים את פרטי ההגדרה הבאים:
    • שם חשבון השירות: מזינים mandiant-ioc-collector-sa (לדוגמה, mandiant-ioc-collector-sa)
    • תיאור חשבון השירות: מזינים Service account for Cloud Run function to collect Mandiant IOC logs
  4. לוחצים על יצירה והמשך.
  5. בקטע Grant this service account access to project (מתן גישה של חשבון השירות הזה לפרויקט), מוסיפים את התפקידים הבאים:
    1. לוחצים על בחירת תפקיד.
    2. מחפשים את Storage Object Admin ובוחרים בו.
    3. לוחצים על + הוספת תפקיד נוסף.
    4. מחפשים את התפקיד Cloud Run Invoker (הפעלת שירותים ב-Cloud Run) ולוחצים עליו.
    5. לוחצים על + הוספת תפקיד נוסף.
    6. מחפשים את האפשרות Cloud Functions Invoker (הפעלת פונקציות של Cloud Functions) ולוחצים עליה.
  6. לוחצים על Continue.
  7. לוחצים על סיום.

התפקידים האלה נדרשים כדי לבצע את הפעולות הבאות: - Storage Object Admin: כתיבת יומנים לקטגוריית GCS וניהול קובצי מצב - Cloud Run Invoker: מתן הרשאה ל-Pub/Sub להפעיל את הפונקציה - Cloud Functions Invoker: מתן הרשאה להפעלת הפונקציה

מתן הרשאות IAM בקטגוריית GCS

נותנים לחשבון השירות הרשאות כתיבה בקטגוריית GCS:

  1. נכנסים אל Cloud Storage > Buckets (קטגוריות).
  2. לוחצים על שם הקטגוריה (לדוגמה, mandiant-ioc-logs).
  3. עוברים לכרטיסייה Permissions.
  4. לוחצים על הענקת גישה.
  5. מספקים את פרטי ההגדרה הבאים:
    • Add principals (הוספת ישויות מורשות): מזינים את כתובת האימייל בחשבון השירות (לדוגמה, mandiant-ioc-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • הקצאת תפקידים: בוחרים באפשרות אדמין של אובייקט אחסון.
  6. לוחצים על Save.

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

יוצרים נושא Pub/Sub ש-Cloud Scheduler יפרסם בו ופונקציית Cloud Run תירשם אליו.

  1. במסוף GCP, עוברים אל Pub/Sub > Topics.
  2. לוחצים על יצירת נושא.
  3. מספקים את פרטי ההגדרה הבאים:
    • מזהה הנושא: מזינים mandiant-ioc-trigger (לדוגמה, mandiant-ioc-trigger)
    • השארת ההגדרות האחרות כברירת מחדל
  4. לוחצים על יצירה.

יצירת פונקציית Cloud Run לאיסוף יומנים

הפונקציה של Cloud Run תופעל על ידי הודעות Pub/Sub מ-Cloud Scheduler כדי לאחזר נתוני IOC מ-Mandiant Threat Intelligence API ולכתוב אותם ב-GCS.

  1. במסוף GCP, עוברים אל Cloud Run.
  2. לוחצים על יצירת שירות.
  3. בוחרים באפשרות Function (שימוש בעורך מוטבע ליצירת פונקציה).
  4. בקטע Configure (הגדרה), מזינים את פרטי ההגדרה הבאים:

    הגדרה ערך
    שם השירות mandiant-ioc-collector (לדוגמה, mandiant-ioc-collector)
    אזור בוחרים אזור שתואם ל-GCS Bucket (לדוגמה, us-central1)
    זמן ריצה בוחרים באפשרות Python 3.12 ואילך
  5. בקטע Trigger (optional) (טריגר (אופציונלי)):

    1. לוחצים על + הוספת טריגר.
    2. בוחרים באפשרות Cloud Pub/Sub.
    3. בקטע Select a Cloud Pub/Sub topic, בוחרים את נושא ה-Pub/Sub (לדוגמה, mandiant-ioc-trigger).
    4. לוחצים על Save.
  6. בקטע אימות:

    1. לוחצים על דרישת אימות.
    2. בודקים את ניהול זהויות והרשאות גישה (IAM).
  1. גוללים למטה ומרחיבים את Containers, Networking, Security (מאגרי נתונים, רשתות, אבטחה).
  2. עוברים לכרטיסייה אבטחה:
    • חשבון שירות: בוחרים את חשבון השירות (לדוגמה, mandiant-ioc-collector-sa)
  3. עוברים לכרטיסייה מאגרי תגים:

    1. לוחצים על Variables & Secrets (משתנים וסודות).
    2. לוחצים על + הוספת משתנה לכל משתנה סביבתי:
    שם המשתנה ערך לדוגמה תיאור
    GCS_BUCKET mandiant-ioc-logs שם קטגוריית GCS
    GCS_PREFIX mandiant-ioc תוספת לשם של קובצי יומן
    STATE_KEY mandiant-ioc/state.json נתיב קובץ המצב
    API_BASE https://api.intelligence.mandiant.com כתובת בסיסית של API
    API_KEY_ID your-key-id מזהה מפתח Mandiant API
    API_SECRET your-key-secret Mandiant API Secret
    MAX_RECORDS 1000 מספר הרשומות המקסימלי בכל הפעלה
    PAGE_SIZE 100 רשומות בכל דף
    LOOKBACK_HOURS 24 תקופת מבט לאחור ראשונית
  4. בקטע Variables & Secrets, גוללים למטה אל Requests:

    • זמן קצוב לתפוגת בקשה: מזינים 600 שניות (10 דקות)
  5. עוברים לכרטיסייה הגדרות:

    • בקטע מקורות מידע:
      • זיכרון: בוחרים באפשרות 512MiB או יותר
      • מעבד: בוחרים באפשרות 1
  6. בקטע Revision scaling:

    • מספר מינימלי של מופעים: מזינים 0
    • מספר מקסימלי של מופעים: מזינים 100 (או משנים בהתאם לעומס הצפוי)
  7. לוחצים על יצירה.

  8. מחכים עד שהשירות ייווצר (דקה או שתיים).

  9. אחרי שיוצרים את השירות, עורך הקוד המוטבע ייפתח באופן אוטומטי.

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

  1. מזינים main בשדה נקודת כניסה.
  2. בעורך הקוד המוטבע, יוצרים שני קבצים:

    • main.py:

      import functions_framework
      from google.cloud import storage
      import json
      import os
      import urllib3
      from datetime import datetime, timezone, timedelta
      import time
      import base64
      
      # Initialize HTTP client with timeouts
      http = urllib3.PoolManager(
          timeout=urllib3.Timeout(connect=5.0, read=30.0),
          retries=False,
      )
      
      # Initialize Storage client
      storage_client = storage.Client()
      
      # Environment variables
      GCS_BUCKET = os.environ.get('GCS_BUCKET')
      GCS_PREFIX = os.environ.get('GCS_PREFIX', 'mandiant-ioc')
      STATE_KEY = os.environ.get('STATE_KEY', 'mandiant-ioc/state.json')
      API_BASE = os.environ.get('API_BASE', 'https://api.intelligence.mandiant.com')
      API_KEY_ID = os.environ.get('API_KEY_ID')
      API_SECRET = os.environ.get('API_SECRET')
      MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '1000'))
      PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100'))
      LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
      
      def to_unix_seconds(dt: datetime) -> int:
          """Convert datetime to Unix epoch seconds."""
          if dt.tzinfo is None:
              dt = dt.replace(tzinfo=timezone.utc)
          dt = dt.astimezone(timezone.utc)
          return int(dt.timestamp())
      
      def parse_datetime(value: str) -> datetime:
          """Parse ISO datetime string to datetime object."""
          if value.endswith("Z"):
              value = value[:-1] + "+00:00"
          return datetime.fromisoformat(value)
      
      @functions_framework.cloud_event
      def main(cloud_event):
          """
          Cloud Run function triggered by Pub/Sub to fetch Mandiant IOC data and write to GCS.
      
          Args:
              cloud_event: CloudEvent object containing Pub/Sub message
          """
      
          if not all([GCS_BUCKET, API_BASE, API_KEY_ID, API_SECRET]):
              print('Error: Missing required environment variables')
              return
      
          try:
              # Get GCS bucket
              bucket = storage_client.bucket(GCS_BUCKET)
      
              # Load state
              state = load_state(bucket, STATE_KEY)
      
              # Determine time window
              now = datetime.now(timezone.utc)
              last_time = None
      
              if isinstance(state, dict) and state.get("last_event_time"):
                  try:
                      last_time = parse_datetime(state["last_event_time"])
                      # Overlap by 2 minutes to catch any delayed events
                      last_time = last_time - timedelta(minutes=2)
                  except Exception as e:
                      print(f"Warning: Could not parse last_event_time: {e}")
      
              if last_time is None:
                  last_time = now - timedelta(hours=LOOKBACK_HOURS)
      
              print(f"Fetching IOCs from {last_time.isoformat()} to {now.isoformat()}")
      
              # Convert to Unix timestamps
              start_epoch = to_unix_seconds(last_time)
              end_epoch = to_unix_seconds(now)
      
              # Fetch IOCs
              records, newest_event_time = fetch_iocs(
                  api_base=API_BASE,
                  api_key_id=API_KEY_ID,
                  api_secret=API_SECRET,
                  start_epoch=start_epoch,
                  end_epoch=end_epoch,
                  page_size=PAGE_SIZE,
                  max_records=MAX_RECORDS,
              )
      
              if not records:
                  print("No new IOC records found.")
                  save_state(bucket, STATE_KEY, now.isoformat())
                  return
      
              # Write to GCS as NDJSON
              timestamp = now.strftime('%Y%m%d_%H%M%S')
              object_key = f"{GCS_PREFIX}/iocs_{timestamp}.ndjson"
              blob = bucket.blob(object_key)
      
              ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in records]) + '\n'
              blob.upload_from_string(ndjson, content_type='application/x-ndjson')
      
              print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}")
      
              # Update state with newest event time
              if newest_event_time:
                  save_state(bucket, STATE_KEY, newest_event_time)
              else:
                  save_state(bucket, STATE_KEY, now.isoformat())
      
              print(f"Successfully processed {len(records)} records")
      
          except Exception as e:
              print(f'Error processing IOCs: {str(e)}')
              raise
      
      def load_state(bucket, key):
          """Load state from GCS."""
          try:
              blob = bucket.blob(key)
              if blob.exists():
                  state_data = blob.download_as_text()
                  return json.loads(state_data)
          except Exception as e:
              print(f"Warning: Could not load state: {e}")
      
          return {}
      
      def save_state(bucket, key, last_event_time_iso: str):
          """Save the last event timestamp to GCS state file."""
          try:
              state = {'last_event_time': last_event_time_iso}
              blob = bucket.blob(key)
              blob.upload_from_string(
                  json.dumps(state, indent=2),
                  content_type='application/json'
              )
              print(f"Saved state: last_event_time={last_event_time_iso}")
          except Exception as e:
              print(f"Warning: Could not save state: {e}")
      
      def get_oauth_token(api_base: str, api_key_id: str, api_secret: str) -> str:
          """Get OAuth 2.0 access token from Mandiant API."""
          token_url = f"{api_base}/token"
      
          # Create Basic Auth header
          auth_string = f"{api_key_id}:{api_secret}"
          auth_bytes = auth_string.encode('utf-8')
          auth_b64 = base64.b64encode(auth_bytes).decode('utf-8')
      
          headers = {
              'Authorization': f'Basic {auth_b64}',
              'Content-Type': 'application/x-www-form-urlencoded',
              'Accept': 'application/json'
          }
      
          body = 'grant_type=client_credentials'
      
          try:
              response = http.request('POST', token_url, body=body, headers=headers)
      
              if response.status != 200:
                  print(f"Token request failed: {response.status}")
                  print(f"Response: {response.data.decode('utf-8')}")
                  raise Exception(f"Failed to get OAuth token: HTTP {response.status}")
      
              token_data = json.loads(response.data.decode('utf-8'))
              return token_data['access_token']
      
          except Exception as e:
              print(f"Error getting OAuth token: {e}")
              raise
      
      def fetch_iocs(api_base: str, api_key_id: str, api_secret: str, start_epoch: int, end_epoch: int, page_size: int, max_records: int):
          """
          Fetch IOCs from Mandiant Threat Intelligence API with pagination and rate limiting.
      
          Args:
              api_base: API base URL
              api_key_id: Mandiant API Key ID
              api_secret: Mandiant API Secret
              start_epoch: Start time in Unix epoch seconds
              end_epoch: End time in Unix epoch seconds
              page_size: Number of records per page
              max_records: Maximum total records to fetch
      
          Returns:
              Tuple of (records list, newest_event_time ISO string)
          """
          # Get OAuth token
          access_token = get_oauth_token(api_base, api_key_id, api_secret)
      
          endpoint = f"{api_base}/v4/indicator"
      
          headers = {
              'Authorization': f'Bearer {access_token}',
              'Accept': 'application/json',
              'Content-Type': 'application/json',
              'User-Agent': 'GoogleSecOps-MandiantCollector/1.0'
          }
      
          records = []
          newest_time = None
          page_num = 0
          backoff = 1.0
          next_url = None
      
          while True:
              page_num += 1
      
              if len(records) >= max_records:
                  print(f"Reached max_records limit ({max_records})")
                  break
      
              # Build request URL
              if next_url:
                  # Use next URL from pagination
                  url = next_url
              else:
                  # First request with time filtering
                  params = []
                  params.append(f"start_epoch={start_epoch}")
                  params.append(f"end_epoch={end_epoch}")
                  params.append(f"limit={min(page_size, max_records - len(records))}")
                  url = f"{endpoint}?{'&'.join(params)}"
      
              try:
                  response = http.request('GET', url, headers=headers)
      
                  # Handle rate limiting with exponential backoff
                  if response.status == 429:
                      retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
                      print(f"Rate limited (429). Retrying after {retry_after}s...")
                      time.sleep(retry_after)
                      backoff = min(backoff * 2, 30.0)
                      continue
      
                  backoff = 1.0
      
                  if response.status != 200:
                      print(f"HTTP Error: {response.status}")
                      response_text = response.data.decode('utf-8')
                      print(f"Response body: {response_text}")
                      return [], None
      
                  data = json.loads(response.data.decode('utf-8'))
      
                  # Extract results
                  page_results = data.get('indicators', [])
      
                  if not page_results:
                      print(f"No more results (empty page)")
                      break
      
                  print(f"Page {page_num}: Retrieved {len(page_results)} IOCs")
                  records.extend(page_results)
      
                  # Track newest event time
                  for ioc in page_results:
                      try:
                          # Extract last_updated timestamp
                          event_time = ioc.get('last_updated')
                          if event_time:
                              if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time):
                                  newest_time = event_time
                      except Exception as e:
                          print(f"Warning: Could not parse event time: {e}")
      
                  # Check for next page
                  next_url = data.get('next')
                  if not next_url:
                      print("No more pages (no next URL)")
                      break
      
              except Exception as e:
                  print(f"Error fetching IOCs: {e}")
                  return [], None
      
          print(f"Retrieved {len(records)} total records from {page_num} pages")
          return records, newest_time
      
    • requirements.txt:

      functions-framework==3.*
      google-cloud-storage==2.*
      urllib3>=2.0.0
      
  3. לוחצים על Deploy (פריסה) כדי לשמור ולפרוס את הפונקציה.

  4. ממתינים עד שהפריסה תושלם (2-3 דקות).

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

‫Cloud Scheduler יפרסם הודעות בנושא Pub/Sub במרווחי זמן קבועים, ויפעיל את הפונקציה של Cloud Run.

  1. במסוף GCP, עוברים אל Cloud Scheduler.
  2. לוחצים על יצירת משימה.
  3. מזינים את פרטי ההגדרה הבאים:

    הגדרה ערך
    שם mandiant-ioc-collector-hourly (לדוגמה, mandiant-ioc-collector-hourly)
    אזור בחירה באותו אזור כמו פונקציית Cloud Run
    תדירות 0 * * * * (כל שעה, בדיוק בשעה)
    אזור זמן בחירת אזור זמן (מומלץ להשתמש ב-UTC)
    סוג הטירגוט Pub/Sub
    נושא בוחרים את נושא ה-Pub/Sub (לדוגמה, mandiant-ioc-trigger).
    גוף ההודעה {} (אובייקט JSON ריק)
  4. לוחצים על יצירה.

אפשרויות לתדירות התזמון

  • בוחרים את התדירות על סמך נפח העדכונים של ה-IOC ודרישות זמן האחזור:

    תדירות ביטוי Cron תרחיש לדוגמה
    כל 5 דקות */5 * * * * נפח גבוה, זמן אחזור נמוך
    כל 15 דקות */15 * * * * נפח חיפושים בינוני
    כל שעה 0 * * * * רגיל (מומלץ)
    כל 6 שעות 0 */6 * * * נפח נמוך, עיבוד באצווה
    יומי 0 0 * * * איסוף נתונים היסטוריים

בדיקת השילוב

  1. במסוף Cloud Scheduler, מחפשים את העבודה (לדוגמה, mandiant-ioc-collector-hourly).
  2. לוחצים על הפעלה בכוח כדי להפעיל את המשימה באופן ידני.
  3. ממתינים כמה שניות.
  4. עוברים אל Cloud Run > Services (שירותים).
  5. לוחצים על שם הפונקציה (לדוגמה, mandiant-ioc-collector).
  6. לוחצים על הכרטיסייה יומנים.
  7. מוודאים שהפונקציה בוצעה בהצלחה. מחפשים:

    Fetching IOCs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
    Page 1: Retrieved X IOCs
    Wrote X records to gs://bucket-name/mandiant-ioc/iocs_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X records
    
  8. נכנסים אל Cloud Storage > Buckets (קטגוריות).

  9. לוחצים על שם הקטגוריה (לדוגמה, mandiant-ioc-logs).

  10. עוברים לתיקיית הקידומת (לדוגמה, mandiant-ioc/).

  11. מוודאים שנוצר קובץ .ndjson חדש עם חותמת הזמן הנוכחית.

אם מופיעות שגיאות ביומנים:

  • HTTP 401: בדיקת מזהה מפתח ה-API והסוד במשתני הסביבה
  • HTTP 403: מוודאים שלחשבון יש הרשאות Threat Intelligence
  • HTTP 429: הגבלת קצב – הפונקציה תנסה שוב באופן אוטומטי עם השהיה לפני ניסיון חוזר (backoff)
  • חסרים משתני סביבה: צריך לוודא שכל המשתנים הנדרשים מוגדרים
  • בקשת הטוקן נכשלה: צריך לוודא שפרטי הכניסה ל-API נכונים ושלא פג התוקף שלהם

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

‫Google SecOps משתמש בחשבון שירות ייחודי כדי לקרוא נתונים ממאגר GCS שלכם. צריך להעניק לחשבון השירות הזה גישה לדלי.

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

  1. עוברים אל SIEM Settings > Feeds (הגדרות SIEM > פידים).
  2. לוחצים על הוספת פיד חדש.
  3. לוחצים על הגדרת פיד יחיד.
  4. בשדה שם הפיד, מזינים שם לפיד (לדוגמה, Mandiant Custom IOC).
  5. בוחרים באפשרות Google Cloud Storage V2 בתור סוג המקור.
  6. בוחרים באפשרות Mandiant Custom IOC בתור סוג היומן.
  7. לוחצים על Get Service Account (קבלת חשבון שירות). תוצג כתובת האימייל בחשבון שירות ייחודי, לדוגמה:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. מעתיקים את כתובת האימייל הזו לשימוש בשלב הבא.

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

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

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

      gs://mandiant-ioc-logs/mandiant-ioc/
      
      • מחליפים את:
        • mandiant-ioc-logs: שם קטגוריית ה-GCS.
        • mandiant-ioc: קידומת או נתיב לתיקייה שבה מאוחסנים יומני ה-IOC.
    • אפשרות למחיקת המקור: בוחרים את אפשרות המחיקה לפי ההעדפה שלכם:

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

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

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

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

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

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

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

לחשבון השירות של Google SecOps צריך להיות התפקיד צפייה באובייקט אחסון בקטגוריה של GCS.

  1. נכנסים אל Cloud Storage > Buckets (קטגוריות).
  2. לוחצים על שם הקטגוריה (לדוגמה, mandiant-ioc-logs).
  3. עוברים לכרטיסייה Permissions.
  4. לוחצים על הענקת גישה.
  5. מספקים את פרטי ההגדרה הבאים:
    • Add principals (הוספת ישויות מורשות): מדביקים את כתובת האימייל בחשבון של חשבון השירות של Google SecOps
    • הקצאת תפקידים: בוחרים באפשרות צפייה באובייקטים באחסון.
  6. לוחצים על Save.

טבלת מיפוי UDM

שדה היומן מיפוי UDM לוגיקה
associated_hashes.value entity.file.md5 הערך מ-associated_hashes.value אם hash.type == "md5", אחרת הערך מ-value אם type == "md5"
value entity.file.md5
associated_hashes.value entity.file.sha1 הערך מ-associated_hashes.value אם hash.type == "sha1", אחרת הערך מ-value אם type == "sha1"
value entity.file.sha1
associated_hashes.value entity.file.sha256 הערך מ-associated_hashes.value אם hash.type == "sha256", אחרת מ-value אם type == "sha256"
value entity.file.sha256
value entity.hostname הערך מועתק ישירות אם הסוג הוא fqdn
value entity.ip הערך מועתק ישירות אם הסוג הוא ipv4
threat_rating.threat_score entity.labels אובייקטים של תוויות שמוזגו עם המפתח threat_score והערך מ-threat_rating.threat_score, המפתח mscore והערך מ-mscore, המפתח is_exclusive והערך מ-is_exclusive, המפתח is_publishable והערך מ-ispublishable, ולכל שדה misp, המפתח misp והערך מ-misp.
mscore entity.labels
is_exclusive entity.labels
is_publishable entity.labels
misp.akamai entity.labels
misp.alexa entity.labels
misp.alexa_1M entity.labels
misp.amazon-aws entity.labels
misp.apple entity.labels
misp.automated-malware-analysis entity.labels
misp.bank-website entity.labels
misp.captive-portals entity.labels
misp.censys-scanning entity.labels
misp.cisco_1M entity.labels
misp.cisco_top1000 entity.labels
misp.cisco_top10k entity.labels
misp.cisco_top20k entity.labels
misp.cisco_top5k entity.labels
misp.cloudflare entity.labels
misp.common-contact-emails entity.labels
misp.common-ioc-false-positive entity.labels
misp.covid entity.labels
misp.covid-19-cyber-threat-coalition-whitelist entity.labels
misp.covid-19-krassi-whitelist entity.labels
misp.crl-hostname entity.labels
misp.crl-ip entity.labels
misp.dax30 entity.labels
misp.digitalside entity.labels
misp.disposable-email entity.labels
misp.dynamic-dns entity.labels
misp.eicar.com entity.labels
misp.empty-hashes entity.labels
misp.fastly entity.labels
misp.findip-host entity.labels
misp.google entity.labels
misp.google-chrome-crux-1million entity.labels
misp.google-gcp entity.labels
misp.google-gmail-sending-ips entity.labels
misp.googlebot entity.labels
misp.ipv6-linklocal entity.labels
misp.majestic_million entity.labels
misp.majestic_million_1M entity.labels
misp.microsoft entity.labels
misp.microsoft-attack-simulator entity.labels
misp.microsoft-azure entity.labels
misp.microsoft-azure-appid entity.labels
misp.microsoft-azure-china entity.labels
misp.microsoft-azure-germany entity.labels
misp.microsoft-azure-us-gov entity.labels
misp.microsoft-office365 entity.labels
misp.microsoft-office365-cn entity.labels
misp.microsoft-office365-ip entity.labels
misp.microsoft-win10-connection-endpoints entity.labels
misp.moz-top500 entity.labels
misp.mozilla-CA entity.labels
misp.mozilla-IntermediateCA entity.labels
misp.multicast entity.labels
misp.nioc-filehash entity.labels
misp.openai-gptbot entity.labels
misp.ovh-cluster entity.labels
misp.parking-domain entity.labels
misp.parking-domain-ns entity.labels
misp.phone_numbers entity.labels
misp.public-dns-hostname entity.labels
misp.public-dns-v4 entity.labels
misp.public-dns-v6 entity.labels
misp.public-ipfs-gateways entity.labels
misp.rfc1918 entity.labels
misp.rfc3849 entity.labels
misp.rfc5735 entity.labels
misp.rfc6598 entity.labels
misp.rfc6761 entity.labels
misp.second-level-tlds entity.labels
misp.security-provider-blogpost entity.labels
misp.sinkholes entity.labels
misp.smtp-receiving-ips entity.labels
misp.smtp-sending-ips entity.labels
misp.stackpath entity.labels
misp.tenable-cloud-ipv4 entity.labels
misp.tenable-cloud-ipv6 entity.labels
misp.ti-falsepositives entity.labels
misp.tlds entity.labels
misp.tranco entity.labels
misp.tranco10k entity.labels
misp.umbrella-blockpage-hostname entity.labels
misp.umbrella-blockpage-v4 entity.labels
misp.umbrella-blockpage-v6 entity.labels
misp.university_domains entity.labels
misp.url-shortener entity.labels
misp.vpn-ipv4 entity.labels
misp.vpn-ipv6 entity.labels
misp.whats-my-ip entity.labels
misp.wikimedia entity.labels
misp.zscaler entity.labels
value entity.url הערך מועתק ישירות אם הסוג הוא url
last_seen event.ioc.active_timerange.end התאריך שנותח מהערך last_seen באמצעות ISO8601
first_seen event.ioc.active_timerange.start התאריך שנותח מהערך של first_seen באמצעות ISO8601
sources.0.category event.ioc.categorization הצטרפות ממערך sources.0.category עם מפריד פסיק
threat_rating.threat_score event.ioc.confidence_score הערך הועתק ישירות
value event.ioc.domain_and_ports.domain הערך מועתק ישירות אם הסוג הוא fqdn
event.ioc.feed_name event.ioc.feed_name ההגדרה היא Mandiant
value event.ioc.ip_and_ports.ip_address הומר ל-ipaddress אם type == "ipv4"
סוג metadata.entity_type הערך הוא FILE אם associated_hashes.type הוא md5 sha1 sha256, אחרת הערך הוא DOMAIN_NAME אם type == "fqdn", ‏ IP_ADDRESS אם type == "ipv4", ‏ URL אם type == "url", ‏ FILE אם type הוא md5 sha1 sha256, ‏ UNKNOWN_ENTITYTYPE אחרת
associated_hashes.type metadata.entity_type
custom_ioc_expire_date metadata.interval.end_time התאריך שנותח מ-custom_ioc_expire_date באמצעות ISO8601 אם הערך של type הוא fqdn ipv4 url
first_seen metadata.interval.start_time התאריך שנותח מהערך של first_seen באמצעות ISO8601 או UNIX
id metadata.product_entity_id הערך הועתק ישירות
threat_det metadata.threat מוזג מ-threat_det
attributed_associations threat_det.associations אובייקטים של threat_association שמוזגו מ-attributed_associations ומ-associations, עם שדות שהוגדרו מ-association.id, ‏ name, ‏ type, ‏ region_code.country_or_region, ‏ associated_actors, ‏ alias, ‏ role, ‏ first_reference_time, ‏ last_reference_time, ‏ industries_affected
שיוכים threat_det.associations
קמפיינים threat_det.campaigns בוצע מיזוג של campaign.name ממערך הקמפיינים
קטגוריה threat_det.category_details מוזג ממערך קטגוריות
threat_rating.confidence_level threat_det.confidence הערך הוא LOW_CONFIDENCE אם ההתאמה נמוכה, MEDIUM_CONFIDENCE אם ההתאמה בינונית, HIGH_CONFIDENCE אם ההתאמה גבוהה, אחרת UNKNOWN_CONFIDENCE
threat_rating.confidence_score threat_det.confidence_details ערך מ-threat_rating.confidence_score שהומר למחרוזת, או מ-mscore אם threat_rating.confidence_score ריק
mscore threat_det.confidence_details
first_seen threat_det.first_discovered_time התאריך מהשדה first_seen אם הוא לא ריק, אחרת מהשדה firstSeen, בפורמט ISO8601 או UNIX
firstSeen threat_det.first_discovered_time
first_seen threat_det.last_updated_time התאריך שנותח מהערך של first_seen באמצעות ISO8601 או UNIX
threat_rating.threat_score threat_det.risk_score הערך מומר למספר נקודה צפה מ-threat_rating.threat_score אם הוא לא ריק, אחרת מ-mscore
mscore threat_det.risk_score
threat_rating.severity_level threat_det.severity מגדירים LOW אם רמת החומרה נמוכה, MEDIUM אם רמת החומרה בינונית, HIGH אם רמת החומרה גבוהה, אחרת מגדירים UNKNOWN_SEVERITY
threat_det.threat_feed_name threat_det.threat_feed_name ההגדרה היא Mandiant
value threat_det.url_back_to_product מגדירים את הערך 'https://advantage.mandiant.com/search?query=%{value}' אם הסוג הוא fqdn ipv4 url md5
verdict_simple.timestamp threat_det.verdict_info המיזוג בוצע מ-verdict_info_simple אחרי הגדרת verdict_time מ-timestamp, ‏ verdict_response מ-verdict אם מדובר בתוכן זדוני, ו-verdict_type מ-verdict_source
verdict_simple.verdict threat_det.verdict_info
verdict_simple.verdict_source threat_det.verdict_info
metadata.product_name metadata.product_name הגדרה כ-MANDIANT_CUSTOM_IOC
metadata.vendor_name metadata.vendor_name הגדרה כ-MANDIANT_CUSTOM_IOC
threat.threat_feed_name Mandiant מופה מיומן השינויים
threat_rating.threat_score entity.entity.labels מופה מיומן השינויים
threat_rating.severity_level threat.severity מופה מיומן השינויים
threat_rating.confidence_level threat.confidence מופה מיומן השינויים
verdict_simple.timestamp verdict_info.verdict_time מופה מיומן השינויים
category threat.category_details מופה מיומן השינויים
threat_rating.confidence_score threat.confidence_details מופה מיומן השינויים
threat_rating.threat_score threat.risk_score מופה מיומן השינויים

שנה רישום

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

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