איסוף יומני ביקורת של Tenable

נתמך ב:

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

Tenable Vulnerability Management (לשעבר Tenable.io) היא פלטפורמה לניהול פגיעויות שמבוססת על ענן (cloud.tenable.com). ביומן הפעילות שלה מתועדים אימות משתמשים, גישה ל-API, שינויים בהגדרות ופעולות ניהול. ממשק Tenable Vulnerability Management API בארכיטקטורת REST מספק גישה פרוגרמטית לאירועים ביומן הפעילות.

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

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

  • מופע של Google SecOps
  • פרויקט ב- Google Cloud עם Cloud Storage API מופעל
  • הרשאות ליצירה ולניהול של קטגוריות ב-Cloud Storage
  • הרשאות לניהול כללי מדיניות של ניהול זהויות והרשאות גישה (IAM) בדליים של Cloud Storage
  • הרשאות ליצירת שירותי Cloud Run, נושאי Pub/Sub ומשימות Cloud Scheduler
  • גישה עם הרשאות ל-Tenable Vulnerability Management‏ (cloud.tenable.com) עם התפקיד Administrator
  • מפתחות API של Tenable Vulnerability Management (מפתח גישה ומפתח סודי) שנוצרו בחשבון משתמש עם התפקיד אדמין

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

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

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

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

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

איסוף פרטי הכניסה ל-Tenable Vulnerability Management API

יצירת מפתחות API

  1. נכנסים אל Tenable Vulnerability Management.
  2. בפינה השמאלית העליונה של כל דף, לוחצים על העיגול הכחול של המשתמש ואז על הפרופיל שלי. מופיע הדף החשבון שלי.
  3. עוברים לכרטיסייה מפתחות API.

  4. לוחצים על יצירה. יופיע חלון Generate API Keys עם אזהרה.

    "זהירות: יצירת מפתחות מחליפה את כל מפתחות ה-API הקיימים בחשבון המשתמש הזה, כולל מפתחות שכבר נמצאים בשימוש בשילובים אחרים. צריך להשתמש בחשבון משתמש ייעודי לשילוב הזה, או לעדכן כל אפליקציה שהשתמשה במפתחות הקודמים".

  5. קוראים את האזהרה ולוחצים על יצירה.

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

    • ‫Access Key (מפתח גישה): מפתח הגישה ל-API
    • ‫Secret Key: מפתח ה-API Secret

אימות הגישה ל-API

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

    # Replace with your actual credentials
ACCESS_KEY="your-access-key"
SECRET_KEY="your-secret-key"

curl -s -H "X-ApiKeys: accessKey=${ACCESS_KEY};secretKey=${SECRET_KEY}" \
    "https://cloud.tenable.com/audit-log/v1/events?limit=1" | head -c 500

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

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

  1. נכנסים אל Tenable Vulnerability Management.
  2. בתפריט הניווט הימני, לוחצים על הגדרות.
  3. לוחצים על הלחצן Access Control.
  4. לוחצים על הכרטיסייה משתמשים ואז על חשבון המשתמש שמשמש לשילוב הזה.

  5. מוודאים שלחשבון מוגדר תפקיד אדמין. נקודת הקצה של יומן הפעילות דורשת את תפקיד המשתמש 'אדמין': אף תפקיד מותאם אישית לא מעניק גישה ליומן הפעילות, וכל תפקיד אחר מקבל תגובת HTTP 403.

  6. אם אין לכם את ההרשאות הנדרשות, עליכם לפנות לאדמין של Tenable Vulnerability Management.

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

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

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

  1. במסוף GCP, עוברים אל IAM & Admin > Service Accounts.
  2. לוחצים על יצירת חשבון שירות.
  3. מספקים את פרטי ההגדרה הבאים:
    • שם חשבון השירות: מזינים tenable-audit-collector-sa
    • תיאור חשבון השירות: מזינים Service account for Cloud Run function to collect Tenable Audit 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. לוחצים על שם הקטגוריה (לדוגמה, tenable-audit-logs).
  3. עוברים לכרטיסייה Permissions.
  4. לוחצים על הענקת גישה.
  5. מספקים את פרטי ההגדרה הבאים:
    • Add principals (הוספת ישויות מורשות): מזינים את כתובת האימייל בחשבון של חשבון השירות (לדוגמה, tenable-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • הקצאת תפקידים: בוחרים באפשרות אדמין של אובייקטים באחסון.
  6. לוחצים על Save.

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

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

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

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

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

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

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

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

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

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

  6. בקטע Authentication:

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

  7. גוללים למטה ומרחיבים את Containers, Networking, Security (מאגרי תגים, רשתות, אבטחה).

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

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

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

    1. לוחצים על Variables & Secrets (משתנים וסודות).
    2. לוחצים על + הוספת משתנה לכל משתנה סביבתי:
    שם המשתנה ערך לדוגמה תיאור
    GCS_BUCKET tenable-audit-logs שם הקטגוריה של Cloud Storage
    GCS_PREFIX tenable קידומת לקובצי יומן
    STATE_KEY tenable-state.json נתיב המצב, מחוץ לקידומת היומן
    TENABLE_ACCESS_KEY your-access-key מפתח גישה ל-Tenable Vulnerability Management
    TENABLE_SECRET_KEY your-secret-key מפתח סודי של Tenable Vulnerability Management
    MAX_RECORDS 5000 מספר הרשומות המקסימלי בכל הפעלה
    PAGE_SIZE 1000 רשומות בכל דף
    LOOKBACK_HOURS 24 תקופת המבט לאחור הראשונית
    OVERLAP_MINUTES 2 מספר הדקות לקריאה חוזרת לפני סימן המים כדי לזהות אירועים שאותרו באיחור

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

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

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

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

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

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

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

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

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

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

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

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

    • הקובץ הראשון – main.py:
    import functions_framework
from google.cloud import storage
from google.cloud.exceptions import NotFound
import json
import os
import re
import urllib3
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()

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'tenable')
# 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', 'tenable-state.json')
TENABLE_ACCESS_KEY = os.environ.get('TENABLE_ACCESS_KEY')
TENABLE_SECRET_KEY = os.environ.get('TENABLE_SECRET_KEY')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000'))
# The audit-log API accepts a limit up to 10000.
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
# The query re-reads this many minutes before the watermark so that events
# Tenable indexes late are still collected. Re-read events are dropped by id.
OVERLAP_MINUTES = int(os.environ.get('OVERLAP_MINUTES', '2'))

