收集 Aware 稽核記錄
本文說明如何使用 Amazon S3,將 Aware 稽核記錄擷取至 Google Security Operations。
事前準備
請確認您已完成下列事前準備事項:
- Google SecOps 執行個體
- Aware 租戶的特殊存取權
- AWS (S3、IAM、Lambda、EventBridge) 的具備權限存取權
收集 Aware 必要條件 (ID、API 金鑰、機構 ID、權杖)
- 登入 Aware 管理控制台。
- 依序前往「系統設定」>「整合」>「API 權杖」。
- 按一下「+ API Token」,然後授予「Audit Logs Read-only」權限。
- 複製下列詳細資料並儲存於安全位置:
- API 權杖
- API 基礎網址:
https://api.aware.work/external/system/auditlogs/v1
為 Google SecOps 設定 AWS S3 值區和 IAM
- 按照這份使用者指南建立 Amazon S3 值區:建立值區
- 儲存 bucket 的「名稱」和「區域」,以供日後參考 (例如
aware-audit-logs
)。 - 按照這份使用者指南建立使用者:建立 IAM 使用者。
- 選取建立的「使用者」。
- 選取「安全憑證」分頁標籤。
- 在「Access Keys」部分中,按一下「Create Access Key」。
- 選取「第三方服務」做為「用途」。
- 點選「下一步」。
- 選用:新增說明標記。
- 按一下「建立存取金鑰」。
- 按一下「下載 CSV 檔案」,儲存「存取金鑰」和「私密存取金鑰」以供日後使用。
- 按一下 [完成]。
- 選取 [權限] 分頁標籤。
- 在「Permissions policies」(權限政策) 區段中,按一下「Add permissions」(新增權限)。
- 選取「新增權限」。
- 選取「直接附加政策」
- 搜尋並選取 AmazonS3FullAccess 政策。
- 點選「下一步」。
- 按一下「Add permissions」。
設定 S3 上傳的身分與存取權管理政策和角色
- 在 AWS 控制台中,依序前往「IAM」>「Policies」>「Create policy」>「JSON」分頁標籤。
輸入下列政策:
{ "Version": "2012-10-17", "Statement": [ { "Sid": "AllowPutObjects", "Effect": "Allow", "Action": "s3:PutObject", "Resource": "arn:aws:s3:::aware-audit-logs/*" }, { "Sid": "AllowGetStateObject", "Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::aware-audit-logs/aware/state.json" } ] }
- 如果您輸入的值區名稱不同,請替換
aware-audit-logs
。
- 如果您輸入的值區名稱不同,請替換
依序點選「Next」>「Create policy」。
依序前往「IAM」>「Roles」>「Create role」>「AWS service」>「Lambda」。
附加新建立的政策。
為角色命名
AwareAuditLambdaRole
,然後按一下「建立角色」。
建立 Lambda 函式
- 在 AWS 控制台中,依序前往「Lambda」>「Functions」>「Create function」。
- 按一下「從頭開始撰寫」。
- 請提供下列設定詳細資料:
設定 | 值 |
---|---|
名稱 | aware-audit-poller |
執行階段 | Python 3.13 |
架構 | x86_64 |
執行角色 | AwareAuditLambdaRole |
建立函式後,開啟「程式碼」分頁,刪除存根並輸入下列程式碼 (
aware-audit-poller.py
):import boto3, gzip, io, json, os, time, urllib.parse import urllib.request from datetime import datetime, timedelta, timezone from botocore.exceptions import ClientError AWARE_ENDPOINT = "https://api.aware.work/external/system/auditlogs/v1" API_TOKEN = os.environ["AWARE_API_TOKEN"] BUCKET = os.environ["S3_BUCKET"] PREFIX = os.environ.get("S3_PREFIX", "aware/audit/") STATE_KEY = os.environ.get("STATE_KEY", "aware/state.json") MAX_PER_PAGE = int(os.environ.get("MAX_PER_PAGE", "500")) s3 = boto3.client("s3") def _load_state(): try: obj = s3.get_object(Bucket=BUCKET, Key=STATE_KEY) return json.loads(obj["Body"].read().decode("utf-8")) except ClientError as e: if e.response.get("Error", {}).get("Code") == "NoSuchKey": return {} raise def _save_state(state): s3.put_object(Bucket=BUCKET, Key=STATE_KEY, Body=json.dumps(state).encode("utf-8")) def handler(event, context): tz_utc = timezone.utc now = datetime.now(tz=tz_utc) state = _load_state() start_date = ( datetime.fromisoformat(state["last_date"]).date() if "last_date" in state else (now - timedelta(days=1)).date() ) end_date = now.date() total = 0 day = start_date while day <= end_date: day_str = day.strftime("%Y-%m-%d") params = {"filter": f"startDate:{day_str},endDate:{day_str}", "limit": str(MAX_PER_PAGE)} offset = 1 out = io.BytesIO() gz = gzip.GzipFile(filename="aware_audit.jsonl", mode="wb", fileobj=out) wrote_any = False while True: q = urllib.parse.urlencode({**params, "offset": str(offset)}) req = urllib.request.Request(f"{AWARE_ENDPOINT}?{q}") req.add_header("X-Aware-Api-Key", API_TOKEN) with urllib.request.urlopen(req, timeout=30) as resp: payload = json.loads(resp.read().decode("utf-8")) items = (payload.get("value") or {}).get("auditLogData") or [] if not items: break for item in items: gz.write((json.dumps(item, separators=(",", ":")) + "n").encode("utf-8")) total += 1 wrote_any = True offset += 1 time.sleep(0.2) gz.close() if wrote_any: key = f"{PREFIX}{day.strftime('%Y/%m/%d')}/aware_audit_{now.strftime('%Y%m%d_%H%M%S')}.jsonl.gz" s3.put_object( Bucket=BUCKET, Key=key, Body=out.getvalue(), ContentType="application/json", ContentEncoding="gzip", ) _save_state({"last_date": day.isoformat()}) day += timedelta(days=1) return {"status": "ok", "written": total}
依序前往「Configuration」>「Environment variables」>「Edit」>「Add new environment variable」。
輸入下列環境變數,並將 換成您的值:
鍵 範例值 S3_BUCKET
aware-audit-logs
S3_PREFIX
aware/audit/
STATE_KEY
aware/state.json
AWARE_API_TOKEN
<your-aware-api-token>
MAX_PER_PAGE
500
建立函式後,請留在該函式的頁面 (或依序開啟「Lambda」>「Functions」 >「your-function**」)。
選取「設定」分頁標籤。
在「一般設定」面板中,按一下「編輯」。
將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒)),然後按一下「Save」(儲存)。
建立 EventBridge 排程
- 依序前往「Amazon EventBridge」>「Scheduler」>「Create schedule」。
- 提供下列設定詳細資料:
- 週期性時間表:費率 (
1 hour
)。 - 目標:您的 Lambda 函式
aware-audit-poller
。 - 名稱:
aware-audit-poller-1h
。
- 週期性時間表:費率 (
- 按一下「建立時間表」。
選用:為 Google SecOps 建立唯讀 IAM 使用者和金鑰
- 在 AWS 控制台中,依序前往「IAM」>「Users」>「Add users」。
- 點選 [Add users] (新增使用者)。
- 提供下列設定詳細資料:
- 使用者:
secops-reader
。 - 存取類型:存取金鑰 - 程式輔助存取。
- 使用者:
- 按一下「建立使用者」。
- 附加最低讀取權限政策 (自訂):依序選取「使用者」>「secops-reader」>「權限」>「新增權限」>「直接附加政策」>「建立政策」。
在 JSON 編輯器中輸入下列政策:
{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": ["s3:GetObject"], "Resource": "arn:aws:s3:::aware-audit-logs/*" }, { "Effect": "Allow", "Action": ["s3:ListBucket"], "Resource": "arn:aws:s3:::aware-audit-logs" } ] }
將名稱設為
secops-reader-policy
。依序前往「建立政策」> 搜尋/選取 >「下一步」>「新增權限」。
依序前往「安全憑證」>「存取金鑰」>「建立存取金鑰」。
下載 CSV (這些值會輸入至動態饋給)。
在 Google SecOps 中設定動態饋給,擷取 Aware 稽核記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「+ 新增動態消息」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Aware Audit logs
)。 - 選取「Amazon S3 V2」做為「來源類型」。
- 選取「Aware Audit」做為「記錄類型」。
- 點選「下一步」。
- 指定下列輸入參數的值:
- S3 URI:
s3://aware-audit-logs/aware/audit/
- 來源刪除選項:根據偏好設定選取刪除選項。
- 檔案存在時間上限:包含在過去天數內修改的檔案。預設值為 180 天。
- 存取金鑰 ID:具有 S3 值區存取權的使用者存取金鑰。
- 存取密鑰:具有 S3 bucket 存取權的使用者私密金鑰。
- 資產命名空間:資產命名空間。
- 擷取標籤:套用至這個動態饋給事件的標籤。
- S3 URI:
- 點選「下一步」。
- 在「完成」畫面中檢查新的動態饋給設定,然後按一下「提交」。
還有其他問題嗎?向社群成員和 Google SecOps 專業人員尋求答案。