收集 Zendesk CRM 記錄

支援的國家/地區：

Google SecOps SIEM

注意： 這項功能受《Google Security Operations 服務專屬條款》的《正式發布前產品條款》規範。正式發布前的功能或許只提供部分支援，而且這類功能的變更可能與其他正式發布前版本不相容。詳情請參閱《Google SecOps 技術支援服務指南》和《Google SecOps 服務專屬條款》。

本文說明如何使用 Amazon S3，將 Zendesk 客戶關係管理 (CRM) 記錄擷取至 Google Security Operations。

事前準備

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

• Google SecOps 執行個體。
• Zendesk 的特殊存取權。
• AWS 的特殊權限 (S3、Identity and Access Management (IAM)、Lambda、EventBridge)。

取得 Zendesk 必要條件

1. 確認方案和角色
   1. 您必須是 Zendesk 管理員，才能建立 API 權杖 / OAuth 用戶端。稽核記錄 API僅適用於 Enterprise 方案。(如果您的帳戶不是 Enterprise，請略過 audit_logs 中的 RESOURCES)。
2. 啟用 API 權杖存取權 (一次性)
   1. 在管理中心中，依序前往「應用程式和整合」>「API」>「API 設定」。
   2. 啟用「允許 API 權杖存取權」。
3. 產生 API 權杖 (適用於基本驗證)
   1. 依序前往「應用程式和整合」>「API」>「API 權杖」。
   2. 依序點選「新增 API 權杖」> (選用) 新增說明 >「儲存」。
   3. 請立即複製並儲存 API 權杖 (之後就無法再查看)。
   4. 儲存將透過這個權杖驗證的管理員電子郵件地址。
      • Lambda 使用的基本驗證格式：email_address/token:api_token
4. (選用) 建立 OAuth 用戶端 (適用於不記名授權，而非 API 權杖)
   1. 依序前往「應用程式和整合」>「API」>「OAuth 用戶端」>「新增 OAuth 用戶端」。
   2. 填寫「名稱」、「專屬 ID」 (自動)、「重新導向網址」 (如果只透過 API 鑄造權杖，可以填入預留位置) 並「儲存」。
   3. 為整合服務建立存取權杖，並授予本指南所需的最低範圍：
      • tickets:read (適用於增量票券)
      • auditlogs:read (適用於稽核記錄；僅限 Enterprise 版)
      • 如果不確定，read 也適用於唯讀存取權。
   4. 複製存取權杖 (貼到 ZENDESK_BEARER_TOKEN 中)，並安全地記錄用戶端 ID/密鑰 (供日後權杖更新流程使用)。

5. 記錄 Zendesk 基本網址

   • 使用 https://<your_subdomain>.zendesk.com (貼入 ZENDESK_BASE_URL 環境變數)。

   複製並儲存內容以供日後使用

   • 基準網址 (例如 https://acme.zendesk.com)
   • 管理員使用者的電子郵件地址 (適用於 API 權杖驗證)
   • API 權杖 (如使用 AUTH_MODE=token)
   • 或 OAuth 存取權杖 (如果使用 AUTH_MODE=bearer)
   • (選用) 用於生命週期管理的 OAuth 用戶端 ID/密鑰

為 Google SecOps 設定 AWS S3 值區和 IAM

1. 按照這份使用者指南建立 Amazon S3 bucket：建立 bucket
2. 儲存 bucket 的「名稱」和「區域」，以供日後參考 (例如 zendesk-crm-logs)。
3. 請按照這份使用者指南建立使用者：建立 IAM 使用者。
4. 選取建立的「使用者」。
5. 選取「安全憑證」分頁標籤。
6. 在「Access Keys」部分中，按一下「Create Access Key」。
7. 選取「第三方服務」做為「用途」。
8. 點選「下一步」。
9. 選用：新增說明標記。
10. 按一下「建立存取金鑰」。
11. 按一下「下載 .CSV 檔案」，儲存「存取金鑰」和「私密存取金鑰」以供日後參考。
12. 按一下 [完成]。
13. 選取「權限」分頁標籤。
14. 在「權限政策」部分中，按一下「新增權限」。
15. 選取「新增權限」。
16. 選取「直接附加政策」。
17. 搜尋「AmazonS3FullAccess」AmazonS3FullAccess政策。
18. 選取政策。
19. 點選「下一步」。
20. 按一下「Add permissions」。

設定 S3 上傳的身分與存取權管理政策和角色

1. 在 AWS 控制台中，依序前往「IAM」>「Policies」(政策)。
2. 按一下「建立政策」>「JSON」分頁。
3. 複製並貼上下列政策。

4. 政策 JSON (如果您輸入的 bucket 名稱不同，請替換 zendesk-crm-logs)：

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "AllowPutObjects",
         "Effect": "Allow",
         "Action": "s3:PutObject",
         "Resource": "arn:aws:s3:::zendesk-crm-logs/*"
       },
       {
         "Sid": "AllowGetStateObject",
         "Effect": "Allow",
         "Action": "s3:GetObject",
         "Resource": "arn:aws:s3:::zendesk-crm-logs/zendesk/crm/state.json"
       }
     ]
   }
   

