איסוף יומני פעילות של משתמשים ב-Workday

נתמך ב:

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

‫Workday היא פלטפורמה לניהול הון אנושי (HCM) ולניהול פיננסי שמבוססת על ענן. הפלטפורמה יוצרת יומני ביקורת של פעילות משתמשים, אימות ואבטחה. ‫API בארכיטקטורת REST של רישום פעילות המשתמשים מספק גישה פרוגרמטית ליומנים האלה.

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

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

  • מופע של Google SecOps
  • פרויקט Google Cloud עם {storage_api} מופעל
  • הרשאות ליצירה ולניהול של קטגוריות ב-Cloud Storage
  • הרשאות לניהול כללי מדיניות של ניהול זהויות והרשאות גישה (IAM) בדליים של Cloud Storage
  • הרשאות ליצירת שירותי Cloud Run, נושאי Pub/Sub ומשימות Cloud Scheduler
  • גישה עם הרשאות מיוחדות ל-Workday עם הרשאות אדמין אבטחה או הרשאות שוות ערך
  • הפעלת רישום פעילות המשתמשים בדייר Workday

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

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

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

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

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

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

הפעלת רישום ביומן של פעילות המשתמשים

  1. נכנסים לדייר ב-Workday כאדמינים של אבטחה.
  2. בסרגל החיפוש של Workday, מחפשים את המשימה Edit Tenant Setup - System ופותחים אותה.
  3. בקטע אבטחה, מסמנים את התיבה הפעלת רישום של פעילות המשתמשים.
  4. לוחצים על OK (אישור) כדי לשמור את השינויים.

יצירת משתמש מערכת שילוב (ISU)

  1. בסרגל החיפוש של Workday, מחפשים את האפשרות Create Integration System User.
  2. מזינים שם משתמש (לדוגמה, ISU_SIEM_Export) וסיסמה חזקה.
  3. מבטלים את הסימון בתיבת הסימון דרישה לסיסמה חדשה בכניסה הבאה.
  4. לוחצים על OK.
  5. מחפשים את האפשרות Create Security Group ויוצרים Integration System Security Group (Unconstrained).
  6. מוסיפים את ה-ISU‏ (ISU_SIEM_Export) לקבוצת האבטחה הזו.
  7. מחפשים את הדומיין View Domain ואז מאתרים את הדומיין User Activity Logging. נותנים לקבוצת האבטחה של ISU בדומיין הזה גישת קריאה.
  8. מחפשים את האפשרות הפעלת שינויים במדיניות האבטחה ומפעילים את השינויים.

רישום לקוח API ל-OAuth

  1. בסרגל החיפוש של Workday, מחפשים את האפשרות Register API Client for Integrations.
  2. מזינים שם לקוח (לדוגמה, SIEM_OAuth_Client).
  3. בוחרים באפשרות Non-Expiring Refresh Tokens (אסימוני רענון שלא פג תוקפם).
  4. מוסיפים את ההיקף: מערכת (או ההיקף שנדרש ל-API של רישום פעילות המשתמש).
  5. לוחצים על OK.
  6. מעתיקים ושומרים את הפרטים הבאים במקום מאובטח:
    • ‫Client ID: מזהה לקוח ה-API.
    • ‫Client Secret: סוד הלקוח ב-API.
  7. מחפשים את האפשרות Manage Refresh Tokens for Integrations (ניהול טוקנים לרענון שילובים).
  8. בוחרים את חשבון ISU (ISU_SIEM_Export).
  9. יוצרים טוקן רענון חדש ללקוח ה-API.
  10. מעתיקים את טוקן הרענון ושומרים אותו.

קביעת כתובת ה-URL של הדייר

נקודת קצה ל-API מבוססת על הדייר שלכם ב-Workday. הפורמט הוא:

רכיב ערך
Token endpoint https://HOST/ccx/oauth2/TENANT/token
Activity Logging API https://HOST/ccx/api/privacy/v1/TENANT/activityLogging

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

