איסוף יומנים של מכשירי Armis
גרסת המנתח: 2.0
במאמר הזה מוסבר איך להטמיע קובצי יומן במכשיר של Armis ב-Google Security Operations באמצעות Google Cloud Storage V2.
Armis Centrix היא פלטפורמה ברמה ארגונית לניהול נכסים, שנועדה לספק הרשאות גישה והגנה למכשירים מנוהלים, לא מנוהלים ו-IoT/OT. זרם הנתונים של Armis Devices מתעד את כל מלאי המכשירים שנצפה ברשתות המנוטרות, כולל פרטי חומרה, מאפייני רשת, הערכות סיכונים ומטא-נתונים של סיווג. אפשר לייצא את יומני מלאי המכשירים האלה דרך ה-API בארכיטקטורת REST של Armis, ולכתוב אותם לקטגוריה של GCS כדי ש-Google SecOps יטמיע אותם.
לפני שמתחילים
חשוב לוודא שיש לכם את הדברים הבאים:
- מכונה של Google SecOps
- פרויקט ב-GCP שבו מופעל Cloud Storage API
- הרשאות ליצירה ולניהול של קטגוריות ב-GCS
- הרשאות ליצירת שירותי Cloud Run, נושאי Pub/Sub ומשימות ב-Cloud Scheduler
- מכונה של Armis Centrix עם גישה ל-API
- מפתח סודי של Armis API עם הרשאות לקריאת נתוני מכשירים
יצירת מפתח API סודי של Armis
- נכנסים למסוף Armis Centrix בכתובת
https://<your-instance>.armis.com. - עוברים אל הגדרות > ניהול נכסים ואבטחה > ניהול ממשקי API.
- לוחצים על Create (יצירה) כדי ליצור מפתח API סודי חדש.
- תיפתח תיבת דו-שיח עם המפתח הסודי.
- מעתיקים ושומרים את המפתח הסודי באופן מאובטח.
אימות הגישה ל-API
לפני שממשיכים בתהליך השילוב, בודקים את פרטי הכניסה:
ARMIS_INSTANCE="https://your-instance.armis.com" ARMIS_SECRET_KEY="your-secret-key" # Get access token ACCESS_TOKEN=$(curl -s -X POST "${ARMIS_INSTANCE}/api/v1/access_token/" \ -d "secret_key=${ARMIS_SECRET_KEY}" \ | python3 -c "import sys,json; print(json.load(sys.stdin)['data']['access_token'])") echo "Access Token: ${ACCESS_TOKEN}" # Test device search curl -s "${ARMIS_INSTANCE}/api/v1/search/" \ -H "Authorization: ${ACCESS_TOKEN}" \ -G \ --data-urlencode "aql=in:devices timeFrame:\"1 Hours\"" \ --data-urlencode "length=1" \ --data-urlencode "fields=id,name,type,category,model,manufacturer,ipAddress,macAddress,operatingSystem,riskLevel,site,firstSeen,lastSeen" \ | python3 -m json.tool
יצירת קטגוריה של Google Cloud Storage
- נכנסים למסוף Google Cloud.
- בוחרים את הפרויקט או יוצרים פרויקט חדש.
- בתפריט הניווט, נכנסים ל-Cloud Storage > Buckets.
- לוחצים על Create bucket.
ממלאים את פרטי ההגדרה הבאים:
הגדרה ערך Name your bucket מזינים שם ייחודי ברמה גלובלית (לדוגמה, armis-device-logs)סוג המיקום בוחרים לפי הצרכים (אזור, שני אזורים, מספר אזורים) Location בוחרים את המיקום (לדוגמה, us-central1)סוג אחסון (storage class) רגיל (מומלץ ליומנים שהגישה אליהם תכופה) בקרת גישה אחידה (מומלץ) כלי הגנה אופציונלי: הפעלת ניהול גרסאות של אובייקטים או מדיניות שמירת נתונים לוחצים על יצירה.
יצירת חשבון שירות לפונקציית Cloud Run
לפונקציית Cloud Run נדרש חשבון שירות עם הרשאות כתיבה בקטגוריית GCS, וצריך להפעיל אותה באמצעות Pub/Sub.
יצירת חשבון השירות
- במסוף GCP, נכנסים לדף IAM & Admin > Service Accounts.
- לוחצים על Create Service Account.
- מזינים את פרטי ההגדרה הבאים:
- Service account name: מזינים
armis-devices-collector-sa - Service account description: מזינים
Service account for Cloud Run function to collect Armis device inventory logs
- Service account name: מזינים
- לוחצים על יצירה והמשך.
- בקטע Grant this service account access to project, מוסיפים את התפקידים הבאים:
- לוחצים על בחירת תפקיד.
- מחפשים את התפקיד Storage Object Admin ובוחרים אותו.
- לוחצים על + הוספת תפקיד.
- מחפשים את התפקיד Cloud Run Invoker ובוחרים בו.
- לוחצים על + הוספת תפקיד.
- מחפשים את התפקיד Cloud Functions Invoker ובוחרים אותו.
- לוחצים על Continue.
- לוחצים על סיום.
התפקידים האלה נדרשים כדי:
- אדמין של אובייקט אחסון: כתיבת יומנים לקטגוריית GCS וניהול קובצי תמונת מצב
- Cloud Run Invoker: מאפשר ל-Pub/Sub להפעיל את הפונקציה
- מפעיל פונקציות של Cloud Functions: מאפשר להפעיל פונקציות
הענקת הרשאות IAM בקטגוריה של GCS
נותנים לחשבון השירות הרשאות כתיבה בקטגוריה של GCS:
- נכנסים לדף Cloud Storage > Buckets.
- לוחצים על שם הקטגוריה.
- עוברים לכרטיסייה Permissions.
- לוחצים על הענקת גישה.
- מזינים את פרטי ההגדרה הבאים:
- הוספת חשבונות משתמשים: מזינים את כתובת האימייל של חשבון השירות (לדוגמה,
armis-devices-collector-sa@PROJECT_ID.iam.gserviceaccount.com) - הקצאת תפקידים: בוחרים באפשרות אדמין של אובייקטים באחסון.
- הוספת חשבונות משתמשים: מזינים את כתובת האימייל של חשבון השירות (לדוגמה,
- לוחצים על Save.
יוצרים נושא Pub/Sub
יוצרים נושא Pub/Sub ש-Cloud Scheduler יפרסם בו ופונקציית Cloud Run תירשם אליו.
- במסוף GCP, נכנסים לדף Pub/Sub > Topics.
- לוחצים על Create topic.
- מזינים את פרטי ההגדרה הבאים:
- מזהה הנושא: מזינים
armis-devices-trigger - משאירים את שאר ההגדרות כברירת מחדל.
- מזהה הנושא: מזינים
- לוחצים על יצירה.
יצירת פונקציית Cloud Run לאיסוף יומנים
הודעות Pub/Sub מ-Cloud Scheduler יפעילו את פונקציית Cloud Run כדי לאחזר רשומות של מלאי המכשירים מ-Armis API בארכיטקטורת REST ולכתוב אותן ב-GCS.
- במסוף GCP, נכנסים ל-Cloud Run.
- לוחצים על Create service.
- בוחרים באפשרות Function (שימוש בעורך ישיר כדי ליצור פונקציה).
בקטע Configure, מזינים את פרטי ההגדרה הבאים:
הגדרה ערך שם השירות armis-devices-collectorאזור בוחרים את האזור שתואם לקטגוריית ה-GCS (לדוגמה, us-central1)זמן ריצה בוחרים באפשרות Python 3.12 או גרסה עדכנית יותר. בקטע טריגר (אופציונלי):
- לוחצים על + הוספת טריגר.
- בוחרים באפשרות Cloud Pub/Sub.
- בקטע Select a Cloud Pub/Sub topic, בוחרים את הנושא
armis-devices-trigger. - לוחצים על Save.
בקטע אימות:
- לוחצים על דרישת אימות.
- בודקים את ניהול הזהויות והרשאות הגישה (IAM).
גוללים למטה ומרחיבים את הקטע Containers, Networking, Security.
עוברים לכרטיסייה אבטחה:
- חשבון שירות: בוחרים את חשבון השירות
armis-devices-collector-sa
- חשבון שירות: בוחרים את חשבון השירות
עוברים לכרטיסייה מאגרי תגים:
- לוחצים על Variables & Secrets.
- לוחצים על + Add variable לכל משתנה סביבה:
שם משתנה ערך לדוגמה תיאור GCS_BUCKETarmis-device-logsשם הקטגוריה ב-GCS GCS_PREFIXarmis-devicesקידומת לקובצי יומן STATE_KEYarmis-devices/state.jsonנתיב קובץ תמונת המצב ARMIS_INSTANCE_URLhttps://your-instance.armis.comכתובת ה-URL של המכונה של Armis ARMIS_SECRET_KEYyour-secret-keyמפתח API סודי של Armis PAGE_SIZE1000רשומות בכל דף API MAX_RECORDS50000מספר הרשומות המקסימלי לכל הרצה LOOKBACK_HOURS24תקופת המבט לאחור הראשונית גוללים למטה בקטע Variables & Secrets עד שמגיעים ל-Requests:
- Request timeout: מזינים
540שניות (9 דקות)
- Request timeout: מזינים
עוברים לכרטיסייה הגדרות:
- בקטע Resources:
- זיכרון: בוחרים באפשרות 512 MiB או יותר
- מעבד (CPU): בוחרים באפשרות 1
- בקטע Resources:
בקטע Revision scaling:
- Minimum number of instances: מזינים
0 - Maximum number of instances: מזינים
100(או משנים בהתאם לעומס הצפוי)
- Minimum number of instances: מזינים
לוחצים על יצירה.
ממתינים ליצירת השירות (דקה או שתיים).
אחרי שיוצרים את השירות, עורך הקוד המוטבע נפתח באופן אוטומטי.
הוספת קוד פונקציה
- בשדה Entry point, מזינים main.
בעורך הקוד, יוצרים שני קבצים:
הקובץ הראשון – main.py:
import functions_framework from google.cloud import storage import json import os import urllib3 import urllib.parse from datetime import datetime, timezone, timedelta import time # Initialize HTTP client with timeouts http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=5.0, read=60.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', 'armis-devices').strip('/') STATE_KEY = os.environ.get('STATE_KEY') or f"{GCS_PREFIX}/state.json" ARMIS_INSTANCE_URL = os.environ.get('ARMIS_INSTANCE_URL', '').rstrip('/') ARMIS_SECRET_KEY = os.environ.get('ARMIS_SECRET_KEY') PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000')) MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '50000')) LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) DEVICE_FIELDS = [ 'id', 'name', 'type', 'category', 'model', 'manufacturer', 'ipAddress', 'macAddress', 'operatingSystem', 'operatingSystemVersion', 'riskLevel', 'site', 'tags', 'firstSeen', 'lastSeen', 'boundaries', 'customProperties', 'dataSources', 'userIds', 'visibility', 'ppiSensitive' ] def parse_datetime(value): """Parse ISO datetime string to datetime object.""" if not value: return None if value.endswith("Z"): value = value[:-1] + "+00:00" try: return datetime.fromisoformat(value) except Exception: try: return datetime.strptime(value, '%Y-%m-%dT%H:%M:%S') except Exception: return None def get_access_token(): """Obtain an access token from the Armis API.""" url = f"{ARMIS_INSTANCE_URL}/api/v1/access_token/" body = urllib.parse.urlencode({'secret_key': ARMIS_SECRET_KEY}) response = http.request( 'POST', url, body=body, headers={'Content-Type': 'application/x-www-form-urlencoded'} ) if response.status != 200: raise Exception( f"Failed to get access token: HTTP {response.status} - " f"{response.data.decode('utf-8')}" ) data = json.loads(response.data.decode('utf-8')) return data['data']['access_token'] @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch Armis Device inventory records and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, ARMIS_INSTANCE_URL, ARMIS_SECRET_KEY]): print('Error: Missing required environment variables') return try: 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 delayed events if last_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 device logs from " f"{last_time.isoformat()} to {now.isoformat()}" ) # Get access token access_token = get_access_token() print("Successfully obtained Armis access token.") # Fetch device records records, newest_event_time = fetch_device_logs( access_token=access_token, start_time=last_time, end_time=now, ) if not records: print("No new device 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 " f"gs://{GCS_BUCKET}/{object_key}" ) # Update state 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 device 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): """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_device_logs(access_token, start_time, end_time): """ Fetch device inventory logs from Armis REST API with pagination. Args: access_token: Armis API access token start_time: Start time for log query end_time: End time for log query Returns: Tuple of (records list, newest_event_time ISO string) """ base_url = f"{ARMIS_INSTANCE_URL}/api/v1/search/" headers = {'Authorization': access_token} # Build AQL query with time filter start_str = start_time.strftime('%Y-%m-%dT%H:%M:%S') aql_query = f'in:devices after:{start_str}' records = [] newest_time = None page_from = 0 backoff = 1.0 while True: if len(records) >= MAX_RECORDS: print(f"Reached max_records limit ({MAX_RECORDS})") break params = { 'aql': aql_query, 'from': str(page_from), 'orderBy': 'lastSeen', 'length': str(PAGE_SIZE), 'fields': ','.join(DEVICE_FIELDS), } try: response = http.request( 'GET', base_url, fields=params, 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). " f"Retrying after {retry_after}s..." ) time.sleep(retry_after) backoff = min(backoff * 2, 30.0) continue # Handle token expiry if response.status == 401: print("Access token expired, refreshing...") access_token = get_access_token() headers = {'Authorization': access_token} 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 records, newest_time data = json.loads(response.data.decode('utf-8')) result = data.get('data', {}).get('results', []) total = data.get('data', {}).get('total', 0) count = data.get('data', {}).get('count', 0) next_from = data.get('data', {}).get('next') if not result: print(f"No more results (empty page)") break print( f"Page from={page_from}: Retrieved {count} devices " f"(total: {total})" ) records.extend(result) # Track newest event time using lastSeen for device in result: try: event_time = device.get('lastSeen') if event_time: if newest_time is None: newest_time = event_time else: current_dt = parse_datetime(event_time) newest_dt = parse_datetime(newest_time) if ( current_dt and newest_dt and current_dt > newest_dt ): newest_time = event_time except Exception as e: print( f"Warning: Could not parse " f"event time: {e}" ) # Check pagination if next_from is None or count < PAGE_SIZE: print("Reached last page") break page_from = next_from except Exception as e: print(f"Error fetching device logs: {e}") return records, newest_time print(f"Retrieved {len(records)} total device records") return records, newest_timeהקובץ השני – requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0
לוחצים על Deploy כדי לשמור ולפרוס את הפונקציה.
מחכים שהפריסה תסתיים (2-3 דקות).
יצירת משימה ב-Cloud Scheduler
Cloud Scheduler יפרסם הודעות בנושא Pub/Sub במרווחי זמן קבועים, ויפעיל את פונקציית Cloud Run.
- במסוף GCP, נכנסים ל-Cloud Scheduler.
- לוחצים על Create Job.
ממלאים את פרטי ההגדרה הבאים:
הגדרה ערך שם armis-devices-collector-dailyאזור בוחרים את אותו אזור כמו של פונקציית Cloud Run. תדירות 0 2 * * *(כל יום בשעה 02:00)אזור זמן בחירת אזור זמן (מומלץ לבחור ב-UTC) סוג היעד Pub/Sub נושא בוחרים את הנושא armis-devices-triggerגוף ההודעה {}(אובייקט JSON ריק)לוחצים על יצירה.
אפשרויות לתדירות התזמון
בוחרים את התדירות על סמך קצב העדכון של מלאי המכשירים ודרישות זמן האחזור:
| תדירות | ביטוי Cron | תרחיש לדוגמה |
|---|---|---|
| כל 6 שעות | 0 */6 * * * |
עדכוני מלאי תכופים |
| יומי | 0 2 * * * |
רגיל (מומלץ) |
| Every 12 hours | 0 */12 * * * |
סביבות עם עוצמת קול נמוכה יותר |
בדיקת השילוב
- במסוף Cloud Scheduler, מוצאים את המשימה.
- לוחצים על Force run כדי להפעיל את המשימה באופן ידני.
- מחכים כמה שניות.
- נכנסים לדף Cloud Run > Services.
- לוחצים על שם הפונקציה
armis-devices-collector. - לוחצים על הכרטיסייה Logs.
מוודאים שהפונקציה בוצעה בהצלחה. חפשו את:
Fetching device logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Successfully obtained Armis access token. Page from=0: Retrieved X devices (total: Y) Wrote X records to gs://armis-device-logs/armis-devices/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X recordsנכנסים לדף Cloud Storage > Buckets.
לוחצים על שם הקטגוריה.
עוברים לתיקיית הקידומת
armis-devices/.מוודאים שנוצר קובץ
.ndjsonחדש עם חותמת הזמן הנוכחית.
אם מופיעות שגיאות ביומנים:
- HTTP 400: מוודאים שהתחביר של שאילתת ה-AQL נכון. מוודאים שבסוף כתובת ה-URL של המכונה של Armis אין לוכסן.
- HTTP 401: פג התוקף של טוקן הגישה או שהוא לא תקין. הפונקציה מרעננת את האסימון באופן אוטומטי בשגיאות 401.
- HTTP 429: הגבלת קצב של יצירת בקשות – הפונקציה תנסה שוב באופן אוטומטי עם השהיה.
- "Access token not found": ודאו שהמפתח הסודי של Armis API נכון.
- משתני סביבה חסרים: ודאו שכל המשתנים הנדרשים מוגדרים.
הגדרת פיד ב-Google SecOps כדי להטמיע יומנים של מכשירי Armis
- עוברים אל הגדרות > ניהול נתונים > פידים. אם אתם משתמשים בפלטפורמה של SIEM בלבד, עוברים אל הגדרות > פידים.
- לוחצים על הוספת פיד חדש.
- לוחצים על הגדרת פיד יחיד.
- בשדה שם הפיד, מזינים שם לפיד (לדוגמה,
Armis Devices). - בוחרים באפשרות Google Cloud Storage V2 בתור סוג המקור.
- בוחרים באפשרות Armis Devices בתור Log type.
לוחצים על Get Service Account (קבלת חשבון שירות). תוצג כתובת אימייל ייחודית של חשבון שירות, לדוגמה:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comמעתיקים את כתובת האימייל הזו כדי להשתמש בה בשלב הבא.
לוחצים על הבא.
מציינים ערכים לפרמטרים הבאים של הקלט:
כתובת ה-URL של קטגוריית האחסון: מזינים את ה-URI של קטגוריית GCS עם נתיב הקידומת:
gs://armis-device-logs/armis-devices/- מחליפים את:
-
armis-device-logs: השם של קטגוריית GCS. -
armis-devices: נתיב אופציונלי לתיקייה או לתחילית שבהם נשמרים היומנים (משאירים ריק בשביל תיקיית הבסיס).
-
- מחליפים את:
אפשרות מחיקת המקור: בוחרים את אפשרות המחיקה המועדפת:
- Never: אף פעם לא למחוק קבצים אחרי העברות (מומלץ לבדיקה).
- מחיקת קבצים שהועברו: מחיקת קבצים אחרי שהם הועברו בהצלחה.
- מחיקת קבצים שהועברו וספריות ריקות: מחיקת קבצים וספריות ריקות אחרי שההעברה מסתיימת.
גיל הקובץ המקסימלי: כולל קבצים ששונו במספר הימים האחרונים (ברירת המחדל היא 180 ימים)
מרחב השמות של הנכס: מרחב השמות של הנכס
תוויות הטמעה: התווית שתתווסף לאירועים מהפיד הזה
לוחצים על הבא.
בודקים את הגדרות הפיד החדש במסך סיום ולוחצים על שליחה.
מתן הרשאות IAM לחשבון השירות של Google SecOps
לחשבון השירות של Google SecOps צריך להיות התפקיד צפייה באובייקט אחסון בקטגוריה של GCS.
- נכנסים לדף Cloud Storage > Buckets.
- לוחצים על שם הקטגוריה.
- עוברים לכרטיסייה Permissions.
- לוחצים על הענקת גישה.
- מזינים את פרטי ההגדרה הבאים:
- Add principals: מדביקים את כתובת האימייל של חשבון השירות של Google SecOps.
- Assign roles: בוחרים באפשרות Storage Object Viewer.
לוחצים על Save.
טבלת מיפוי UDM
| שדה ביומן | מיפוי UDM | Logic (פונקציה לוגית) |
|---|---|---|
datasource.firstSeen |
metadata.event_timestamp |
נותח בתור yyyy-MM-ddTHH:mm:ssZZ |
datasource.lastSeen |
metadata.event_timestamp |
נותח בתור yyyy-MM-ddTHH:mm:ssZZ |
firstSeen |
metadata.event_timestamp |
נותח בתור yyyy-MM-ddTHH:mm:ssZZ |
lastSeen |
metadata.event_timestamp |
נותח בתור yyyy-MM-ddTHH:mm:ssZZ |
time |
metadata.event_timestamp |
נותח בתור yyyy-MM-ddTHH:mm:ss.SSSZZ |
id |
metadata.product_entity_id |
מיפוי ישיר |
relation |
relations |
שינוי שם/מיפוי |
| לא רלוונטי | metadata.entity_type |
קבוע: ASSET |
| לא רלוונטי | metadata.product_name |
קבוע: ARMIS |
| לא רלוונטי | metadata.vendor_name |
קבוע: ARMIS |
יומן שינויים
הצגת יומן השינויים של המנתח הזה
הבעיה עדיין לא נפתרה? קבלת תשובות מחברי הקהילה וממומחי Google SecOps.