איסוף יומנים מ-Atlassian Confluence

גרסת המנתח: 5.0

נתמך ב:

במאמר הזה מוסבר איך להטמיע יומנים של Atlassian Confluence ב-Google Security Operations. המנתח מנסה קודם לחלץ שדות מהודעת היומן הגולמית באמצעות ביטויים רגולריים (תבניות grok) שמיועדים ליומנים של Atlassian Confluence. אם הניתוח של Grok נכשל או שהיומן בפורמט JSON, הקוד מנסה לנתח את ההודעה כ-JSON. בשלב האחרון, השדות שחולצו ממופים לסכימת ה-UDM של Google SecOps ומועשרים בהקשר נוסף.

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

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

  • מכונה של Google SecOps
  • חשבון Atlassian Confluence Cloud עם גישה ליומן הביקורת או Confluence Data Center/Server עם הרשאת אדמין
  • בשיטה שמבוססת על GCP: גישה עם הרשאות ל-GCP ‏(GCS, ‏IAM, ‏Cloud Run, ‏Pub/Sub, ‏Cloud Scheduler)
  • לשיטת Bindplane: ‏Windows Server 2016 ואילך, או מארח Linux עם systemd

סקירה כללית של אפשרויות השילוב

המדריך הזה כולל שני מסלולי שילוב:

  • אפשרות 1: Confluence Data Center/Server באמצעות Bindplane + Syslog
  • אפשרות 2: יומני ביקורת של Confluence Cloud באמצעות פונקציית Cloud Run של GCP ו-GCS (בפורמט JSON)

בוחרים את האפשרות שהכי מתאימה לסוג הפריסה ולתשתית של Confluence.

אפשרות 1: Confluence Data Center/Server באמצעות Bindplane + Syslog

האפשרות הזו מגדירה את Confluence Data Center או את Confluence Server לשלוח יומנים דרך syslog לסוכן Bindplane, שמעביר אותם ל-Google SecOps.

קבלת פרטי כניסה ל-Google SecOps בשביל סוכן Bindplane

כדי לאמת את סוכן Bindplane ב-Google SecOps, בוחרים באחת מהשיטות הבאות:

  • פריסות חדשות: משתמשים ב-Chronicle API (אפשרות 1).
  • פריסות קודמות: ממשיכים להשתמש בקובץ האימות להטמעת נתונים (אפשרות 2) עד שה-Ingestion API יופסק ב-20 ביולי 2027.
  1. נכנסים למסוףGoogle Cloud ובוחרים את הפרויקט ב- Google Cloud שמקושר למכונה של Google SecOps.
  2. נכנסים לדף IAM & Admin > Service Accounts.
  3. לוחצים על Create service account.
  4. מזינים את פרטי ההגדרה הבאים:
    • Service account name: מזינים שם לחשבון (לדוגמה, bindplane-ingest).
    • Service account description: מזינים תיאור (לדוגמה, Bindplane agent log ingestion).
  5. לוחצים על Create and continue.
  6. בקטע Grant this service account access to project, לוחצים על Select a role ובוחרים באפשרות Chronicle Editor.
  7. לוחצים על המשך ואז על סיום.
  8. לוחצים על חשבון השירות שיצרתם ועוברים לכרטיסייה Keys.
  9. לוחצים על Add key > Create new key.
  10. בוחרים באפשרות JSON ולוחצים על Create. קובץ המפתח יורד למחשב.
  11. שומרים את קובץ המפתח בתור ingestion-auth.json בספריית Bindplane collector (/opt/observiq-otel-collector/ingestion-auth.json ב-Linux או C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json ב-Windows), או מעדכנים את creds_file_path ב-config.yaml כך שיתאים לנתיב שבו שמרתם אותו.
  12. עוברים אל IAM & Admin > Settings ומעתיקים את הערך של Project number. תצטרכו אותו כדי להגדיר את הייצוא.

