收集 Jamf Pro 環境記錄

支援的國家/地區：

Google SecOps SIEM

本文說明如何使用 Google Cloud Storage，搭配 Cloud Run 函式、Pub/Sub 和 Cloud Scheduler，將 Jamf Pro 環境記錄 (裝置和使用者環境) 擷取至 Google Security Operations。Jamf Pro 是 Apple 裝置的完整管理解決方案，提供裝置清單、使用者環境和設定管理功能。

事前準備

請確認您已完成下列事前準備事項：

Google SecOps 執行個體
已啟用 Cloud Storage API 的 GCP 專案
建立及管理 GCS 值區的權限
管理 Google Cloud Storage 值區 IAM 政策的權限
建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
Jamf Pro 租戶的特殊存取權

設定 Jamf API 角色

登入 Jamf 網頁版使用者介面。
依序前往「設定」>「系統」部分 >「API 角色和用戶端」。
選取「API 角色」分頁標籤。
按一下 [新增]。
輸入 API 角色顯示名稱 (例如 context_role)。
在「Jamf Pro API role privileges」(Jamf Pro API 角色權限) 中輸入權限名稱，然後從選單中選取：
- 電腦清單
- 行動裝置清單
注意： 您可以繼續新增更多權限，直到完成為止。
按一下 [儲存]。

設定 Jamf API 用戶端

在 Jamf Pro 中，依序前往「設定」>「系統」部分 >「API 角色和用戶端」。
選取「API 用戶端」分頁標籤。
按一下 [新增]。
輸入 API 用戶端的顯示名稱 (例如 context_client)。
在「API Roles」欄位中，新增您先前建立的 context_role 角色。
在「存取權杖有效時間」下方，輸入存取權杖的有效時間 (以秒為單位)。
按一下 [儲存]。
按一下 [編輯]。
按一下「啟用 API 用戶端」。
按一下 [儲存]。

設定 Jamf 用戶端密鑰

在 Jamf Pro 中，前往新建立的 API 用戶端。
按一下「產生用戶端密鑰」。
在確認畫面中，按一下「建立密鑰」。
將下列參數儲存在安全的位置：
- 基本網址：https://<your>.jamfcloud.com
- 用戶端 ID：UUID。
- 用戶端密鑰：這個值只會顯示一次。

建立 Google Cloud Storage 值區

前往 Google Cloud 控制台。
選取專案或建立新專案。
在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
按一下「建立值區」。

請提供下列設定詳細資料：

設定	值
為 bucket 命名	輸入全域不重複的名稱 (例如 `jamfpro`)
位置類型	根據需求選擇 (區域、雙區域、多區域)
位置	選取位置 (例如 `us-central1`)
儲存空間級別	標準 (建議用於經常存取的記錄)
存取控管	統一 (建議)
保護工具	選用：啟用物件版本管理或保留政策

點選「建立」。

為 Cloud Run 函式建立服務帳戶

Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶，並由 Pub/Sub 叫用。

建立服務帳戶

在 GCP 主控台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
按一下 [Create Service Account] (建立服務帳戶)。
請提供下列設定詳細資料：
- 服務帳戶名稱：輸入 jamf-pro-collector-sa。
- 服務帳戶說明：輸入 Service account for Cloud Run function to collect Jamf Pro context logs。
按一下「建立並繼續」。
在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
1. 按一下「選擇角色」。
2. 搜尋並選取「Storage 物件管理員」。
3. 點選「+ 新增其他角色」。
4. 搜尋並選取「Cloud Run Invoker」。
5. 點選「+ 新增其他角色」。
6. 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
按一下「繼續」。
按一下 [完成]。

這些角色適用於：

Storage 物件管理員：將記錄檔寫入 GCS 值區，並管理狀態檔案
Cloud Run 叫用者：允許 Pub/Sub 叫用函式
Cloud Functions 叫用者：允許函式叫用

授予 GCS 值區的 IAM 權限

