איסוף יומני מודיעין איומי סייבר של Group-IB

נתמך ב:

מאמרי עזרה זה מסביר איך להטמיע יומנים של Group-IB מודיעין איומי סייבר ב-Google Security Operations באמצעות Google Cloud Storage.

‫Group-IB Threat Intelligence & Attribution (TI&A) היא פלטפורמה לניתוח מודיעין איומים ברשת, שמספקת נתונים בזמן אמת על גורמי איומים, אינדיקטורים לפריצה (IOC), תוכנות זדוניות, תשתית של שליטה ובקרה (C2), פרטי כניסה שנפרצו, קמפיינים של פישינג ונקודות חולשה. הוא אוסף מידע ממקורות באינטרנט הגלוי, העמוק והאפל, ומאפשר לצוותי האבטחה לזהות איומים ולהגיב להם באופן יזום.

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

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

  • מכונה של Google SecOps
  • חשבון Group-IB TI&A עם גישת API מופעלת
  • גישה לפורטל TI&A של Group-IB‏ (tap.group-ib.com)
  • פרויקט ב-Google Cloud שבו מופעלים ממשקי ה-API הבאים:
    • Cloud Storage API
    • Cloud Functions API
    • Cloud Scheduler API
    • Cloud Build API

יצירת מפתח API של Group-IB

  1. נכנסים לפורטל Group-IB TI&A בכתובת https://tap.group-ib.com.
  2. לוחצים על השם בפינה השמאלית העליונה ובוחרים באפשרות פרופיל.
  3. לוחצים על מעבר להגדרות שלי.
  4. עוברים לכרטיסייה אבטחה וגישה.
  5. בקטע Personal token (טוקן אישי), לוחצים על Generate new token (יצירת טוקן חדש).

  6. מעתיקים ושומרים את מפתח ה-API בצורה מאובטחת.

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

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

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

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

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

פריסת Cloud Function כדי לשלוף נתונים של Group-IB

יוצרים פונקציית Cloud Functions ששולפת נתוני מודיעין איומי סייבר מ-Group-IB TI&A API וכותבת אותם לקטגוריית GCS כקובצי NDJSON לצורך הטמעה ב-Google SecOps.

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

  1. נכנסים אל מסוף Google Cloud.
  2. עוברים אל Cloud Functions.
  3. לוחצים על Create Function.

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

    הגדרה ערך
    סביבה דור 2
    שם הפונקציה groupib-to-gcs
    אזור בוחרים את האזור שהכי קרוב לדלי GCS
    סוג הטריגר HTTPS
    אימות נדרש אימות
    הזיכרון שהוקצה ‫512MB (צריך להגדיל את הערך הזה אם מאחזרים אוספים גדולים)
    פסק זמן ‫540 שניות

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

  6. מגדירים את Runtime ל-Python 3.11 (או גרסה מאוחרת יותר).

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

  8. מחליפים את התוכן של main.py בקוד הבא:

    import json
import os
import requests
import functions_framework
from datetime import datetime, timedelta, timezone
from urllib.parse import urljoin
from requests.auth import HTTPBasicAuth
from google.cloud import storage

GIB_API_URL = os.environ.get('GIB_API_URL', 'https://tap.group-ib.com/api/v2/')
GIB_USERNAME = os.environ.get('GIB_USERNAME')
GIB_API_KEY = os.environ.get('GIB_API_KEY')
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'groupib-ti')
COLLECTIONS = os.environ.get('GIB_COLLECTIONS', 'compromised/account,malware/cnc,apt/threat,hi/threat').split(',')
DEFAULT_DAYS_BACK = int(os.environ.get('DEFAULT_DAYS_BACK', '3'))

# Max items per request: 100 for most collections, 20 for apt/threat and hi/threat
BIG_DATA_COLLECTIONS = ['apt/threat', 'hi/threat']

# File to persist seqUpdate values between runs (use GCS for durability)
STATE_BLOB = '_state/seq_updates.json'

def load_state():
    """Load seqUpdate state from GCS."""
    client = storage.Client()
    bucket = client.bucket(GCS_BUCKET)
    blob = bucket.blob(f"{GCS_PREFIX}/{STATE_BLOB}")
    if blob.exists():
        return json.loads(blob.download_as_text())
    return {}