5. 依序點選「下一步」>「建立政策」。

6. 依序前往「IAM」>「Roles」>「Create role」>「AWS service」>「Lambda」。

7. 附加新建立的政策。

8. 為角色命名 ZendeskCRMToS3Role，然後按一下「建立角色」。

建立 Lambda 函式

1. 在 AWS 控制台中，依序前往「Lambda」>「Functions」>「Create function」。
2. 按一下「從頭開始撰寫」。

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

   設定	值
   名稱	zendesk_crm_to_s3
   執行階段	Python 3.13
   架構	x86_64
   執行角色	ZendeskCRMToS3Role

4. 建立函式後，開啟「程式碼」分頁，刪除存根並貼上以下程式碼 (zendesk_crm_to_s3.py)。

   #!/usr/bin/env python3
   
   import os, json, time, base64
   from urllib.request import Request, urlopen
   from urllib.error import HTTPError, URLError
   import boto3
   
   S3_BUCKET    = os.environ["S3_BUCKET"]
   S3_PREFIX    = os.environ.get("S3_PREFIX", "zendesk/crm/")
   STATE_KEY    = os.environ.get("STATE_KEY", "zendesk/crm/state.json")
   BASE_URL     = os.environ["ZENDESK_BASE_URL"].rstrip("/")  # e.g. https://your_subdomain.zendesk.com
   AUTH_MODE    = os.environ.get("AUTH_MODE", "token").lower()  # token|bearer
   EMAIL        = os.environ.get("ZENDESK_EMAIL", "")
   API_TOKEN    = os.environ.get("ZENDESK_API_TOKEN", "")
   BEARER       = os.environ.get("ZENDESK_BEARER_TOKEN", "")
   RESOURCES    = [r.strip() for r in os.environ.get("RESOURCES", "audit_logs,incremental_tickets").split(",") if r.strip()]
   MAX_PAGES    = int(os.environ.get("MAX_PAGES", "20"))
   LOOKBACK     = int(os.environ.get("LOOKBACK_SECONDS", "3600"))  # 1h default
   HTTP_TIMEOUT = int(os.environ.get("HTTP_TIMEOUT", "60"))
   HTTP_RETRIES = int(os.environ.get("HTTP_RETRIES", "3"))
   
   s3 = boto3.client("s3")
   
   def _headers() -> dict:
       if AUTH_MODE == "bearer" and BEARER:
           return {"Authorization": f"Bearer {BEARER}", "Accept": "application/json"}
       if AUTH_MODE == "token" and EMAIL and API_TOKEN:
           token = base64.b64encode(f"{EMAIL}/token:{API_TOKEN}".encode()).decode()
           return {"Authorization": f"Basic {token}", "Accept": "application/json"}
       raise RuntimeError("Invalid auth settings: provide token (EMAIL + API_TOKEN) or BEARER")
   
   def _get_state() -> dict:
       try:
           obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
           b = obj["Body"].read()
           return json.loads(b) if b else {"audit_logs": {}, "incremental_tickets": {}}
       except Exception:
           return {"audit_logs": {}, "incremental_tickets": {}}
   
   def _put_state(st: dict) -> None:
       s3.put_object(
           Bucket=S3_BUCKET, Key=STATE_KEY,
           Body=json.dumps(st, separators=(",", ":")).encode("utf-8"),
           ContentType="application/json",
       )
   
   def _http_get_json(url: str) -> dict:
       attempt = 0
       while True:
           try:
               req = Request(url, method="GET")
               for k, v in _headers().items():
                   req.add_header(k, v)
               with urlopen(req, timeout=HTTP_TIMEOUT) as r:
                   return json.loads(r.read().decode("utf-8"))
           except HTTPError as e:
               if e.code in (429, 500, 502, 503, 504) and attempt < HTTP_RETRIES:
                   ra = 1 + attempt
                   try:
                       ra = int(e.headers.get("Retry-After", ra))
                   except Exception:
                       pass
                   time.sleep(max(1, ra))
                   attempt += 1
                   continue
               raise
           except URLError:
               if attempt < HTTP_RETRIES:
                   time.sleep(1 + attempt)
                   attempt += 1
                   continue
               raise
   
   def _put_page(payload: dict, resource: str) -> str:
       ts = time.gmtime()
       key = f"{S3_PREFIX}/{time.strftime('%Y/%m/%d/%H%M%S', ts)}-zendesk-{resource}.json"
       s3.put_object(
           Bucket=S3_BUCKET, Key=key,
           Body=json.dumps(payload, separators=(",", ":")).encode("utf-8"),
           ContentType="application/json",
       )
       return key
   
   def fetch_audit_logs(state: dict):
       """GET /api/v2/audit_logs.json with pagination via `next_page` (Zendesk)."""
       next_url = state.get("next_url") or f"{BASE_URL}/api/v2/audit_logs.json?page=1"
       pages = 0
       written = 0
       last_next = None
       while pages < MAX_PAGES and next_url:
           data = _http_get_json(next_url)
           _put_page(data, "audit_logs")
           written += len(data.get("audit_logs", []))
           last_next = data.get("next_page")
           next_url = last_next
           pages += 1
       return {"resource": "audit_logs", "pages": pages, "written": written, "next_url": last_next}
   
   def fetch_incremental_tickets(state: dict):
       """Cursor-based incremental export: /api/v2/incremental/tickets/cursor.json (pagination via `links.next`)."""
       next_link = state.get("next")
       if not next_link:
           start = int(time.time()) - LOOKBACK
           next_link = f"{BASE_URL}/api/v2/incremental/tickets/cursor.json?start_time={start}"
       pages = 0
       written = 0
       last_next = None
       while pages < MAX_PAGES and next_link:
           data = _http_get_json(next_link)
           _put_page(data, "incremental_tickets")
           written += len(data.get("tickets", []))
           links = data.get("links") or {}
           next_link = links.get("next")
           last_next = next_link
           pages += 1
       return {"resource": "incremental_tickets", "pages": pages, "written": written, "next": last_next}
   
   def lambda_handler(event=None, context=None):
       state = _get_state()
       summary = []
   
       if "audit_logs" in RESOURCES:
           res = fetch_audit_logs(state.get("audit_logs", {}))
           state["audit_logs"] = {"next_url": res.get("next_url")}
           summary.append(res)
   
       if "incremental_tickets" in RESOURCES:
           res = fetch_incremental_tickets(state.get("incremental_tickets", {}))
           state["incremental_tickets"] = {"next": res.get("next")}
           summary.append(res)
   
       _put_state(state)
       return {"ok": True, "summary": summary}
   
   if __name__ == "__main__":
       print(lambda_handler())
   