授予服務帳戶 GCS bucket 的寫入權限：

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱。
前往「權限」分頁標籤。
按一下「授予存取權」。
請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如 jamf-pro-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。
- 指派角色：選取「Storage 物件管理員」。
按一下 [儲存]。

建立 Pub/Sub 主題

建立 Pub/Sub 主題，Cloud Scheduler 會將訊息發布至該主題，而 Cloud Run 函式會訂閱該主題。

在 GCP Console 中，前往「Pub/Sub」>「Topics」(主題)。
按一下「建立主題」。
請提供下列設定詳細資料：
- 主題 ID：輸入 jamf-pro-context-trigger。
- 其他設定保留預設值。
點選「建立」。

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 Jamf Pro API 擷取記錄，並將記錄寫入 GCS。

前往 GCP Console 的「Cloud Run」。
按一下「Create service」(建立服務)。
選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分，提供下列設定詳細資料：

設定值

服務名稱 jamf-pro-context-collector

區域選取與 GCS bucket 相符的區域 (例如 us-central1)

執行階段 選取「Python 3.12」以上版本
在「Trigger (optional)」(觸發條件 (選用)) 專區：
1. 按一下「+ 新增觸發條件」。
2. 選取「Cloud Pub/Sub」。
3. 在「選取 Cloud Pub/Sub 主題」中，選擇 Pub/Sub 主題 (jamf-pro-context-trigger)。
4. 按一下 [儲存]。
在「Authentication」(驗證) 部分：
1. 選取「需要驗證」。
2. 檢查 Identity and Access Management (IAM)。
注意： 叫用函式時，Pub/Sub 會自動處理驗證。
向下捲動並展開「Containers, Networking, Security」。
前往「安全性」分頁：
- 服務帳戶：選取服務帳戶 (jamf-pro-collector-sa)。

設定	值
服務名稱	`jamf-pro-context-collector`
區域	選取與 GCS bucket 相符的區域 (例如 `us-central1`)
執行階段	選取「Python 3.12」以上版本

前往「容器」分頁：

按一下「變數與密鑰」。
針對每個環境變數，按一下「+ 新增變數」：

變數名稱	範例值
`GCS_BUCKET`	`jamfpro`
`GCS_PREFIX`	`jamf-pro/context/`
`JAMF_CLIENT_ID`	輸入 Jamf 用戶端 ID
`JAMF_CLIENT_SECRET`	輸入 Jamf 用戶端密鑰
`JAMF_BASE_URL`	輸入 Jamf 網址，並將 `https://<your>.jamfcloud.com` 中的 `<your>`
`PAGE_SIZE`	`200`

在「變數與密鑰」分頁中向下捲動至「要求」：
- 要求逾時：輸入 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 gzip
import io
import time

# Initialize HTTP client
http = urllib3.PoolManager()

# Initialize Storage client
storage_client = storage.Client()

# Configuration
BASE_URL = os.environ.get("JAMF_BASE_URL", "").rstrip("/")
CLIENT_ID = os.environ.get("JAMF_CLIENT_ID")
CLIENT_SECRET = os.environ.get("JAMF_CLIENT_SECRET")
GCS_BUCKET = os.environ.get("GCS_BUCKET")
GCS_PREFIX = os.environ.get("GCS_PREFIX", "jamf-pro/context/")
PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "200"))

SECTIONS = [
    "GENERAL", "HARDWARE", "OPERATING_SYSTEM", "USER_AND_LOCATION",
    "DISK_ENCRYPTION", "SECURITY", "EXTENSION_ATTRIBUTES", "APPLICATIONS",
    "CONFIGURATION_PROFILES", "LOCAL_USER_ACCOUNTS", "CERTIFICATES",
    "SERVICES", "PRINTERS", "SOFTWARE_UPDATES", "GROUP_MEMBERSHIPS",
    "CONTENT_CACHING", "STORAGE", "FONTS", "PACKAGE_RECEIPTS",
    "PLUGINS", "ATTACHMENTS", "LICENSED_SOFTWARE", "IBEACONS", "PURCHASING"
]

def _now_iso():
    return datetime.now(timezone.utc).isoformat()