כדי לוודא של-ISU יש את ההרשאות הנדרשות:

  1. נכנסים ל-Workday כאדמינים של אבטחה.
  2. מחפשים את View Integration System User (הצגת משתמש מערכת שילוב) ובוחרים את ה-ISU‏ (ISU_SIEM_Export).
  3. מוודאים שהמשתמש חבר בקבוצת האבטחה עם גישת קריאה לדומיין User Activity Logging.
  4. אם הגישה חסרה, צריך לעדכן את קבוצת האבטחה ולהפעיל את השינויים במדיניות האבטחה.

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

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

    ```bash
# Replace with your actual credentials
CLIENT_ID="your-client-id"
CLIENT_SECRET="your-client-secret"
REFRESH_TOKEN="your-refresh-token"
HOST="your-workday-host"
TENANT="your-tenant"

# Get OAuth access token
TOKEN=$(curl -s -X POST "https://${HOST}/ccx/oauth2/${TENANT}/token" \
    -H "Content-Type: application/x-www-form-urlencoded" \
    -d "grant_type=refresh_token&client_id=${CLIENT_ID}&client_secret=${CLIENT_SECRET}&refresh_token=${REFRESH_TOKEN}" \
    | jq -r '.access_token')

# Test API access
curl -v -H "Authorization: Bearer ${TOKEN}" \
    "https://${HOST}/ccx/api/privacy/v1/${TENANT}/activityLogging?from=2024-01-01T00:00:00Z&to=2024-01-01T01:00:00Z"
```

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

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

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

  1. במסוף GCP, עוברים אל IAM & Admin > Service Accounts.
  2. לוחצים על יצירת חשבון שירות.
  3. מספקים את פרטי ההגדרה הבאים:
    • שם חשבון השירות: מזינים workday-activity-collector-sa.
    • תיאור חשבון השירות: מזינים Service account for Cloud Run function to collect Workday User Activity 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: כתיבת יומנים לקטגוריה של Cloud Storage וניהול קובצי מצב
  • ‫Cloud Run Invoker: מתן הרשאה ל-Pub/Sub להפעיל את הפונקציה
  • Cloud Functions Invoker: מאפשר הפעלה של פונקציות

הענקת הרשאות IAM בקטגוריית Cloud Storage

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

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

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

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

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

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

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

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

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

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

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

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

    • שם הטריגר: משאירים את השם שנוצר או מזינים שם לטריגר.
    • Trigger type (סוג הטריגר): בוחרים באפשרות Google Sources (מקורות Google).
    • Event provider: בוחרים באפשרות Pub/Sub.
    • סוג האירוע: בוחרים באפשרות google.cloud.pubsub.topic.v1.messagePublished.
    • Select a Cloud Pub/Sub topic (בחירת נושא ב-Cloud Pub/Sub): בוחרים את הנושא workday-activity-trigger.
    • אזור: בוחרים את אותו אזור כמו הפונקציה.
    • חשבון שירות: בוחרים את חשבון השירות workday-activity-collector-sa.

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

  7. בקטע Authentication:

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

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

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

    • חשבון שירות: בוחרים את חשבון השירות workday-activity-collector-sa.

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

    1. לוחצים על Variables & Secrets (משתנים וסודות).
    2. לוחצים על + הוספת משתנה לכל משתנה סביבתי:
    שם המשתנה ערך לדוגמה תיאור
    GCS_BUCKET workday-activity-logs שם הקטגוריה של Cloud Storage
    GCS_PREFIX workday קידומת לקובצי יומן
    STATE_KEY workday-state.json נתיב המצב, מחוץ לקידומת היומן
    WORKDAY_HOST your-workday-host שם המארח של Workday
    WORKDAY_TENANT your-tenant שם הדייר/ת ב-Workday
    CLIENT_ID your-client-id מזהה לקוח ב-OAuth
    CLIENT_SECRET your-client-secret סוד לקוח OAuth
    REFRESH_TOKEN your-refresh-token טוקן רענון OAuth
    MAX_RECORDS 10000 מספר הרשומות המקסימלי בכל הפעלה
    PAGE_SIZE 100 רשומות בכל דף
    LOOKBACK_HOURS 24 תקופת המבט לאחור הראשונית

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

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

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

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

  13. בקטע שינוי קנה מידה של העריכה:

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

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

  15. ממתינים ליצירת השירות (1-2 דקות).

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

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

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

  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 hashlib

# 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', 'workday')
STATE_KEY = os.environ.get('STATE_KEY', 'workday-state.json')
WORKDAY_HOST = os.environ.get('WORKDAY_HOST', '')
WORKDAY_TENANT = os.environ.get('WORKDAY_TENANT', '')
CLIENT_ID = os.environ.get('CLIENT_ID', '')
CLIENT_SECRET = os.environ.get('CLIENT_SECRET', '')
REFRESH_TOKEN = os.environ.get('REFRESH_TOKEN', '')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100'))
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))

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)