5. 依序前往「設定」>「環境變數」。

6. 依序點選「編輯」> 新增環境變數。

7. 輸入下表提供的環境變數，並將範例值換成您的值。

   環境變數

   鍵	範例值
   S3_BUCKET	zendesk-crm-logs
   S3_PREFIX	zendesk/crm/
   STATE_KEY	zendesk/crm/state.json
   ZENDESK_BASE_URL	https://your_subdomain.zendesk.com
   AUTH_MODE	token
   ZENDESK_EMAIL	[e-mail]
   ZENDESK_API_TOKEN	<api_token>
   ZENDESK_BEARER_TOKEN	<leave empty unless using OAuth bearer>
   RESOURCES	audit_logs,incremental_tickets
   MAX_PAGES	20
   LOOKBACK_SECONDS	3600
   HTTP_TIMEOUT	60

8. 建立函式後，請留在函式頁面 (或依序開啟「Lambda」>「Functions」>「your-function」)。

9. 選取「設定」分頁標籤。

10. 在「一般設定」面板中，按一下「編輯」。

11. 將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒))，然後按一下「Save」(儲存)。

    注意： 逾時是上限，也就是說，如果函式在計時器達到 5 分鐘時仍在執行，AWS Lambda 就會終止函式。如果執行時間較短，系統會正常完成執行作業，並只收取實際執行時間的費用。