אפשרות 2: הורדה של קובץ האימות להטמעת נתונים (גרסה קודמת)

  1. נכנסים למסוף Google SecOps.
  2. עוברים אל הגדרות > ניהול נתונים > סוכני איסוף. אם אתם משתמשים בפלטפורמת SIEM בלבד, עוברים אל הגדרות > סוכני איסוף.
  3. מורידים את קובץ האימות להטמעת נתונים.
  4. שומרים את הקובץ בצורה מאובטחת במערכת שבה יותקן הסוכן של Bindplane.

איך מקבלים את מספר הלקוח ב-Google SecOps

  1. נכנסים למסוף Google SecOps.
  2. עוברים אל הגדרות > ניהול > פרופיל. אם אתם משתמשים בפלטפורמה של SIEM בלבד, עוברים אל הגדרות > פרופיל.

  3. מעתיקים את מספר הלקוח מהקטע פרטי הארגון ושומרים אותו.

התקנת הסוכן של Bindplane

מתקינים את הסוכן של Bindplane במערכת ההפעלה Windows או Linux לפי ההוראות הבאות.

התקנת חלונות

  1. פותחים את שורת הפקודה או את PowerShell כאדמין.

  2. מריצים את הפקודה הבאה:

    msiexec /i "https://github.com/observIQ/bindplane-otel-collector/releases/latest/download/observiq-otel-collector.msi" /quiet

  3. מחכים שההתקנה תסתיים.

  4. כדי לוודא שההתקנה בוצעה, מריצים את הפקודה:

    sc query observiq-otel-collector

השירות אמור להופיע כ-RUNNING.

התקנה ב-Linux

  1. פותחים טרמינל עם הרשאות root או sudo.

  2. מריצים את הפקודה הבאה:

    sudo sh -c "$(curl -fsSlL https://github.com/observIQ/bindplane-otel-collector/releases/latest/download/install_unix.sh)" install_unix.sh

  3. מחכים שההתקנה תסתיים.

  4. כדי לוודא שההתקנה בוצעה, מריצים את הפקודה:

    sudo systemctl status observiq-otel-collector

השירות אמור להופיע כפעיל (פועל).

מקורות מידע נוספים בנושא התקנה

במדריך להתקנת הסוכן של Bindplane מפורטות אפשרויות התקנה נוספות ופתרונות לבעיות.

הגדרת סוכן Bindplane להטמעת syslog ושליחה אל Google SecOps

מאתרים את קובץ התצורה

  • Linux:

    sudo nano /opt/observiq-otel-collector/config.yaml

  • Windows:

    notepad "C:\Program Files\observIQ OpenTelemetry Collector\config.yaml"

עריכת קובץ התצורה

  • מחליפים את כל התוכן של config.yaml בהגדרות הבאות:

    receivers:
  udplog:
    listen_address: "0.0.0.0:514"

exporters:
  chronicle/confluence_logs:
    protocol: https
    # Region of your instance, for example us, europe, asia-southeast1
    location: <LOCATION>
    # Project number of the Google Cloud project linked to the instance
    project: '<PROJECT_NUMBER>'
    compression: gzip
    creds_file_path: '/opt/observiq-otel-collector/ingestion-auth.json'
    customer_id: 'YOUR_CUSTOMER_ID'
    endpoint: chronicle.googleapis.com
    log_type: ATLASSIAN_CONFLUENCE
    raw_log_field: body
    ingestion_labels:
      service: confluence

service:
  pipelines:
    logs/confluence:
      receivers:
        - udplog
      exporters:
        - chronicle/confluence_logs

פרמטרים להגדרה