def save_state(state):
    """Save seqUpdate state to GCS."""
    client = storage.Client()
    bucket = client.bucket(GCS_BUCKET)
    blob = bucket.blob(f"{GCS_PREFIX}/{STATE_BLOB}")
    blob.upload_from_string(json.dumps(state), content_type='application/json')

def gib_request(session, url, params=None):
    """Send authenticated GET request to Group-IB API."""
    resp = session.get(url, params=params)
    if resp.status_code == 301:
        raise Exception('IP not whitelisted by Group-IB. Contact Group-IB support.')
    resp.raise_for_status()
    return resp.json()

def get_seq_update_by_date(session, collection, date_str):
    """Get seqUpdate value for a collection starting from a given date."""
    url = urljoin(GIB_API_URL, 'sequence_list')
    data = gib_request(session, url, {'date': date_str, 'collection': collection})
    return data.get('list', {}).get(collection)

def fetch_collection(session, collection, seq_update):
    """Fetch all new items from a collection starting after the given seqUpdate."""
    limit = 20 if collection in BIG_DATA_COLLECTIONS else 100
    url = urljoin(GIB_API_URL, f"{collection}/updated")
    all_items = []
    last_seq = seq_update

    while True:
        data = gib_request(session, url, {'seqUpdate': str(last_seq), 'limit': limit})
        items = data.get('items', [])
        if not items:
            break
        all_items.extend(items)
        last_seq = items[-1].get('seqUpdate')

    return all_items, last_seq

def write_to_gcs(items, collection_name):
    """Write items to GCS as NDJSON."""
    if not items:
        return 0
    client = storage.Client()
    bucket = client.bucket(GCS_BUCKET)
    timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
    safe_name = collection_name.replace('/', '_')
    blob_path = f"{GCS_PREFIX}/{safe_name}_{timestamp}.ndjson"
    blob = bucket.blob(blob_path)
    ndjson = '\n'.join(json.dumps(item, ensure_ascii=False) for item in items) + '\n'
    blob.upload_from_string(ndjson, content_type='application/x-ndjson')
    return len(items)

@functions_framework.http
def main(request):
    """Cloud Function entry point."""
    session = requests.Session()
    session.auth = HTTPBasicAuth(GIB_USERNAME, GIB_API_KEY)
    session.headers.update({'Accept': '*/*'})

    state = load_state()
    total = 0

    for collection in COLLECTIONS:
        collection = collection.strip()
        seq_update = state.get(collection)

        if seq_update is None:
            default_date = (datetime.now(timezone.utc) - timedelta(days=DEFAULT_DAYS_BACK)).strftime('%Y-%m-%d')
            seq_update = get_seq_update_by_date(session, collection, default_date)
            if seq_update is None:
                continue

        items, last_seq = fetch_collection(session, collection, seq_update)
        if items:
            write_to_gcs(items, collection)
            total += len(items)
        state[collection] = last_seq

    save_state(state)
    return json.dumps({'status': 'success', 'total_items': total}), 200

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

    functions-framework==3.*
requests>=2.28.0
google-cloud-storage>=2.0.0

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

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

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

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

    משתנה ערך
    GIB_API_URL https://tap.group-ib.com/api/v2/ (או https://bt.group-ib.com/api/v2/, בהתאם לאזור)
    GIB_USERNAME כתובת האימייל בחשבון Group-IB
    GIB_API_KEY מפתח ה-API של Group-IB (טוקן אישי)
    GCS_BUCKET שם הקטגוריה ב-GCS (לדוגמה, groupib-ti-logs)
    GCS_PREFIX קידומת לקובצי יומן (לדוגמה, groupib-ti)
    GIB_COLLECTIONS רשימה מופרדת בפסיקים של אוספים לאחזור (ראו בהמשך)
    DEFAULT_DAYS_BACK מספר הימים שצריך לחזור אחורה בהרצה הראשונה (ברירת מחדל: 3)

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

אוספים זמינים של Group-IB

מגדירים את המשתנה GIB_COLLECTIONS עם האוספים שרלוונטיים לתרחיש לדוגמה:

אוסף תיאור ערך מקסימלי
compromised/account פרטי כניסה לחשבון שנפרץ (פרטי התחברות, סיסמה, דומיין) 100
compromised/card כרטיסי בנק שנחשפו 100
compromised/mule חשבונות של בלדרי כספים (Money Mules) 100
compromised/imei מספרי IMEI של מכשירים ניידים שנפרצו 100
compromised/file קבצים שנפרצו עם שיוך לתוכנה זדונית 100
attacks/ddos נתונים של התקפות DDoS (כתובות IP, דומיינים של היעד) 100
attacks/deface אירועים של השחתת אתרים 100
attacks/phishing כתובות URL ודומיינים של פישינג 100
attacks/phishing_kit גיבובים של ערכות פישינג ומותגים שהן מכוונות אליהם 100
bp/phishing הגנה על המותג – אירועי פישינג 100
bp/phishing_kit הגנה על המותג – ערכות פישינג 100
hi/threat דוחות על איומים מצד פושעי סייבר (HI) עם IOC ומיפוי MITRE ATT&CK 20
hi/threat_actor פרופילים של גורמי איום פושעי סייבר 100
apt/threat דוחות על איומי APT (מדינתיים) עם IOC ומיפוי MITRE ATT&CK 20
apt/threat_actor פרופילים של גורמי איום מסוג APT 100
malware/cnc אינדיקטורים של שרת Command-and-Control (כתובות IP, דומיינים) 100
malware/malware תיאורים של תוכנות זדוניות ורמות איום 100
malware/targeted_malware דוגמאות של תוכנות זדוניות מטורגטות (גיבובים, שמות קבצים) 100
osi/git_leak דליפות נתונים ממאגרי Git 100
osi/public_leak הדלפות של נתונים ציבוריים (הדבקות, קובצי נתונים) 100
osi/vulnerability נתוני נקודות חולשה עם ציוני CVSS 100
suspicious_ip/tor_node כתובות IP של צומתי יציאה של Tor 100
suspicious_ip/open_proxy פתיחת כתובות IP של שרתי proxy 100
suspicious_ip/socks_proxy כתובות IP של שרת proxy מסוג SOCKS 100

  • דוגמה: כדי לאסוף נתונים על פרטי כניסה שנפרצו, על תשתית C2 ועל APT:

    compromised/account,malware/cnc,apt/threat,hi/threat

תזמון הפונקציה של Cloud Functions

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

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

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

    הגדרה ערך
    שם groupib-to-gcs-schedule
    אזור באותו אזור כמו הפונקציה של Cloud Functions
    תדירות ‫0 */1 * * * (כל שעה) או 0 0 * * * (מדי יום)
    אזור זמן בחירת אזור הזמן

  4. בקטע Configure the execution (הגדרת ההרצה):

    • סוג היעד: בוחרים באפשרות HTTP.
    • כתובת URL: מזינים את כתובת ה-URL שמשמשת כטריגר של הפונקציה של Cloud Functions.
    • שיטת HTTP: בוחרים באפשרות POST.
    • Auth header: בוחרים באפשרות Add OIDC token.
    • חשבון שירות: בוחרים חשבון שירות עם הרשאה roles/cloudfunctions.invoker.

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

בדיקת ייצוא הנתונים

  1. במסוף Cloud Scheduler, לוחצים על Force Run לצד המשימה כדי להפעיל את הפונקציה באופן ידני.
  2. בודקים את היומנים של Cloud Functions ב-Cloud Logging כדי לוודא שהנתונים נמשכו מ-Group-IB.
  3. במסוף Google Cloud, נכנסים לדף Cloud Storage > Buckets.
  4. לוחצים על שם הקטגוריה (לדוגמה, groupib-ti-logs).
  5. עוברים לתיקיית הקידומת (לדוגמה, groupib-ti/).

  6. מוודאים שקבצים חדשים מסוג .ndjson מופיעים בדלי.

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

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

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

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

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

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

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

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

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

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

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

הגדרת פיד ב-Google SecOps כדי להטמיע יומנים של Group-IB מודיעין איומי סייבר

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

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

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

      gs://groupib-ti-logs/groupib-ti/

      מחליפים את:

      • ‫groupib-ti-logs: השם של קטגוריית GCS.
      • ‫groupib-ti: הקידומת או נתיב התיקייה שבהם מאוחסנים היומנים.

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

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

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

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

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

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

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

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