建立 EventBridge 排程

1. 依序前往「Amazon EventBridge」>「Scheduler」>「Create schedule」。
2. 提供下列設定詳細資料：
   • 週期性時間表：費率 (1 hour)。
   • 目標：您的 Lambda 函式 zendesk_crm_to_s3。
   • 名稱：zendesk_crm_to_s3-1h。
3. 按一下「建立時間表」。

(選用) 為 Google SecOps 建立唯讀 IAM 使用者和金鑰

1. 依序前往 AWS 管理中心 > IAM >「Users」(使用者) >「Add users」(新增使用者)。
2. 點選 [Add users] (新增使用者)。
3. 提供下列設定詳細資料：
   • 使用者：輸入 secops-reader。
   • 存取類型：選取「存取金鑰 - 程式輔助存取」。
4. 按一下「建立使用者」。
5. 附加最低讀取權限政策 (自訂)：依序選取「Users」(使用者) >「secops-reader」>「Permissions」(權限) >「Add permissions」(新增權限) >「Attach policies directly」(直接附加政策) >「Create policy」(建立政策)。

6. JSON：

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": ["s3:GetObject"],
         "Resource": "arn:aws:s3:::zendesk-crm-logs/*"
       },
       {
         "Effect": "Allow",
         "Action": ["s3:ListBucket"],
         "Resource": "arn:aws:s3:::zendesk-crm-logs"
       }
     ]
   }
   

7. Name = secops-reader-policy。

8. 依序點選「建立政策」> 搜尋/選取 >「下一步」>「新增權限」。

9. 為 secops-reader 建立存取金鑰：依序點選「安全憑證」>「存取金鑰」。

10. 按一下「建立存取金鑰」。

11. 下載 .CSV。(您會將這些值貼到動態饋給中)。

在 Google SecOps 中設定動態饋給，擷取 Zendesk CRM 記錄

1. 依序前往「SIEM 設定」>「動態饋給」。
2. 按一下「+ 新增動態消息」。
3. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Zendesk CRM logs)。
4. 選取「Amazon S3 V2」做為「來源類型」。
5. 選取「Zendesk CRM」做為「記錄類型」。
6. 點選「下一步」。
7. 指定下列輸入參數的值：
   • S3 URI：s3://zendesk-crm-logs/zendesk/crm/
   • 來源刪除選項：根據偏好設定選取刪除選項。
   • 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。
   • 存取金鑰 ID：具有 S3 值區存取權的使用者存取金鑰。
   • 存取密鑰：具有 S3 bucket 存取權的使用者私密金鑰。
   • 資產命名空間：資產命名空間。
   • 擷取標籤：套用至這個動態饋給事件的標籤。
8. 點選「下一步」。
9. 在「完成」畫面中檢查新的動態饋給設定，然後按一下「提交」。

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。