מחליפים את ה-placeholders הבאים:

  • ‫listen_address: מחליפים את היציאה וכתובת ה-IP לפי הצורך בתשתית שלכם. משתמשים ב-0.0.0.0:514 כדי להאזין לכל הממשקים ביציאה 514.
  • ‫creds_file_path: מעדכנים את הנתיב שבו נשמר קובץ המפתח של חשבון השירות (אפשרות 1):
    • Linux: /opt/observiq-otel-collector/ingestion-auth.json
    • Windows: C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json
  • ‫customer_id: מחליפים את YOUR_CUSTOMER_ID במספר הלקוח האמיתי מהשלב הקודם.
  • ‫location: מחליפים את <LOCATION> באזור של המכונה של Google SecOps (לדוגמה, us,‏ europe או asia-southeast1). המייצא מוסיף את הערך הזה בתחילת chronicle.googleapis.com. לרשימה המלאה, ראו נקודות קצה אזוריות.
  • ‫project: מחליפים את <PROJECT_NUMBER> במספר הפרויקט שמופיע בקטע IAM & Admin > Settings.
  • ‫endpoint: משאירים את הערך chronicle.googleapis.com.

מייצא API להטמעה מדור קודם

  • אם אתם משתמשים בקובץ האימות להטמעת נתונים מאפשרות 2, החליפו את הבלוק exporters בקוד הבא. מגדירים את creds_file_path לנתיב של קובץ האימות להטמעת נתונים. אם המכונה נמצאת מחוץ לארה"ב, מחליפים את malachiteingestion-pa.googleapis.com בנקודת הקצה האזורית (לדוגמה, europe-malachiteingestion-pa.googleapis.com).

    exporters:
  chronicle/confluence_logs:
    compression: gzip
    creds_file_path: '/opt/observiq-otel-collector/ingestion-auth.json'
    customer_id: 'YOUR_CUSTOMER_ID'
    endpoint: malachiteingestion-pa.googleapis.com
    log_type: ATLASSIAN_CONFLUENCE
    raw_log_field: body
    ingestion_labels:
      service: confluence

שמירת קובץ התצורה

אחרי העריכה, שומרים את הקובץ:

  • Linux: לוחצים על Ctrl+O, אחר כך על Enter ואז על Ctrl+X
  • Windows: לוחצים על קובץ > שמירה

מפעילים מחדש את הסוכן של Bindplane כדי להחיל את השינויים.

כדי להפעיל מחדש את סוכן Bindplane ב-Linux:

  1. מריצים את הפקודה הבאה:

    sudo systemctl restart observiq-otel-collector

  2. מוודאים שהשירות פועל:

    sudo systemctl status observiq-otel-collector

  3. בודקים אם יש שגיאות ביומנים:

    sudo journalctl -u observiq-otel-collector -f

כדי להפעיל מחדש את הסוכן של Bindplane ב-Windows:

  1. בוחרים אחת מהאפשרויות האלה:

    • שורת הפקודה או PowerShell כאדמין:
    net stop observiq-otel-collector && net start observiq-otel-collector
    • מסוף השירותים:
      1. לוחצים על Win+R, מקלידים services.msc ולוחצים על Enter.
      2. מאתרים את observIQ OpenTelemetry Collector.
      3. לוחצים לחיצה ימנית ובוחרים באפשרות הפעלה מחדש.

  2. מוודאים שהשירות פועל:

    sc query observiq-otel-collector

  3. בודקים אם יש שגיאות ביומנים:

    type "C:\Program Files\observIQ OpenTelemetry Collector\log\collector.log"

הגדרת העברת Syslog ב-Confluence Data Center/Server

  1. מגדירים את Confluence כך שיכתוב יומנים לקבצים (פעולת ברירת המחדל).

  2. אם rsyslog לא מותקן, מתקינים אותו:

    sudo apt-get install rsyslog  # Debian/Ubuntu
sudo yum install rsyslog      # RHEL/CentOS

  3. יוצרים קובץ תצורה של rsyslog‏ /etc/rsyslog.d/confluence.conf:

    # Forward Confluence logs to Bindplane
$ModLoad imfile

# Application logs
$InputFileName /opt/atlassian/confluence/logs/atlassian-confluence.log
$InputFileTag confluence-app:
$InputFileStateFile stat-confluence-app
$InputFileSeverity info
$InputFileFacility local0
$InputRunFileMonitor