# Activity Logging entries carry no per-event identifier (taskId and
# sessionId are reused), so the content hash is the identity.
ID_FIELDS = ()

def event_keys(event):
    """Return every identity this event can be recognised by."""
    keys = {'sha256:' + hashlib.sha256(
        json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()}
    for field in ID_FIELDS:
        value = event.get(field)
        if value:
            keys.add(f'{field}:{value}')
    return keys

def get_token():
    """Get OAuth 2.0 access token using refresh token flow."""
    token_url = f"https://{WORKDAY_HOST}/ccx/oauth2/{WORKDAY_TENANT}/token"

    body = f"grant_type=refresh_token&client_id={CLIENT_ID}&client_secret={CLIENT_SECRET}&refresh_token={REFRESH_TOKEN}"

    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Accept': 'application/json'
    }

    backoff = 1.0
    max_retries = 3

    for attempt in range(max_retries):
        response = http.request('POST', token_url, body=body.encode('utf-8'), headers=headers)

        if response.status == 429:
            retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
            print(f"Rate limited (429) on token request. Retrying after {retry_after}s...")
            time.sleep(retry_after)
            backoff = min(backoff * 2, 30.0)
            continue

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

        data = json.loads(response.data.decode('utf-8'))
        return data['access_token']

    raise RuntimeError(f"Failed to get token after {max_retries} retries due to rate limiting")

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch Workday User Activity logs and write to GCS.

    Args:
        cloud_event: CloudEvent object containing Pub/Sub message
    """

    if not all([GCS_BUCKET, WORKDAY_HOST, WORKDAY_TENANT, CLIENT_ID, CLIENT_SECRET, REFRESH_TOKEN]):
        print('Error: Missing required environment variables')
        return

    try:
        bucket = storage_client.bucket(GCS_BUCKET)

        # Load state
        state = load_state(bucket, STATE_KEY)
        seen_keys = load_seen(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"])
                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 logs from {last_time.isoformat()} to {now.isoformat()}")

        # Get access token
        token = get_token()

        # Fetch logs
        records, newest_event_time = fetch_logs(
            token=token,
            start_time=last_time,
            end_time=now,
            page_size=PAGE_SIZE,
            max_records=MAX_RECORDS,
        )

        # Drop the events the overlap window re-read from the previous run
        fresh = [r for r in records if not (event_keys(r) & seen_keys)]
        all_keys = set()
        for record in records:
            all_keys |= event_keys(record)

        if not fresh:
            print("No new log records found.")
            return

        # Write to GCS as NDJSON
        timestamp = now.strftime('%Y%m%d_%H%M%S')
        object_key = f"{GCS_PREFIX}/logs_{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}")

        # Remember every fetched record so the next overlap window skips it
        save_seen(bucket, STATE_KEY, all_keys)

        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(fresh)} records")

    except Exception as e:
        print(f'Error processing logs: {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}")
        raise

    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}")
        raise

def load_seen(bucket, key):
    """Load the event keys written by the previous run (sidecar object)."""
    try:
        blob = bucket.blob(key + '.seen')
        if blob.exists():
            return set(json.loads(blob.download_as_text()))
    except Exception as e:
        print(f"Warning: Could not load seen keys: {e}")
        raise

    return set()

def save_seen(bucket, key, keys):
    """Save the event keys of this run to the sidecar object."""
    try:
        blob = bucket.blob(key + '.seen')
        blob.upload_from_string(
            json.dumps(sorted(keys)),
            content_type='application/json'
        )
        print(f"Saved {len(keys)} event keys for deduplication")
    except Exception as e:
        print(f"Warning: Could not save seen keys: {e}")
        raise

def fetch_logs(token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
    """
    Fetch user activity logs from Workday Activity Logging API with pagination and rate limiting.

    Args:
        token: OAuth 2.0 access token
        start_time: Start time for log query
        end_time: End time for log query
        page_size: Number of records per page
        max_records: Maximum total records to fetch

    Returns:
        Tuple of (records list, newest_event_time ISO string)
    """
    base_url = f"https://{WORKDAY_HOST}/ccx/api/privacy/v1/{WORKDAY_TENANT}/activityLogging"

    headers = {
        'Authorization': f'Bearer {token}',
        'Accept': 'application/json',
        'User-Agent': 'GoogleSecOps-WorkdayCollector/1.0'
    }

    records = []
    newest_time = None
    page_num = 0
    backoff = 1.0
    offset = 0

    from_iso = start_time.strftime('%Y-%m-%dT%H:%M:%SZ')
    to_iso = end_time.strftime('%Y-%m-%dT%H:%M:%SZ')

    while True:
        page_num += 1

        if len(records) >= max_records:
            print(f"Reached max_records limit ({max_records})")
            break

        current_limit = min(page_size, max_records - len(records))
        url = f"{base_url}?from={from_iso}&to={to_iso}&limit={current_limit}&offset={offset}"

        try:
            response = http.request('GET', url, headers=headers)

            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'))

            page_results = data.get('data', [])

            if not page_results:
                print(f"No more results (empty page)")
                break

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

            # Track newest event time
            for event in page_results:
                try:
                    event_time = event.get('requestTime') or event.get('systemAccount', {}).get('signOnDateTime')
                    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 pagination
            total = data.get('total', 0)
            if offset + len(page_results) >= total:
                print(f"Reached end of results (offset={offset + len(page_results)}, total={total})")
                break

            offset += len(page_results)

        except Exception as e:
            print(f"Error fetching logs: {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. לוחצים על Create Job (יצירת משימה).

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

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

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

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

בחירת התדירות בהתאם לנפח היומן ולדרישות זמן האחזור:

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

בדיקת השילוב

  1. במסוף Cloud Scheduler, מאתרים את המשימה (workday-activity-collector-hourly).
  2. כדי להפעיל את הטריגר באופן ידני, לוחצים על הפעלה מאולצת.
  3. מחכים כמה שניות ועוברים אל Cloud Run > Services > workday-activity-collector > Logs.

  4. מוודאים שהפונקציה בוצעה בהצלחה. מחפשים:

    Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
Page 1: Retrieved X events
Wrote X records to gs://workday-activity-logs/workday/logs_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

  5. בודקים את הקטגוריה של Cloud Storage (workday-activity-logs) כדי לוודא שהיומנים נכתבו.

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

  • HTTP 401: בדיקת פרטי הכניסה של OAuth במשתני הסביבה
  • ‫HTTP 403: צריך לוודא של-ISU יש את ההרשאות הנדרשות לדומיין של רישום פעילות המשתמש
  • ‫HTTP 429: הגבלת קצב – הפונקציה תנסה שוב באופן אוטומטי עם השהיה לפני ניסיון חוזר (backoff)
  • השגת אסימון הגישה נכשלה: צריך לוודא שהערכים של WORKDAY_HOST, WORKDAY_TENANT, CLIENT_ID, CLIENT_SECRET ו-REFRESH_TOKEN נכונים

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

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

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

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

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

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

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

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

      gs://workday-activity-logs/workday/
      • מחליפים את:
        • ‫workday-activity-logs: שם הקטגוריה שלכם ב-Cloud Storage.
        • ‫workday: קידומת או נתיב אופציונליים שבהם נשמרים היומנים (אם רוצים להשתמש בנתיב הבסיס, משאירים את השדה ריק).

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

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

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

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

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

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

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

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

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

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

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

טבלת מיפוי UDM

שדה ביומן מיפוי UDM היגיון
Account_Locked__Disabled_or_Expired_label additional.fields ממוזג
Authentication_Type_label additional.fields ממוזג
Created_Moment_label additional.fields ממוזג
Current_OMS_Version_label additional.fields ממוזג
Device_is_Trusted_label additional.fields ממוזג
Elapsed_Time__Minutes_label additional.fields ממוזג
Failed_Sign_On_label additional.fields ממוזג
Hire_Date_for_Signon_Worker_label additional.fields ממוזג
Job_Family_for_Signon_Worker_label additional.fields ממוזג
Operating_System_label additional.fields ממוזג
Pay_Rate_type_label additional.fields ממוזג
Request_Originator_label additional.fields ממוזג
SAML_Identity_Provider_label additional.fields ממוזג
Sign-on_Time_label additional.fields ממוזג
Signoff_Time_label additional.fields ממוזג
Signon_Day_of_the_Week_label additional.fields ממוזג
Signon_Hour_of_Day_label additional.fields ממוזג
Tenant_Access_Read_Only_label additional.fields ממוזג
UI_Client_Type_label additional.fields ממוזג
Worker_Active_for_Signon_Worker_label additional.fields ממוזג
Worker_Type_for_Signon_Worker_label additional.fields ממוזג
awaiting_details_label additional.fields ממוזג
data_label additional.fields ממוזג
evp_nme_label additional.fields ממוזג
evpminus_label additional.fields ממוזג
signon_ip_address_label additional.fields ממוזג
@requestTime metadata.event_timestamp ניתוח הכתובות שברשימה: ISO8601
Sign-on_Time metadata.event_timestamp ניתוח הכתובות שברשימה: yyyy-MM-ddTHH:mm:ssZZ
event_type metadata.event_type מיפוי ישיר
status_code metadata.product_event_type מיפוי ישיר
tenant metadata.product_event_type מיפוי ישיר
taskId metadata.product_log_id מיפוי ישיר
User_Agent network.http.parsed_user_agent שם שונה/מיפוי
Browser_Type network.http.user_agent מיפוי ישיר
User_Agent network.http.user_agent מיפוי ישיר
userAgent network.http.user_agent מיפוי ישיר
Session_ID network.session_id מיפוי ישיר
sessionId network.session_id מיפוי ישיר
Client_TLS_Version network.tls.version שם שונה/מיפוי
Worker principal.asset.hostname מיפוי ישיר
ip principal.asset.ip ממוזג
ipaddress principal.asset.ip ממוזג
Worker principal.hostname מיפוי ישיר
manager_name principal.hostname מיפוי ישיר
ip principal.ip ממוזג
ipaddress principal.ip ממוזג
Signon_Worker_label principal.user.attribute.labels ממוזג
Supervisory_Organization_of_Signon_Worker_label principal.user.attribute.labels ממוזג
System_Account_label principal.user.attribute.labels ממוזג
Workday_Account_Sign_On_label principal.user.attribute.labels ממוזג
user_credentials_label principal.user.attribute.labels ממוזג
User_Name principal.user.email_addresses ממוזג
Role principal.user.user_display_name מיפוי ישיר
manager_id principal.user.userid מיפוי ישיר
Active_Session_label security_result.detection_fields ממוזג
Authentication_Failure_Message_label security_result.detection_fields ממוזג
Forgotten_Password_Reset_Request_label security_result.detection_fields ממוזג
Invalid_Credentials_label security_result.detection_fields ממוזג
Is_Device_Managed_label security_result.detection_fields ממוזג
MFA_Exempt_label security_result.detection_fields ממוזג
Password_Changed_label security_result.detection_fields ממוזג
Position_label security_result.detection_fields ממוזג
taskDisplayName security_result.summary מיפוי ישיר
hostname target.hostname מיפוי ישיר
CITY_NME target.location.city מיפוי ישיר
Location target.location.city מיפוי ישיר
COUNTRY_NME target.location.country_or_region מיפוי ישיר
Location_Country_for_Signon_Worker target.location.country_or_region מיפוי ישיר
Location_State target.location.state מיפוי ישיר
STATE_NME target.location.state מיפוי ישיר
Exempt_labels target.resource.attribute.labels ממוזג
browser_type_label target.resource.attribute.labels ממוזג
class_of_instance_label target.resource.attribute.labels ממוזג
device target.resource.resource_subtype מיפוי ישיר
device1 target.resource.resource_subtype מיפוי ישיר
device target.resource.type מיפוי: (?i)(Desktop/Laptop/Mobile/Phone) → DEVICE
device1 target.resource.type מיפוי: (?i)(Desktop/Laptop/Mobile/Phone) → DEVICE
BUSINESSGROUP_NME target.user.department ממוזג
email_name target.user.email_addresses ממוזג
EMPLOYEE_ID target.user.employee_id מיפוי ישיר
Employee_ID target.user.employee_id מיפוי ישיר
FIRST_NME target.user.first_name מיפוי ישיר
EMPLOYEE_TYPE target.user.group_identifiers ממוזג
LAST_NME target.user.last_name מיפוי ישיר
JOBTITLE_NME target.user.title מיפוי ישיר
title target.user.title מיפוי ישיר
target.id target.user.userid מיפוי ישיר
לא רלוונטי metadata.product_name קבוע: WORKDAY_USER_ACTIVITY
לא רלוונטי metadata.vendor_name קבוע: WORKDAY_USER_ACTIVITY
לא רלוונטי target.resource.type קבוע: DEVICE

יומן שינויים

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

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