איסוף יומנים של פלטפורמת MuleSoft Anypoint

נתמך ב:

במאמר הזה מוסבר איך להטמיע אירועים של נתיב ביקורת מיומנים של פלטפורמת MuleSoft Anypoint ב-Google Security Operations באמצעות Cloud Storage.

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

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

  • מכונה של Google SecOps
  • פרויקט ב- Google Cloud עם Cloud Storage API מופעל
  • הרשאות ליצירה ולניהול של קטגוריות של Cloud Storage
  • הרשאות לניהול כללי מדיניות של ניהול זהויות והרשאות גישה (IAM) בקטגוריות של Cloud Storage
  • הרשאות ליצירת פונקציות Cloud Run, נושאי Pub/Sub ומשימות Cloud Scheduler
  • הרשאות ליצירת חשבונות שירות
  • גישה עם הרשאות ל-MuleSoft Anypoint Platform

איך מקבלים את מזהה הארגון ב-MuleSoft

  1. נכנסים אל Anypoint Platform.
  2. עוברים אל ניהול גישה > ארגונים.
  3. בטבלה Business Groups, לוחצים על שם הארגון.
  4. מעתיקים את מזהה הארגון (לדוגמה, 0a12b3c4-d5e6-789f-1021-1a2b34cd5e6f).

לחלופין, עוברים אל MuleSoft Business Groups (קבוצות עסקיות של MuleSoft) ומעתיקים את המזהה מכתובת ה-URL.

יצירת האפליקציה המקושרת של MuleSoft

  1. נכנסים אל Anypoint Platform.
  2. עוברים אל Access Management > Connected Apps > Create App.
  3. מזינים את פרטי ההגדרה הבאים:
    • שם האפליקציה: מזינים שם ייחודי (לדוגמה, Google SecOps export).
    • בוחרים באפשרות האפליקציה פועלת בשם עצמה (פרטי כניסה של לקוח).
  4. לוחצים על הוספת היקפים > מציג יומני ביקורת > הבא.
  5. בוחרים את כל קבוצות המיקומים שרלוונטיים לעסק שאתם צריכים את היומנים שלהן.
  6. לוחצים על הבא > הוספת היקפים.

  7. לוחצים על Save ומעתיקים את מזהה לקוח ואת סוד לקוח.

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

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

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

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

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

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

כדי להשתמש בפונקציית Cloud Run, צריך חשבון שירות עם הרשאות כתיבה בקטגוריה של Cloud Storage.

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

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

התפקידים האלה נדרשים כדי:

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

הקצאת הרשאות IAM בקטגוריה של Cloud Storage