# Audit logs (JSON format in DC/Server)
$InputFileName <confluence-home-directory>/log/audit/audit.log
$InputFileTag confluence-audit:
$InputFileStateFile stat-confluence-audit
$InputFileSeverity info
$InputFileFacility local1
$InputRunFileMonitor

# Forward to Bindplane agent
*.* @@BINDPLANE_AGENT_IP:514

  4. מחליפים את BINDPLANE_AGENT_IP בכתובת ה-IP של סוכן Bindplane (לדוגמה, 192.168.1.100).

  5. משנים את הנתיבים של קובצי היומן בהתאם להתקנה של Confluence:

    • יומני האפליקציות בדרך כלל: <confluence-install>/logs/ או <local-home>/logs/
    • יומני ביקורת: <confluence-home-directory>/log/audit/ (פורמט JSON)
    • כדי למצוא את ספריית הבית של Confluence, עוברים אל הגדרות > הגדרות כלליות > פרטי המערכת ומחפשים את Confluence Home או את Local Home.

  6. מפעילים מחדש את rsyslog:

    sudo systemctl restart rsyslog

אפשרות ב': הגדרת העברת Syslog ב-Log4j2

  1. נכנסים לשרת Confluence באמצעות SSH או RDP.

  2. מאתרים את קובץ התצורה של Log4j2 במיקום:

    <confluence-install>/confluence/WEB-INF/classes/log4j2.xml

  3. עורכים את קובץ התצורה כדי להוסיף Syslog appender:

    <Configuration>
  <Appenders>
    <!-- Existing appenders -->
    <Syslog name="SyslogAppender" 
            host="BINDPLANE_AGENT_IP" 
            port="514" 
            protocol="UDP"
            format="RFC 5424"
            facility="LOCAL0">
      <PatternLayout pattern="%d{ISO8601} %p [%t] [%c{1}] %m%n"/>
    </Syslog>
  </Appenders>

  <Loggers>
    <Root level="info">
      <AppenderRef ref="SyslogAppender"/>
      <!-- Other appender refs -->
    </Root>

    <!-- Audit logger -->
    <Logger name="com.atlassian.confluence.event.events.security.AuditEvent" 
            level="info" 
            additivity="false">
      <AppenderRef ref="SyslogAppender"/>
    </Logger>
  </Loggers>
</Configuration>

  4. מחליפים את BINDPLANE_AGENT_IP בכתובת ה-IP של סוכן Bindplane (לדוגמה, 192.168.1.100).

  5. מפעילים מחדש את Confluence כדי להחיל את השינויים:

    sudo systemctl restart confluence