def get_token():
    """OAuth2 client credentials > access_token"""
    url = f"{BASE_URL}/api/oauth/token"

    # Encode credentials for form data
    fields = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }

    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }

    response = http.request(
        'POST',
        url,
        fields=fields,
        headers=headers,
        timeout=30.0
    )

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

    data = json.loads(response.data.decode('utf-8'))
    return data["access_token"], int(data.get("expires_in", 1200))

def fetch_page(token, page):
    """GET /api/v1/computers-inventory with sections & pagination"""
    url = f"{BASE_URL}/api/v1/computers-inventory"

    # Build query parameters
    params = [("page", str(page)), ("page-size", str(PAGE_SIZE))]
    params.extend([("section", s) for s in SECTIONS])

    # Encode parameters
    query_string = "&".join([f"{k}={v}" for k, v in params])
    full_url = f"{url}?{query_string}"

    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json"
    }

    response = http.request(
        'GET',
        full_url,
        headers=headers,
        timeout=60.0
    )

    if response.status != 200:
        raise Exception(f"Failed to fetch page {page}: {response.status} {response.data.decode('utf-8')}")

    return json.loads(response.data.decode('utf-8'))

def to_context_event(item):
    inv = item.get("inventory", {}) or {}
    general = inv.get("general", {}) or {}
    hardware = inv.get("hardware", {}) or {}
    osinfo = inv.get("operatingSystem", {}) or {}
    loc = inv.get("location", {}) or inv.get("userAndLocation", {}) or {}

    computer = {
        "udid": general.get("udid") or hardware.get("udid"),
        "deviceName": general.get("name") or general.get("deviceName"),
        "serialNumber": hardware.get("serialNumber") or general.get("serialNumber"),
        "model": hardware.get("model") or general.get("model"),
        "osVersion": osinfo.get("version") or general.get("osVersion"),
        "osBuild": osinfo.get("build") or general.get("osBuild"),
        "macAddress": hardware.get("macAddress"),
        "alternateMacAddress": hardware.get("wifiMacAddress"),
        "ipAddress": general.get("ipAddress"),
        "reportedIpV4Address": general.get("reportedIpV4Address"),
        "reportedIpV6Address": general.get("reportedIpV6Address"),
        "modelIdentifier": hardware.get("modelIdentifier"),
        "assetTag": general.get("assetTag"),
    }

    user_block = {
        "userDirectoryID": loc.get("username") or loc.get("userDirectoryId"),
        "emailAddress": loc.get("emailAddress"),
        "realName": loc.get("realName"),
        "phone": loc.get("phone") or loc.get("phoneNumber"),
        "position": loc.get("position"),
        "department": loc.get("department"),
        "building": loc.get("building"),
        "room": loc.get("room"),
    }

    return {
        "webhook": {"name": "api.inventory"},
        "event_type": "ComputerInventory",
        "event_action": "snapshot",
        "event_timestamp": _now_iso(),
        "event_data": {
            "computer": {k: v for k, v in computer.items() if v not in (None, "")},
            **{k: v for k, v in user_block.items() if v not in (None, "")}
        },
        "_jamf": {
            "id": item.get("id"),
            "inventory": inv
        }
    }