נותנים לחשבון השירות הרשאות מסוג write בקטגוריה של Cloud Storage, באופן הבא:

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

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

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

  1. במסוף GCP, נכנסים לדף Pub/Sub > Topics.
  2. לוחצים על Create topic.
  3. מזינים את פרטי ההגדרה הבאים:
    • מזהה הנושא: מזינים את הערך mulesoft-audit-trigger.
    • משאירים את ההגדרות האחרות כברירת מחדל.
  4. לוחצים על יצירה.

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

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

  1. במסוף GCP, נכנסים ל-Cloud Run.
  2. לוחצים על כתיבת פונקציה.

  3. בקטע Configure, מזינים את פרטי ההגדרה הבאים:

    הגדרה ערך
    שם השירות mulesoft-audit-collector
    אזור בוחרים את האזור שתואם לקטגוריה של Cloud Storage (לדוגמה, us-central1)
    זמן ריצה בוחרים באפשרות Python 3.12 או גרסה עדכנית יותר.

  4. בקטע Trigger, לוחצים על Add trigger ובוחרים באפשרות Pub/Sub trigger.

  5. בחלונית Eventarc trigger, מזינים את פרטי ההגדרה הבאים:

    • שם הטריגר: משאירים את השם שנוצר או מזינים שם לטריגר.
    • סוג הטריגר: בוחרים באפשרות מקורות של Google.
    • Event provider: בוחרים באפשרות Pub/Sub.
    • סוג האירוע: בוחרים באפשרות google.cloud.pubsub.topic.v1.messagePublished.
    • ‫Select a Cloud Pub/Sub topic: בוחרים את הנושא mulesoft-audit-trigger.
    • ‫Region: בוחרים את אותו אזור כמו של הפונקציה.
    • חשבון שירות: בוחרים את חשבון השירות mulesoft-logs-collector-sa.

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

  7. בקטע אימות:

    1. לוחצים על דרישת אימות.
    2. בודקים את ניהול הזהויות והרשאות הגישה (IAM).

  8. עוברים לקטע Containers, Networking, Security ומרחיבים אותו.

  9. עוברים לכרטיסייה אבטחה:

    • ‫Service account: בוחרים את חשבון השירות של mulesoft-logs-collector-sa.

  10. עוברים לכרטיסייה מאגרי תגים:

    1. לוחצים על Variables & Secrets.
    2. לוחצים על ‎+ Add variable לכל משתנה סביבה:
    שם משתנה ערך לדוגמה תיאור
    GCS_BUCKET mulesoft-audit-logs שם הקטגוריה של Cloud Storage
    GCS_PREFIX mulesoft-audit קידומת לקובצי יומן
    STATE_KEY mulesoft-audit-state.json נתיב המצב, מחוץ לקידומת היומן
    MULE_ORG_ID your_org_id מזהה הארגון ב-MuleSoft
    CLIENT_ID your_client_id מזהה הלקוח של האפליקציה המקושרת
    CLIENT_SECRET your_client_secret סוד הלקוח של האפליקציה המקושרת
    MAX_RECORDS 10000 מספר הרשומות המקסימלי לכל הרצה
    PAGE_SIZE 200 רשומות בכל דף (המספר המקסימלי ב-API הוא 200)
    LOOKBACK_HOURS 24 תקופת המבט לאחור הראשונית בהפעלה הראשונה

  11. עוברים לכרטיסייה Variables & Secrets (משתנים וסודות) ואז ל-Requests (בקשות):

    • ‫Request timeout: מזינים 600 שניות (10 דקות).

  12. עוברים לכרטיסייה הגדרות בקטע מאגרים:

    • בקטע Resources:
      • זיכרון: בוחרים באפשרות ‎512 MiB או יותר.
      • מעבד (CPU): בוחרים באפשרות 1.
    • לוחצים על סיום.

  13. עוברים אל Execution environment:

    • בוחרים באפשרות ברירת מחדל (מומלץ).

  14. בקטע Revision scaling:

    • מספר מינימלי של מכונות: מזינים 0.
    • Maximum number of instances: מזינים 100 (או משנים בהתאם לעומס הצפוי).

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

  16. ממתינים ליצירת השירות (דקה או שתיים).

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

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

  1. מזינים main בשדה Function entry point.

  2. בעורך הקוד, יוצרים שני קבצים:

    • הקובץ הראשון – main.py:

      import functions_framework
from google.cloud import storage
from google.cloud.exceptions import NotFound
import hashlib
import json
import os
import urllib3
from urllib.parse import urlencode
from datetime import datetime, timezone, timedelta
import time

# 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()

# MuleSoft API endpoints
TOKEN_URL = "https://anypoint.mulesoft.com/accounts/api/v2/oauth2/token"
AUDIT_URL = "https://anypoint.mulesoft.com/audit/v2/organizations/{org_id}/query?cursorPagination=true"

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'mulesoft-audit')
# STATE_KEY must stay OUTSIDE GCS_PREFIX. The feed ingests every object under
# its bucket URI and, with a deletion option selected, deletes what it
# transferred. A state file inside the prefix would be ingested as log data and
# then deleted, resetting collection and re-ingesting duplicates.
STATE_KEY = os.environ.get('STATE_KEY', 'mulesoft-audit-state.json')
MULE_ORG_ID = os.environ.get('MULE_ORG_ID')
CLIENT_ID = os.environ.get('CLIENT_ID')
CLIENT_SECRET = os.environ.get('CLIENT_SECRET')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
# The Audit Log Query API returns at most 200 entries per JSON query.
PAGE_SIZE = min(int(os.environ.get('PAGE_SIZE', '200')), 200)
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
# The query re-reads this many minutes before the watermark so that events the
# platform records late are still collected. Re-read events are removed by hash.
OVERLAP_MINUTES = int(os.environ.get('OVERLAP_MINUTES', '2'))