TENABLE_API_BASE = 'https://cloud.tenable.com'

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

  The watermark must never advance on a failed fetch, otherwise every event in
  the failed window is skipped permanently.
  """

def parse_datetime(value: str) -> datetime:
  """Parse a Tenable `received` timestamp into an aware datetime.

  Tenable returns ISO 8601 with either whole seconds (2018-12-31T23:09:40Z) or
  fractional seconds (2024-01-16T15:12:47.334Z). Fractional digits are trimmed
  to microseconds because fromisoformat accepts at most six.
  """
  text = str(value).strip()
  if text.endswith('Z'):
    text = text[:-1] + '+00:00'
  text = re.sub(r'\.(\d{6})\d+', r'.\1', text)
  dt = datetime.fromisoformat(text)
  if dt.tzinfo is None:
    dt = dt.replace(tzinfo=timezone.utc)
  return dt.astimezone(timezone.utc)

@functions_framework.cloud_event
def main(cloud_event):
  """Fetch Tenable Vulnerability Management activity log events and write them to Cloud Storage.

  Args:
    cloud_event: CloudEvent object containing the Pub/Sub message.
  """
  if not all([GCS_BUCKET, TENABLE_ACCESS_KEY, TENABLE_SECRET_KEY]):
    # 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_ids = set(state.get('seen_ids', []))

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

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

  # A FetchError propagates: the run fails, the watermark is untouched, and the
  # next invocation retries the same window.
  records, newest_event_time = fetch_logs(
    start_time=start_time,
    page_size=PAGE_SIZE,
    max_records=MAX_RECORDS,
  )

  # Drop events already written by an earlier run. Without this the overlap
  # window re-emits its events on every invocation, and an idle tenant has its
  # newest events rewritten to a new object every hour.
  fresh = [r for r in records if str(r.get('id', '')) not in seen_ids]
  print(f"Fetched {len(records)} records, {len(fresh)} new after deduplication")

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

  if not newest_event_time:
    raise FetchError('Records were returned but no `received` timestamp could be parsed')

  timestamp = now.strftime('%Y%m%dT%H%M%SZ')
  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}")

  # 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 ids 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 = []
  for record in records:
    received = record.get('received')
    if not received:
      continue
    try:
      if parse_datetime(received) >= cutoff:
        retained.append(str(record.get('id', '')))
    except ValueError:
      continue

  save_state(bucket, STATE_KEY, {
    'last_event_time': new_watermark.isoformat(),
    'seen_ids': sorted(i for i in set(retained) if i),
  })

  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 fetch_logs(start_time: datetime, page_size: int, max_records: int):
  """Fetch audit events from the Tenable Vulnerability Management API.

  Uses offset pagination, which is what the endpoint documents: the request
  accepts `limit` (maximum 10000) and `offset`, and the response `pagination`
  object returns `offset`, `limit`, `count` and `total`.

  Args:
    start_time: Exclusive lower bound for the `received` timestamp
    page_size: Records per page (the API accepts up to 10000)
    max_records: Maximum total records to fetch in one run

  Returns:
    Tuple of (records list, newest `received` value as an ISO 8601 string).

  Raises:
    FetchError: on any API or transport failure, so the caller cannot mistake
      a failed fetch for an empty result and advance the watermark.
  """
  endpoint = f"{TENABLE_API_BASE}/audit-log/v1/events"

  headers = {
    'X-ApiKeys': f'accessKey={TENABLE_ACCESS_KEY};secretKey={TENABLE_SECRET_KEY}',
    'Accept': 'application/json',
    'User-Agent': 'GoogleSecOps-TenableAuditCollector/1.0'
  }

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

  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

    # Every parameter is rebuilt per page. `sort=received:asc` is required:
    # a time watermark is only correct over an ascending scan, otherwise a
    # truncated run advances the watermark past events it never read.
    params = {
      'f': f'date.gt:{start_time.strftime("%Y-%m-%dT%H:%M:%SZ")}',
      'sort': 'received:asc',
      'limit': min(page_size, max_records - len(records)),
      'offset': offset,
    }
    url = f"{endpoint}?" + '&'.join(f"{k}={v}" for k, v in params.items())

    try:
      response = http.request('GET', url, 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')
      raw_retry_after = response.headers.get('Retry-After')
      try:
        # Retry-After may also be an HTTP date, which int() cannot parse.
        retry_after = int(raw_retry_after) if raw_retry_after else int(backoff)
      except (TypeError, ValueError):
        retry_after = 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
    rate_limit_retries = 0

    if response.status != 200:
      body = response.data.decode('utf-8')
      raise FetchError(f'HTTP {response.status} from the Tenable audit log API: {body}')

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

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

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

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

    for event in page_results:
      received = event.get('received')
      if not received:
        continue
      try:
        if newest_time is None or parse_datetime(received) > parse_datetime(newest_time):
          newest_time = received
      except ValueError as e:
        print(f"Warning: Could not parse event time {received!r}: {e}")

    offset += len(page_results)
    total = data.get('pagination', {}).get('total')
    if total is not None and offset >= total:
      print("No more pages (all matching events retrieved)")
      break

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

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

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

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

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

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

בדיקת השילוב

  1. במסוף Cloud Scheduler, מוצאים את העבודה.
  2. לוחצים על הפעלת המשימה בכוח כדי להפעיל את המשימה באופן ידני.
  3. מחכים כמה שניות.
  4. עוברים אל Cloud Run > Services.
  5. לוחצים על tenable-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
Fetched X records, Y new after deduplication
Wrote Y records to gs://tenable-audit-logs/tenable/logs_YYYYMMDDTHHMMSSZ.ndjson
Saved state: last_event_time=YYYY-MM-DDTHH:MM:SS+00:00
Successfully processed Y records

  8. עוברים אל Cloud Storage > Buckets (קטגוריות).

  9. לוחצים על שם הקטגוריה (tenable-audit-logs).

  10. עוברים לתיקייה tenable/.

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

אם לא מגיעים נתונים חדשים במהלך הרצה, הפונקציה רושמת ביומן את הערך No new log records found. Watermark left unchanged. ולא כותבת אובייקט. זה צפוי: כתיבה מחדש של אותם אירועים תשכפל אותם ב-Google SecOps.

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

  • ‫HTTP 401: בדיקת מפתחות API במשתני סביבה
  • HTTP 403: החשבון לא מוגדר כאדמין. כדי להשתמש בנקודת הקצה של יומן הפעילות, צריך להיות לכם תפקיד משתמש אדמין [64].
  • HTTP 429: הגבלת קצב. הפונקציה מנסה שוב עם השהיה לפני ניסיון חוזר (backoff), ואז הריצה נכשלת בלי להקדים את Watermark, כך שלא מתבצע דילוג על אירועים.
  • חסרים משתני סביבה: צריך לוודא שכל המשתנים הנדרשים מוגדרים

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

טבלת מיפוי UDM

שדה ביומן מיפוי UDM היגיון
crud_label additional.fields ממוזג
fields_label additional.fields ממוזג
field_name extensions.auth.mechanism מיפוי: X-Access-Type → mech_label
mech_label extensions.auth.mechanism ממוזג
extension_value extensions.auth.type מיפוי ישיר
description metadata.description מיפוי ישיר
received metadata.event_timestamp ניתוח הכתובות שברשימה: ISO8601
has_principal metadata.event_type ערכים ממופים (5 סך הכול, למשל true → USER_LOGIN, ‏ true → USER_CREATION, ‏ true → `USER...
has_user metadata.event_type מיפוי: true → USER_UNCATEGORIZED
action metadata.product_event_type מיפוי ישיר
id metadata.product_log_id מיפוי ישיר
field_name principal.asset.ip מיפוי: X-Forwarded-For → ip
ip principal.asset.ip ממוזג
field_name principal.ip מיפוי: X-Forwarded-For → ip
ip principal.ip ממוזג
actor.name principal.user.email_addresses ממוזג
actor.id principal.user.userid מיפוי ישיר
AUTH_VIOLOATION security_result.category ממוזג
is_failure security_result.category מיפוי: true → AUTH_VIOLOATION
is_anonymous_label security_result.detection_fields ממוזג
is_failure_label security_result.detection_fields ממוזג
target1.name target.user.email_addresses ממוזג
target1.type target.user.role_name מיפוי ישיר
target1.id target.user.userid מיפוי ישיר
לא רלוונטי extensions.auth.type קבוע: AUTHTYPE_UNSPECIFIED
לא רלוונטי metadata.event_type קבוע: USER_LOGIN
לא רלוונטי metadata.product_name קבוע: TENABLE AUDIT
לא רלוונטי metadata.vendor_name קבוע: TENABLE

יומן שינויים

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

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