def write_ndjson_gz(objs, when):
    buf = io.BytesIO()
    with gzip.GzipFile(filename="-", mode="wb", fileobj=buf, mtime=int(time.time())) as gz:
        for obj in objs:
            line = json.dumps(obj, separators=(",", ":")) + "\n"
            gz.write(line.encode("utf-8"))

    buf.seek(0)

    prefix = GCS_PREFIX.strip("/") + "/" if GCS_PREFIX else ""
    key = f"{prefix}{when:%Y/%m/%d}/jamf_pro_context_{int(when.timestamp())}.ndjson.gz"

    bucket = storage_client.bucket(GCS_BUCKET)
    blob = bucket.blob(key)
    blob.upload_from_file(buf, content_type="application/gzip")

    return key

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

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

    if not all([BASE_URL, CLIENT_ID, CLIENT_SECRET, GCS_BUCKET]):
        print("Error: Missing required environment variables")
        return

    try:
        token, _ttl = get_token()

        page = 0
        total = 0
        batch = []
        now = datetime.now(timezone.utc)

        while True:
            payload = fetch_page(token, page)
            results = payload.get("results") or []

            if not results:
                break

            for item in results:
                batch.append(to_context_event(item))
                total += 1

            if len(batch) >= 5000:
                key = write_ndjson_gz(batch, now)
                print(f"Wrote {len(batch)} records to gs://{GCS_BUCKET}/{key}")
                batch = []

            if len(results) < PAGE_SIZE:
                break

            page += 1

        if batch:
            key = write_ndjson_gz(batch, now)
            print(f"Wrote {len(batch)} records to gs://{GCS_BUCKET}/{key}")

        print(f"Successfully processed {total} total records")

    except Exception as e:
        print(f"Error processing Jamf Pro context 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」。
點選「建立工作」。

請提供下列設定詳細資料：

設定	值
名稱	`jamfpro-context-schedule-1h`
區域	選取與 Cloud Run 函式相同的區域
頻率	`0 * * * *` (每小時整點)
時區	選取時區 (建議使用世界標準時間)
目標類型	Pub/Sub
主題	選取 Pub/Sub 主題 (`jamf-pro-context-trigger`)
郵件內文	`{}` (空白 JSON 物件)

點選「建立」。

測試排程器工作

在 Cloud Scheduler 控制台中找出您的工作。
按一下「強制執行」即可手動觸發。
等待幾秒鐘，然後依序前往「Cloud Run」>「Services」(服務) >「jamf-pro-context-collector」>「Logs」(記錄)。
確認函式是否已順利執行。
檢查 GCS 值區，確認是否已寫入記錄。

擷取 Google SecOps 服務帳戶

Google SecOps 會使用專屬服務帳戶，從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。

取得服務帳戶電子郵件地址

依序前往「SIEM 設定」>「動態饋給」。
按一下「新增動態消息」。
按一下「設定單一動態饋給」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Jamf Pro Context logs)。
選取「Google Cloud Storage V2」做為「來源類型」。
選取「Jamf Pro 內容」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬的服務帳戶電子郵件地址，例如：
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
複製這個電子郵件地址，以便在下一步中使用。

注意： 每個 Google SecOps 執行個體都有專屬的服務帳戶。請勿使用其他說明文件或範例中的服務帳戶。

將 IAM 權限授予 Google SecOps 服務帳戶

Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱。
前往「權限」分頁標籤。
按一下「授予存取權」。
請提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」。
按一下 [儲存]。

注意： 如果打算使用「刪除已轉移的檔案」或「刪除已轉移的檔案並清空目錄」刪除選項，請授予「Storage 物件管理員」角色，而非「Storage 物件檢視者」角色。

在 Google SecOps 中設定資訊提供，擷取 Jamf Pro 內容記錄

依序前往「SIEM 設定」>「動態饋給」。
按一下「新增動態消息」。
按一下「設定單一動態饋給」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Jamf Pro Context logs)。
選取「Google Cloud Storage V2」做為「來源類型」。
選取「Jamf Pro 內容」做為「記錄類型」。
點選 [下一步]。
指定下列輸入參數的值：
- 儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：
```
gs://jamfpro/jamf-pro/context/
```
  - 請將 jamfpro 替換為實際值區名稱。
  注意： URI 結尾一律要加上尾端斜線 (/)。
- 來源刪除選項：根據偏好設定選取刪除選項：
  - 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
  - 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
  - 刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。
    
    注意： 如果選取刪除選項，服務帳戶必須具備「Storage 物件管理員」角色，而非「Storage 物件檢視者」角色。請據此更新 IAM 權限。
- 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。
- 資產命名空間：資產命名空間。
- 擷取標籤：要套用至這個動態饋給事件的標籤。
點選 [下一步]。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

需要其他協助嗎？向社群成員和 Google SecOps 專業人員尋求答案。