class FetchError(Exception):
    """Raised when the MuleSoft API call fails.

    The watermark must never advance on a failed fetch, otherwise the events in
    the failed window are skipped permanently.
    """

def to_unix_millis(dt: datetime) -> int:
    """Convert datetime to Unix epoch milliseconds."""
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return int(dt.astimezone(timezone.utc).timestamp() * 1000)

def parse_datetime(value: str) -> datetime:
    """Parse an ISO 8601 datetime string into an aware datetime."""
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    dt = datetime.fromisoformat(value)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt

def event_keys(event: dict) -> set:
    """Return the identity this event can be recognised by.

    Audit log entries carry no unique event ID, so a record counts as already
    collected when its full JSON content was seen before. The overlap window
    returns the same JSON again, which this catches without any knowledge of
    the payload schema.
    """
    return {'sha256:' + hashlib.sha256(
        json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()}

def event_time(event: dict) -> str:
    """Return the event timestamp as an ISO 8601 string, or an empty string.

    Audit log entries carry the event time in `timestamp` as Unix milliseconds.
    """
    raw = event.get('timestamp')
    if raw is None or raw == '':
        return ''
    try:
        return datetime.fromtimestamp(int(raw) / 1000, tz=timezone.utc).isoformat()
    except (TypeError, ValueError):
        return ''

@functions_framework.cloud_event
def main(cloud_event):
    """Fetch MuleSoft Anypoint audit logs and write them to Cloud Storage.

    Args:
        cloud_event: CloudEvent object containing the Pub/Sub message.
    """
    if not all([GCS_BUCKET, MULE_ORG_ID, CLIENT_ID, CLIENT_SECRET]):
        # Raise rather than return: a bare return acks the Pub/Sub message and
        # reports the run as successful, silently discarding the schedule tick.
        raise RuntimeError('Missing required environment variables')

    bucket = storage_client.bucket(GCS_BUCKET)
    state = load_state(bucket, STATE_KEY)

    now = datetime.now(timezone.utc)
    watermark = None
    if state.get('last_event_time'):
        watermark = parse_datetime(state['last_event_time'])
    seen_keys = set(state.get('seen_keys', []))

    if watermark is None:
        start_time = now - timedelta(hours=LOOKBACK_HOURS)
    else:
        start_time = watermark - timedelta(minutes=OVERLAP_MINUTES)

    print(f"Fetching audit logs from {start_time.isoformat()} to {now.isoformat()}")

    token = get_token(CLIENT_ID, CLIENT_SECRET)

    # A FetchError here propagates: the run fails, the watermark is untouched,
    # and the next invocation retries the same window.
    records, newest_event_time = fetch_audit(
        org_id=MULE_ORG_ID,
        token=token,
        start_time=start_time,
        end_time=now,
        page_size=PAGE_SIZE,
        max_records=MAX_RECORDS,
    )

    # Remove events already written by an earlier run. Without this, the overlap
    # window re-emits its events on every invocation.
    fresh = [r for r in records if not (event_keys(r) & seen_keys)]
    print(f"Fetched {len(records)} records, {len(fresh)} new after deduplication")

    if not fresh:
        print("No new audit log records found. Watermark left unchanged.")
        return

    if not newest_event_time:
        # A watermark that cannot be derived from the data is not a watermark.
        # Advancing on wall-clock here would skip every late-recorded event.
        raise FetchError('Records were returned but no event timestamp could be parsed')

    timestamp = now.strftime('%Y%m%dT%H%M%SZ')
    object_key = f"{GCS_PREFIX}/audit_{timestamp}.ndjson"
    blob = bucket.blob(object_key)

    ndjson = '\n'.join(json.dumps(record, ensure_ascii=False) for record in fresh) + '\n'
    blob.upload_from_string(ndjson, content_type='application/x-ndjson')

    print(f"Wrote {len(fresh)} records to gs://{GCS_BUCKET}/{object_key}")

    # Advance the watermark only after the data is durably written.
    new_watermark = parse_datetime(newest_event_time)
    if watermark and new_watermark < watermark:
        new_watermark = watermark

    # Retain only the hashes still inside the overlap window, so the state file
    # stays small while covering every event the next query can re-read.
    cutoff = new_watermark - timedelta(minutes=OVERLAP_MINUTES)
    retained = set()
    for record in records:
        stamp = event_time(record)
        if not stamp:
            continue
        if parse_datetime(stamp) >= cutoff:
            retained.update(event_keys(record))

    save_state(bucket, STATE_KEY, {
        'last_event_time': new_watermark.isoformat(),
        'seen_keys': sorted(retained),
    })

    print(f"Successfully processed {len(fresh)} records")

def load_state(bucket, key):
    """Read the collector state from Cloud Storage.

    Only a missing object is treated as a cold start. Any other error is raised:
    swallowing it would silently reset collection to the full lookback window
    and re-ingest that entire period.
    """
    blob = bucket.blob(key)
    try:
        return json.loads(blob.download_as_text())
    except NotFound:
        print('No state file found. Starting from the lookback window.')
        return {}

def save_state(bucket, key, state: dict):
    """Write the collector state to Cloud Storage.

    Failures are raised, not logged. If the state write fails after the data was
    uploaded, the next run repeats the same window and duplicates it.
    """
    blob = bucket.blob(key)
    blob.upload_from_string(
        json.dumps(state, indent=2),
        content_type='application/json',
    )
    print(f"Saved state: last_event_time={state.get('last_event_time')}")

def retry_after_seconds(response, backoff: float) -> int:
    """Parse Retry-After defensively; it may be an HTTP date, which int() cannot parse."""
    raw = response.headers.get('Retry-After')
    try:
        return int(raw) if raw else int(backoff)
    except (TypeError, ValueError):
        return int(backoff)

def get_token(client_id: str, client_secret: str) -> str:
    """Get an OAuth 2.0 access token from MuleSoft using client credentials."""
    body = urlencode({
        'grant_type': 'client_credentials',
        'client_id': client_id,
        'client_secret': client_secret,
    }).encode('utf-8')

    backoff = 1.0
    for attempt in range(5):
        try:
            response = http.request(
                'POST',
                TOKEN_URL,
                body=body,
                headers={'Content-Type': 'application/x-www-form-urlencoded'},
            )
        except Exception as e:
            raise FetchError(f'Token request failed: {e}') from e

        if response.status == 429:
            wait = retry_after_seconds(response, backoff)
            print(f"Rate limited (429) on token request. Retrying after {wait}s...")
            time.sleep(wait)
            backoff = min(backoff * 2, 30.0)
            continue

        if response.status != 200:
            raise FetchError(f'Failed to get token: HTTP {response.status} {response.data.decode("utf-8", "replace")}')

        try:
            return json.loads(response.data.decode('utf-8'))['access_token']
        except (json.JSONDecodeError, KeyError) as e:
            raise FetchError(f'Malformed token response: {e}') from e

    raise FetchError('Rate limited repeatedly on token request; giving up without advancing the watermark')

def fetch_audit(org_id: str, token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
    """Fetch audit log entries from the MuleSoft Audit Log Query API.

    Uses cursor pagination. The API returns entries in ascending timestamp order,
    which a time watermark requires: with newest-first ordering a truncated run
    would advance the watermark past unread events.

    Returns:
        Tuple of (records list, newest_event_time ISO string).

    Raises:
        FetchError: on any API or transport failure, so that the caller does not
            mistake a failed fetch for an empty result and advance the watermark.
    """
    endpoint = AUDIT_URL.format(org_id=org_id)
    headers = {
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json',
        'Accept': 'application/json',
    }

    records = []
    newest_time = None
    cursor = None
    page_num = 0
    backoff = 1.0
    rate_limit_retries = 0
    MAX_RATE_LIMIT_RETRIES = 5

    while True:
        page_num += 1

        if len(records) >= max_records:
            # Stop cleanly. The remaining events stay ahead of the watermark and
            # are collected by the next run.
            print(f"Reached max_records limit ({max_records})")
            break

        # Rebuild the full query on every page; the time window must never be dropped.
        body = {
            'startDate': to_unix_millis(start_time),
            'endDate': to_unix_millis(end_time),
            'limit': page_size,
            'ascending': True,
        }
        if cursor:
            body['cursor'] = cursor

        try:
            response = http.request(
                'POST',
                endpoint,
                body=json.dumps(body).encode('utf-8'),
                headers=headers,
            )
        except Exception as e:
            raise FetchError(f'Request to {endpoint} failed: {e}') from e

        if response.status == 429:
            rate_limit_retries += 1
            if rate_limit_retries > MAX_RATE_LIMIT_RETRIES:
                raise FetchError('Rate limited repeatedly; giving up without advancing the watermark')
            wait = retry_after_seconds(response, backoff)
            print(f"Rate limited (429). Retrying after {wait}s...")
            time.sleep(wait)
            backoff = min(backoff * 2, 30.0)
            continue

        backoff = 1.0
        rate_limit_retries = 0

        if response.status != 200:
            raise FetchError(f'HTTP {response.status} from {endpoint}: {response.data.decode("utf-8", "replace")}')

        try:
            data = json.loads(response.data.decode('utf-8'))
        except json.JSONDecodeError as e:
            raise FetchError(f'Malformed JSON response from {endpoint}: {e}') from e

        page_results = data.get('data') or []
        if not page_results:
            print("No more results (empty page)")
            break

        print(f"Page {page_num}: Retrieved {len(page_results)} entries")
        records.extend(page_results)

        for event in page_results:
            stamp = event_time(event)
            if not stamp:
                continue
            if newest_time is None or parse_datetime(stamp) > parse_datetime(newest_time):
                newest_time = stamp

        cursor = data.get('cursor')
        if not cursor:
            break

    print(f"Retrieved {len(records)} total entries 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 דקות).

שיקולים חשובים

  • הגבלת קצב של יצירת בקשות: נקודת הקצה של שאילתת יומן הביקורת מחילה מגבלות קצב לכל כתובת IP בשלושת מישורי הבקרה. מישור הבקרה בארה"ב מאפשר 700 בקשות לדקה לכל כתובת IP, ואילו מישורי הבקרה באיחוד האירופי וב-Gov מאפשרים 40 בקשות לדקה לכל כתובת IP. הפונקציה מטמיעה השהיה מעריכית לפני ניסיון חוזר (exponential backoff) כדי לטפל בהגבלת קצב של יצירת בקשות באופן אוטומטי.

  • תפוגת תוקף של טוקנים: בדרך כלל, התוקף של טוקני גישה פג כ-30 עד 60 דקות אחרי שהם מונפקים. הפונקציה מבקשת אסימון חדש לכל הפעלה, וזה קורה הרבה לפני שהאסימון הקודם פג תוקף.

  • ביטול כפילויות וקובץ תמונת מצב: הפונקציה שומרת Watermark של זמן האירוע החדש ביותר ואת הגיבובים של האירועים בתוך חלון החפיפה בקובץ תמונת המצב. בכל הפעלה, המערכת קוראת מחדש את הנתונים שתי דקות לפני סימן המים כדי לזהות רשומות שתועדו באיחור, ומוחקת את הרשומות שהיא כבר כתבה. כך, ניסיון חוזר, הפעלה ידנית או עיכוב בתזמון לא יוצרים אירועים כפולים. קריאה שנכשלה ל-API גורמת לכישלון של ההרצה ומשאירה את סימן המים ללא שינוי, כך שההרצה הבאה מנסה שוב את אותו חלון.

  • שמירה של יומני ביקורת: תקופת השמירה שמוגדרת כברירת מחדל ליומני ביקורת היא שנה אחת. אם הארגון שלכם נוצר לפני 10 ביולי 2023 ולא שיניתם באופן ידני את תקופת השמירה, תקופת השמירה היא שש שנים. אם אתם צריכים לשמור את היומנים מעבר לתקופת השמירה שהוגדרה, הורידו אותם מדי פעם.

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

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

  1. במסוף GCP, נכנסים ל-Cloud Scheduler.
  2. לוחצים על Create Job.

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

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

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

בדיקת המשימה המתוזמנת

  1. במסוף Cloud Scheduler, מוצאים את המשימה.
  2. לוחצים על הפעלה ידנית כדי להפעיל את המשימה באופן ידני.
  3. מחכים כמה שניות ונכנסים לדף Cloud Run > Services > mulesoft-audit-collector > Logs.
  4. מוודאים שהפונקציה בוצעה בהצלחה.
  5. בודקים את הקטגוריה של Cloud Storage כדי לוודא שהיומנים נכתבו.

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

  1. עוברים אל הגדרות > ניהול נתונים > פידים. אם אתם משתמשים בפלטפורמה של SIEM בלבד, עוברים אל הגדרות > פידים.
  2. לוחצים על הוספת פיד חדש.
  3. לוחצים על הגדרת פיד יחיד.
  4. בשדה שם הפיד, מזינים שם לפיד (לדוגמה, MuleSoft Logs).
  5. בוחרים באפשרות Google Cloud Storage V2 בתור סוג המקור.
  6. בוחרים באפשרות Mulesoft בתור סוג היומן.

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

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

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

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

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

    • ‫Storage bucket URL: מזינים את ה-URI של הקטגוריה ב-Cloud Storage:

      gs://mulesoft-audit-logs/
      • מחליפים את mulesoft-audit-logs בשם האמיתי של הקטגוריה.

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

      • אף פעם לא למחוק קבצים: אף פעם לא למחוק קבצים מהמקור (מומלץ לבדיקה).
      • Delete transferred files and empty directories: מחיקת קבצים וספריות ריקות מהמקור אחרי שהאחזור הושלם בהצלחה.

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

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

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

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

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

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

לחשבון השירות של Google SecOps נדרשים שני תפקידים בקטגוריה של Cloud Storage: צפייה באובייקט אחסון כדי לקרוא את אובייקטי היומן, ותפקיד ברמת הקטגוריה כדי לקרוא את המטא-נתונים של הקטגוריה.

  1. נכנסים לדף Cloud Storage > Buckets.
  2. לוחצים על שם הקטגוריה.
  3. עוברים לכרטיסייה Permissions.
  4. לוחצים על הענקת גישה.
  5. מזינים את פרטי ההגדרה הבאים:
    • הוספת חשבונות משתמשים: מדביקים את כתובת האימייל של חשבון השירות של Google SecOps.
    • הקצאת תפקידים: בוחרים את שתי האפשרויות הבאות:
    • צפייה באובייקטים של אחסון: קריאת אובייקטים של יומנים.
    • קריאה בקטגוריה באחסון מדור קודם: קריאת המטא-נתונים של הקטגוריה. אם בחרתם באפשרות המחיקה Delete transferred files and empty directories, בחרו במקום זאת באפשרות Storage Legacy Bucket Writer, שגם מעניקה את הרשאת המחיקה.

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

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