Snipe-IT 記錄
本文說明如何使用 Google Cloud Storage,將 Snipe-IT 記錄檔擷取至 Google Security Operations。Snipe-IT 是一套開放原始碼的 IT 資產管理系統,可追蹤硬體、軟體授權、配件、耗材和其他 IT 資產的整個生命週期。
事前準備
請確認您已完成下列事前準備事項:
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS 值區的權限
- 管理 Google Cloud Storage 值區 IAM 政策的權限
- 建立 Cloud Run 函式、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- Snipe-IT 租戶的特殊存取權
收集 Snipe-IT 必要條件 (API 權杖和基準網址)
- 登入 Snipe-IT。
- 開啟使用者選單 (右上角的大頭貼),然後按一下「管理 API 金鑰」。
- 按一下「Create New Token」:
- 名稱/標籤:輸入描述性標籤 (例如
Google SecOps export)。
- 名稱/標籤:輸入描述性標籤 (例如
- 按一下 [產生]。
- 複製 API 權杖 (只會顯示一次)。妥善儲存。
- 找出 API 基本網址,通常為:
https://<your-domain>/api/v1- 範例:
https://snipeit.example.com/api/v1
- 範例:
建立 Google Cloud Storage 值區
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中,依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
請提供下列設定詳細資料:
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 snipe-it-logs)位置類型 根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用:啟用物件版本管理或保留政策 點選「建立」。
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶。
建立服務帳戶
- 在 GCP 主控台中,依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下 [Create Service Account] (建立服務帳戶)。
- 請提供下列設定詳細資料:
- 服務帳戶名稱:輸入
snipeit-assets-to-gcs-sa。 - 服務帳戶說明:輸入
Service account for Cloud Run function to collect Snipe-IT logs。
- 服務帳戶名稱:輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分:
- 按一下「選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下 [完成]。
這些角色適用於:
- Storage 物件管理員:將記錄檔寫入 GCS 值區,並管理狀態檔案
- Cloud Run 叫用者:允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者:允許函式叫用
授予 GCS 值區的 IAM 權限
授予服務帳戶 GCS bucket 的寫入權限:
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱。
- 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:輸入服務帳戶電子郵件地址 (例如
snipeit-assets-to-gcs-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色:選取「Storage 物件管理員」。
- 新增主體:輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題,Cloud Scheduler 會將訊息發布至該主題,而 Cloud Run 函式會訂閱該主題。
- 在 GCP Console 中,前往「Pub/Sub」>「Topics」(主題)。
- 按一下「建立主題」。
- 請提供下列設定詳細資料:
- 主題 ID:輸入
snipeit-assets-trigger。 - 其他設定保留預設值。
- 主題 ID:輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 Snipe-IT API 擷取硬體資產,並將這些資產寫入 GCS。
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分,提供下列設定詳細資料:
設定 值 服務名稱 snipeit-assets-to-gcs區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取「Python 3.12」以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區:
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中,選擇主題 (
snipeit-assets-trigger)。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分:
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「Containers, Networking, Security」。
前往「安全性」分頁:
- 服務帳戶:選取服務帳戶 (
snipeit-assets-to-gcs-sa)。
- 服務帳戶:選取服務帳戶 (
前往「容器」分頁:
- 按一下「變數與密鑰」。
- 針對每個環境變數,按一下「+ 新增變數」:
變數名稱 範例值 GCS_BUCKETsnipe-it-logsGCS_PREFIXsnipeit/assetsSNIPE_BASE_URLhttps://snipeit.example.com/api/v1SNIPE_API_TOKENyour-api-tokenPAGE_SIZE500MAX_PAGES200在「變數與密鑰」分頁中向下捲動至「要求」:
- 要求逾時:輸入
600秒 (10 分鐘)。
- 要求逾時:輸入
前往「容器」中的「設定」分頁:
- 在「資源」部分:
- 記憶體:選取 512 MiB 以上。
- CPU:選取 1。
- 按一下 [完成]。
- 在「資源」部分:
捲動至「執行環境」:
- 選取「預設」 (建議選項)。
在「修訂版本資源調度」部分:
- 執行個體數量下限:輸入
0。 - 「Maximum number of instances」(執行個體數量上限):輸入
100(或根據預期負載調整)。
- 執行個體數量下限:輸入
點選「建立」。
等待服務建立完成 (1 到 2 分鐘)。
服務建立完成後,系統會自動開啟內嵌程式碼編輯器。
新增函式程式碼
- 在「Function entry point」(函式進入點) 中輸入 main
在內嵌程式碼編輯器中建立兩個檔案:
第一個檔案:main.py:
import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, timezone import urllib.parse # Initialize HTTP client http = urllib3.PoolManager() # Initialize Storage client storage_client = storage.Client() @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch hardware assets from Snipe-IT API and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ # Get environment variables bucket_name = os.environ.get('GCS_BUCKET') prefix = os.environ.get('GCS_PREFIX', 'snipeit/assets') base_url = os.environ.get('SNIPE_BASE_URL', '').rstrip('/') token = os.environ.get('SNIPE_API_TOKEN') page_size = int(os.environ.get('PAGE_SIZE', '500')) max_pages = int(os.environ.get('MAX_PAGES', '200')) if not all([bucket_name, base_url, token]): print('Error: Missing required environment variables') return try: # Get GCS bucket bucket = storage_client.bucket(bucket_name) # Fetch and write pages ts = datetime.now(timezone.utc) offset = 0 page = 0 total = 0 headers = { 'Authorization': f'Bearer {token}', 'Accept': 'application/json', 'Content-Type': 'application/json' } while page < max_pages: # Fetch page from Snipe-IT API params = { 'limit': page_size, 'offset': offset, 'sort': 'id', 'order': 'asc' } qs = urllib.parse.urlencode(params) url = f"{base_url}/hardware?{qs}" print(f'Fetching page {page} from {url}') response = http.request('GET', url, headers=headers, timeout=60.0) if response.status != 200: print(f'Error: API returned status {response.status}') break data = json.loads(response.data.decode('utf-8')) # Get rows from response (Snipe-IT uses 'rows' field) rows = data.get('rows', []) total_available = data.get('total', 0) # Write page to GCS blob_name = f"{prefix}/{ts.strftime('%Y/%m/%d')}/snipeit-hardware-{page:05d}.json" blob = bucket.blob(blob_name) body = json.dumps(data, separators=(',', ':')).encode('utf-8') blob.upload_from_string(body, content_type='application/json') print(f'Wrote page {page} with {len(rows)} rows to {blob_name}') total += len(rows) # Check if we've reached the end # Snipe-IT returns empty rows array when no more data if len(rows) == 0: print(f'No more results (empty rows array)') break # Also check if we've fetched all available records if offset + len(rows) >= total_available: print(f'Reached end of data (offset {offset + len(rows)} >= total {total_available})') break page += 1 offset += page_size print(f'Successfully processed {page + 1} pages with {total} total objects') except Exception as e: print(f'Error processing logs: {str(e)}') raise ```
第二個檔案:requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0
點選「部署」來儲存並部署函式。
等待部署作業完成 (2 到 3 分鐘)。
建立 Cloud Scheduler 工作
Cloud Scheduler 會定期將訊息發布至 Pub/Sub 主題,觸發 Cloud Run 函式。
- 前往 GCP 主控台的「Cloud Scheduler」。
- 點選「建立工作」。
請提供下列設定詳細資料:
設定 值 名稱 snipeit-assets-to-gcs-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取主題 ( snipeit-assets-trigger)郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
根據記錄檔量和延遲時間要求選擇頻率:
頻率 Cron 運算式 用途 每 5 分鐘 */5 * * * *高容量、低延遲 每 15 分鐘檢查一次 */15 * * * *普通量 每小時 0 * * * *標準 (建議採用) 每 6 小時 0 */6 * * *少量、批次處理 每日 0 0 * * *歷來資料集合
測試排程器工作
- 在 Cloud Scheduler 控制台中找出您的工作。
- 按一下「強制執行」即可手動觸發。
- 等待幾秒鐘,然後依序前往「Cloud Run」>「Services」(服務) >「snipeit-assets-to-gcs」>「Logs」(記錄)。
- 確認函式是否已順利執行。
- 檢查 GCS 值區,確認是否已寫入記錄。
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶,從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Snipe-IT logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Snipe-IT」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬的服務帳戶電子郵件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址,以便在下一步中使用。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱。
- 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色:選取「Storage 物件檢視者」。
按一下 [儲存]。
在 Google SecOps 中設定資訊提供,擷取 Snipe-IT 記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Snipe-IT logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Snipe-IT」做為「記錄類型」。
- 點選 [下一步]。
指定下列輸入參數的值:
儲存空間 bucket URL:輸入 GCS bucket URI,並加上前置路徑:
gs://snipe-it-logs/snipeit/assets/取代:
snipe-it-logs:您的 GCS bucket 名稱。snipeit/assets:儲存記錄的前置字元/資料夾路徑。
來源刪除選項:根據偏好設定選取刪除選項:
- 永不:移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
刪除已轉移的檔案和空白目錄:成功轉移後刪除檔案和空白目錄。
檔案存在時間上限:包含在過去天數內修改的檔案。預設值為 180 天。
資產命名空間:資產命名空間。
擷取標籤:要套用至這個動態饋給事件的標籤。
點選 [下一步]。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定,然後按一下「Submit」(提交)。
需要其他協助嗎?向社群成員和 Google SecOps 專業人員尋求答案。