אפשרות 2: יומני ביקורת של Confluence Cloud באמצעות פונקציית Cloud Run של GCP ו-GCS

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

  1. נכנסים לחשבון Atlassian.
  2. עוברים לכתובת https://id.atlassian.com/manage-profile/security/api-tokens.
  3. לוחצים על יצירת טוקן API.
  4. מזינים תווית לטוקן (לדוגמה, Google Security Operations Integration).
  5. לוחצים על יצירה.
  6. מעתיקים ושומרים את אסימון ה-API בצורה מאובטחת.
  7. רושמים את כתובת ה-URL של האתר ב-Confluence Cloud (לדוגמה, https://yoursite.atlassian.net).

  8. שימו לב לכתובת האימייל של חשבון Atlassian (שמשמשת לאימות).

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

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

  1. נכנסים ל-Confluence Cloud.
  2. לוחצים על סמל ההגדרות בפינה השמאלית העליונה.
  3. אם האפשרות מעקב > יומן ביקורת מופיעה בתפריט הניווט הימני, יש לכם את ההרשאות הנדרשות.
  4. אם האפשרות הזו לא מופיעה, פנו לאדמין שלכם כדי לקבל הרשאת אדמין ב-Confluence.

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

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

    # Replace with your actual credentials
CONFLUENCE_EMAIL="your-email@example.com"
CONFLUENCE_API_TOKEN="your-api-token"
CONFLUENCE_URL="https://yoursite.atlassian.net"

# Test API access
curl -u "${CONFLUENCE_EMAIL}:${CONFLUENCE_API_TOKEN}" \
  -H "Accept: application/json" \
  "${CONFLUENCE_URL}/wiki/rest/api/audit"

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

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

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

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

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

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

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

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

  1. במסוף GCP, נכנסים לדף IAM & Admin > Service Accounts.
  2. לוחצים על Create Service Account.
  3. מזינים את פרטי ההגדרה הבאים:
    • שם חשבון השירות: מזינים confluence-audit-collector-sa.
    • Service account description: מזינים Service account for Cloud Run function to collect Confluence Cloud audit 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. לוחצים על סיום.

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

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

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

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

  1. נכנסים לדף Cloud Storage > Buckets.
  2. לוחצים על שם הקטגוריה.
  3. עוברים לכרטיסייה Permissions.
  4. לוחצים על הענקת גישה.
  5. מזינים את פרטי ההגדרה הבאים:
    • הוספת חשבונות משתמשים: מזינים את כתובת האימייל של חשבון השירות (לדוגמה, confluence-audit-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. מזינים את פרטי ההגדרה הבאים:
    • מזהה הנושא: מזינים את הערך confluence-audit-trigger.
    • משאירים את ההגדרות האחרות כברירת מחדל.
  4. לוחצים על יצירה.

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

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

  1. במסוף GCP, נכנסים ל-Cloud Run.
  2. לוחצים על Create service.
  3. בוחרים באפשרות Function (שימוש בעורך ישיר כדי ליצור פונקציה).

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

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

  5. בקטע טריגר (אופציונלי):

    1. לוחצים על + הוספת טריגר.
    2. בוחרים באפשרות Cloud Pub/Sub.
    3. בקטע Select a Cloud Pub/Sub topic, בוחרים את confluence-audit-trigger.
    4. לוחצים על Save.

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

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

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

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

    • חשבון שירות: בוחרים באפשרות confluence-audit-collector-sa.

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

    1. לוחצים על Variables & Secrets.
    2. לוחצים על ‎+ Add variable לכל משתנה סביבה:
    שם משתנה ערך לדוגמה תיאור
    GCS_BUCKET confluence-audit-logs שם הקטגוריה ב-GCS
    GCS_PREFIX confluence-audit קידומת לקובצי יומן
    STATE_KEY confluence-audit/state.json נתיב קובץ תמונת המצב
    CONFLUENCE_URL https://yoursite.atlassian.net כתובת האתר של Confluence
    CONFLUENCE_EMAIL your-email@example.com כתובת האימייל של החשבון ב-Atlassian
    CONFLUENCE_API_TOKEN your-api-token-here טוקן API
    MAX_RECORDS 1000 מספר הרשומות המקסימלי לכל הרצה

  10. בקטע Variables & Secrets, גוללים למטה אל Requests:

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

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

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

  12. בקטע Revision scaling:

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

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

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

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

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

  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 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', 'confluence-audit/')
STATE_KEY = os.environ.get('STATE_KEY', 'confluence-audit/state.json')
CONFLUENCE_URL = os.environ.get('CONFLUENCE_URL')
CONFLUENCE_EMAIL = os.environ.get('CONFLUENCE_EMAIL')
CONFLUENCE_API_TOKEN = os.environ.get('CONFLUENCE_API_TOKEN')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '1000'))

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

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 Confluence Cloud audit logs and write to GCS.

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

    if not all([GCS_BUCKET, CONFLUENCE_URL, CONFLUENCE_EMAIL, CONFLUENCE_API_TOKEN]):
        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=24)

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

        # Convert to Unix milliseconds
        start_millis = to_unix_millis(last_time)
        end_millis = to_unix_millis(now)

        # Fetch logs
        records, newest_event_time = fetch_logs(
            api_base=CONFLUENCE_URL,
            email=CONFLUENCE_EMAIL,
            api_token=CONFLUENCE_API_TOKEN,
            start_time_ms=start_millis,
            end_time_ms=end_millis,
            max_records=MAX_RECORDS,
        )

        if not records:
            print("No new log 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}/logs_{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 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}")

    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 fetch_logs(api_base: str, email: str, api_token: str, start_time_ms: int, end_time_ms: int, max_records: int):
    """
    Fetch logs from Confluence Cloud Audit API with pagination and rate limiting.

    Args:
        api_base: Confluence site URL
        email: Atlassian account email
        api_token: API token
        start_time_ms: Start time in Unix milliseconds
        end_time_ms: End time in Unix milliseconds
        max_records: Maximum total records to fetch

    Returns:
        Tuple of (records list, newest_event_time ISO string)
    """
    # Clean up URL
    base_url = api_base.rstrip('/')

    # Build authentication header
    auth_string = f"{email}:{api_token}"
    auth_bytes = auth_string.encode('utf-8')
    auth_b64 = base64.b64encode(auth_bytes).decode('utf-8')
    headers = {
        'Authorization': f'Basic {auth_b64}',
        'Accept': 'application/json',
        'User-Agent': 'GoogleSecOps-ConfluenceCollector/1.0'
    }

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

    while True:
        page_num += 1

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

        # Build request URL
        url = f"{base_url}/wiki/rest/api/audit?startDate={start_time_ms}&endDate={end_time_ms}&start={start_index}&limit=100"

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

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

            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:
                    # creationDate is in Unix milliseconds
                    event_time_ms = event.get('creationDate')
                    if event_time_ms:
                        event_dt = datetime.fromtimestamp(event_time_ms / 1000, tz=timezone.utc)
                        event_time = event_dt.isoformat()
                        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 more results
            current_size = data.get('size', 0)
            if current_size < 100:
                print(f"Reached last page (size={current_size} < limit=100)")
                break

            start_index += current_size

        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[:max_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. ממלאים את פרטי ההגדרה הבאים:

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

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

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

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

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

בדיקת השילוב

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

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

    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://bucket-name/prefix/logs_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

  8. נכנסים לדף Cloud Storage > Buckets.

  9. לוחצים על שם הקטגוריה.

  10. עוברים לתיקייה confluence-audit/.

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

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

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

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

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

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

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

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

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

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

    • כתובת ה-URL של קטגוריית האחסון: מזינים את ה-URI של קטגוריית GCS עם נתיב הקידומת:
    gs://confluence-audit-logs/confluence-audit/
    • מחליפים את:
    • ‫confluence-audit-logs: השם של קטגוריית GCS.
    • ‫confluence-audit: נתיב אופציונלי לתיקייה או לתחילית שבהם נשמרים היומנים (משאירים ריק בשביל תיקיית הבסיס).

    לדוגמה:

    • קטגוריית הבסיס: gs://company-logs/
    • עם הקידומת: gs://company-logs/confluence-audit/
    • עם תיקיית משנה: gs://company-logs/confluence/audit/
    • אפשרות מחיקת המקור: בוחרים את אפשרות המחיקה הרצויה:
      • Never: אף פעם לא למחוק קבצים אחרי העברות (מומלץ לבדיקה).
      • מחיקת קבצים שהועברו: מחיקת קבצים אחרי שהם הועברו בהצלחה.
      • מחיקת קבצים שהועברו וספריות ריקות: מחיקת קבצים וספריות ריקות אחרי שההעברה מסתיימת.
    • גיל הקובץ המקסימלי: כולל קבצים ששונו במספר הימים האחרונים. ברירת המחדל היא 180 ימים.
    • מרחב השמות של הנכס: מרחב השמות של הנכס.
    • תוויות הטמעה: התווית שתתווסף לאירועים מהפיד הזה.

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

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

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

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

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

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

טבלת מיפוי UDM

שדה ביומן מיפוי UDM Logic (פונקציה לוגית)
סוכן read_only_udm.network.http.user_agent הערך נלקח מהשדה "סוכן".
app_protocol read_only_udm.network.application_protocol נגזר מהשדה app_protocol. אם "app_protocol" מכיל את הערכים "HTTPS",‏ "HTTP",‏ "SSH" או "RDP", המערכת תשתמש בפרוטוקול המתאים. אחרת, ברירת המחדל היא "UNKNOWN_APPLICATION_PROTOCOL".
app_protocol read_only_udm.network.application_protocol_version הערך נלקח מהשדה app_protocol.
auditType.action read_only_udm.security_result.action נגזר מהשדה "auditType.action". אם "auditType.action" מכיל את המילה "successful", הערך מוגדר כ-"ALLOW". אם הוא מכיל את המילה "restricted", הערך מוגדר כ-"BLOCK".
auditType.action read_only_udm.security_result.summary הערך נלקח מהשדה "auditType.action" כשהשדה "auditType" לא ריק והשדה "auditType_area" הוא "SECURITY".
auditType.actionI18nKey read_only_udm.metadata.product_event_type הערך נלקח מהשדה "auditType.actionI18nKey" כשהשדה "auditType" לא ריק.
auditType.area read_only_udm.security_result.detection_fields.value הערך נלקח מהשדה auditType.area ומוקצה לשדה value של שדה זיהוי, כשהשדה key מוגדר כ-auditType area. המיפוי הזה מתבצע כשהשדה auditType לא ריק.
auditType.category read_only_udm.security_result.category_details הערך נלקח מהשדה "auditType.category" כשהשדה "auditType" לא ריק.
auditType.categoryI18nKey read_only_udm.security_result.detection_fields.value הערך נלקח מהשדה auditType.categoryI18nKey ומוקצה לשדה value של שדה זיהוי, כשהשדה key מוגדר כ-auditType categoryI18nKey. המיפוי הזה מתבצע כשהשדה auditType לא ריק.
auditType.level read_only_udm.security_result.detection_fields.value הערך נלקח מהשדה auditType.level ומוקצה לשדה value של שדה זיהוי, כשהשדה key מוגדר כ-auditType level. המיפוי הזה מתבצע כשהשדה auditType לא ריק.
author.displayName read_only_udm.principal.user.user_display_name הערך נלקח מהשדה "author.displayName".
author.externalCollaborator read_only_udm.security_result.about.resource.attribute.labels.value הערך נלקח מהשדה "author.externalCollaborator" ומוקצה לשדה "value" של תווית שהשדה "key" שלה מוגדר כ-"externalCollaborator".
author.id read_only_udm.principal.user.userid הערך נלקח מהשדה "author.id" כשהשדה "author.type" הוא "user" והשדה "principal_user_present" הוא "false".
author.isExternalCollaborator read_only_udm.security_result.about.resource.attribute.labels.value הערך נלקח מהשדה "author.isExternalCollaborator" ומוקצה לשדה "value" של תווית שהשדה "key" שלה מוגדר כ-"isExternalCollaborator".
author.name read_only_udm.principal.user.user_display_name הערך נלקח מהשדה "author.name" כשהשדה "author.type" הוא "user" והשדה "principal_user_present" הוא "false".
bytes_in read_only_udm.network.received_bytes הערך נלקח מהשדה "bytes_in" אם הוא מכיל ספרות. אחרת, ערך ברירת המחדל הוא 0.
קטגוריה read_only_udm.security_result.category_details הערך נלקח מהשדה "קטגוריה".
changedValues read_only_udm.principal.resource.attribute.labels עובר על כל רכיב ב-changedValues ויוצר תוויות עם מפתחות כמו changedValue [index] [key] וערכים מהערכים התואמים במערך changedValues.
creationDate read_only_udm.metadata.event_timestamp הערך נלקח מהשדה creationDate, ומנותח כחותמת זמן של UNIX או UNIX_MS.
extraAttributes read_only_udm.principal.resource.attribute.labels עובר על כל רכיב ב-"extraAttributes" ויוצר תוויות עם מפתחות שמבוססים על השדות "name" ו-"nameI18nKey" וערכים מהשדה "value" התואם.
http_verb read_only_udm.network.http.method הערך נלקח מהשדה "http_verb".
ip read_only_udm.target.ip הערך נלקח מהשדה "ip".
principal_host read_only_udm.principal.hostname הערך נלקח מהשדה "principal_host".
referral_url read_only_udm.network.http.referral_url הערך נלקח מהשדה referral_url.
remoteAddress read_only_udm.principal.ip הערך נלקח מהשדה remoteAddress, ומנותח ככתובת IP.
response_code read_only_udm.network.http.response_code הערך נלקח מהשדה response_code.
session_duration read_only_udm.additional.fields.value.string_value הערך נלקח מהשדה session_duration ומוקצה לשדה string_value של תווית שהשדה key שלה מוגדר כ-Session Duration.
מקור read_only_udm.principal.ip הערך נלקח מהשדה "מקור", ומנותח ככתובת IP.
src_ip read_only_udm.principal.ip הערך נלקח מהשדה src_ip אם השדה remoteAddress ריק.
סיכום read_only_udm.security_result.summary הערך נלקח מהשדה "סיכום".
sysAdmin read_only_udm.security_result.about.resource.attribute.labels.value הערך נלקח מהשדה sysAdmin ומוקצה לשדה value של תווית שהשדה key שלה מוגדר כ-sysAdmin.
superAdmin read_only_udm.security_result.about.resource.attribute.labels.value הערך נלקח מהשדה "superAdmin" ומוקצה לשדה "value" של תווית שהשדה "key" שלה מוגדר כ-"superAdmin".
target_url read_only_udm.target.url הערך נלקח מהשדה target_url.
חותמת זמן read_only_udm.metadata.event_timestamp הערך נלקח מהשדה timestamp, ומנותח כמחרוזת של תאריך ושעה.
user_id read_only_udm.principal.user.userid הערך נלקח מהשדה "user_id".
read_only_udm.metadata.event_type הערך בשדה הזה נקבע על ידי סדרה של בדיקות, וברירת המחדל היא "GENERIC_EVENT". הוא מוגדר לערכים ספציפיים כמו NETWORK_HTTP,‏ USER_UNCATEGORIZED או STATUS_UPDATE, על סמך הנוכחות והתוכן של שדות אחרים כמו principal_host,‏ user_id,‏ has_principal ו-author.type.
read_only_udm.metadata.vendor_name מגדירים את הערך "ATLASSIAN".
read_only_udm.metadata.product_name צריך להגדיר את הערך "CONFLUENCE".
read_only_udm.metadata.log_type מגדירים את הערך "ATLASSIAN_CONFLUENCE".
read_only_udm.principal.user.user_display_name הערך של השדה הזה יכול להגיע מ-"author.displayName" או מ-"affectedObject.name", בהתאם להקשר.
read_only_udm.target.process.pid הערך של השדה הזה יכול להגיע מ-"principal_host" או מ-"pid", בהתאם להקשר.
read_only_udm.principal.resource.attribute.labels השדה הזה מאוכלס בתוויות שונות שנגזרות משדות כמו "affectedObjects",‏ "changedValues" ו-"extraAttributes". המפתחות והערכים של התוויות האלה נוצרים באופן דינמי על סמך התוכן הספציפי של השדות האלה.
affectedObjects.id", "affectedObjects.name", "affectedObjects.type principal.resource.attribute.labels מופה מיומן השינויים
affectedObjects.id principal.user.userid מופה מיומן השינויים
affectedObjects.name principal.user.user_display_name מופה מיומן השינויים
changedValues.from", "changedValues.i18nKey", "changedValues.key" and "changedValues.to principal.resource.attribute.labels מופה מיומן השינויים
extraAttributes.name", "extraAttributes.nameI18nKey" and "extraAttributes.value principal.resource.attribute.labels מופה מיומן השינויים
system principal.hostname מופה מיומן השינויים
session_duration additional fields מופה מיומן השינויים

יומן שינויים

הצגת יומן השינויים של המנתח הזה

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