このページは Cloud Translation API によって翻訳されました。

Swimlane Platform のログを収集する

以下でサポートされています。

Google SecOps SIEM

このドキュメントでは、Amazon S3 を使用して Swimlane Platform ログを Google Security Operations に取り込む方法について説明します。

始める前に

次の前提条件を満たしていることを確認してください。

Google SecOps インスタンス
Swimlane への特権アクセス（個人アクセストークンを生成できるアカウント管理者）
AWS（S3、IAM、Lambda、EventBridge）への特権アクセス

Swimlane Platform の前提条件（ID、API キー、組織 ID、トークン）を収集する

アカウント管理者として Swimlane Platform にログインします。
[プロフィールオプション] に移動します。
[プロフィール] をクリックして、プロフィールエディタを開きます。
[個人用アクセストークン] セクションに移動します。
[Generate token] をクリックして、新しい個人用アクセストークンを作成します。
トークンをすぐにコピーして安全に保管します（トークンは再度表示されません）。
統合に関する次の詳細を記録します。
- 個人用アクセストークン（PAT）: API 呼び出しの Private-Token ヘッダーで使用されます。
- アカウント ID: 監査ログ API パス /api/public/audit/account/{ACCOUNT_ID}/auditlogs に必要です。アカウント ID がわからない場合は、Swimlane 管理者にお問い合わせください。
- ベース URL: Swimlane ドメイン（https://eu.swimlane.app、https://us.swimlane.app など）。

Google SecOps 用に AWS S3 バケットと IAM を構成する

バケットの作成のユーザーガイドに沿って、Amazon S3 バケットを作成します。
後で参照できるように、バケットの名前とリージョンを保存します（例: swimlane-audit）。
IAM ユーザーの作成のユーザーガイドに沿って、ユーザーを作成します。
作成したユーザーを選択します。
[セキュリティ認証情報] タブを選択します。
[アクセスキー] セクションで [アクセスキーを作成] をクリックします。
[ユースケース] で [サードパーティサービス] を選択します。
[次へ] をクリックします。
省略可: 説明タグを追加します。
[アクセスキーを作成] をクリックします。
[CSV ファイルをダウンロード] をクリックして、[アクセスキー] と [シークレットアクセスキー] を保存し、後で使用できるようにします。
[完了] をクリックします。
[権限] タブを選択します。
[権限ポリシー] セクションで、[権限を追加] をクリックします。
[権限を追加] を選択します。
[ポリシーを直接アタッチする] を選択します。
AmazonS3FullAccess ポリシーを検索して選択します。
[次へ] をクリックします。
[権限を追加] をクリックします。

S3 アップロードの IAM ポリシーとロールを構成する

AWS コンソールで、[IAM] > [ポリシー] > [ポリシーの作成] > [JSON] タブに移動します。

次のポリシーを入力します。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutSwimlaneAuditObjects",
      "Effect": "Allow",
      "Action": ["s3:PutObject"],
      "Resource": "arn:aws:s3:::swimlane-audit/swimlane/audit/*"
    },
    {
      "Sid": "AllowStateReadWrite",
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:PutObject"],
      "Resource": "arn:aws:s3:::swimlane-audit/swimlane/audit/state.json"
    }
  ]
}

別のバケット名を入力した場合は、swimlane-audit を置き換えます。

[次へ] > [ポリシーを作成] をクリックします。
[IAM] > [ロール] > [ロールの作成] > [AWS サービス] > [Lambda] に移動します。
新しく作成したポリシーと AWS マネージドポリシーを関連付けます。
- 上記で作成したカスタムポリシー
- service-role/AWSLambdaBasicExecutionRole（CloudWatch Logs）
ロールに「WriteSwimlaneAuditToS3Role」という名前を付けて、[ロールを作成] をクリックします。

Lambda 関数を作成する

AWS コンソールで、[Lambda] > [Functions] > [Create function] に移動します。
[Author from scratch] をクリックします。
次の構成情報を提供してください。

設定値

名前 swimlane_audit_to_s3

ランタイム Python 3.13

アーキテクチャ x86_64

実行ロール WriteSwimlaneAuditToS3Role

設定	値
名前	`swimlane_audit_to_s3`
ランタイム	Python 3.13
アーキテクチャ	x86_64
実行ロール	`WriteSwimlaneAuditToS3Role`

関数を作成したら、[コード] タブを開き、スタブを削除して次のコード（swimlane_audit_to_s3.py）を入力します。

#!/usr/bin/env python3
import os, json, gzip, io, uuid, datetime as dt, urllib.parse, urllib.request
import boto3

# ---- Environment ----
S3_BUCKET = os.environ["S3_BUCKET"]
S3_PREFIX = os.environ.get("S3_PREFIX", "swimlane/audit/")
STATE_KEY = os.environ.get("STATE_KEY", S3_PREFIX + "state.json")
BASE_URL = os.environ["SWIMLANE_BASE_URL"].rstrip("/")  # e.g., https://eu.swimlane.app
ACCOUNT_ID = os.environ["SWIMLANE_ACCOUNT_ID"]
TENANT_LIST = os.environ.get("SWIMLANE_TENANT_LIST", "")  # comma-separated; optional
INCLUDE_ACCOUNT = os.environ.get("INCLUDE_ACCOUNT", "true").lower() == "true"
PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "100"))  # max 100
WINDOW_MINUTES = int(os.environ.get("WINDOW_MINUTES", "15"))  # time range per run
PAT_TOKEN = os.environ["SWIMLANE_PAT_TOKEN"]  # Personal Access Token
TIMEOUT = int(os.environ.get("TIMEOUT", "30"))

AUDIT_URL = f"{BASE_URL}/api/public/audit/account/{ACCOUNT_ID}/auditlogs"

s3 = boto3.client("s3")

# ---- Helpers ----

def _http(req: urllib.request.Request):
    return urllib.request.urlopen(req, timeout=TIMEOUT)

def _now():
    return dt.datetime.utcnow()

def get_state() -> dict:
    try:
        obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
        return json.loads(obj["Body"].read())
    except Exception:
        return {}

def put_state(state: dict) -> None:
    state["updated_at"] = _now().isoformat() + "Z"
    s3.put_object(Bucket=S3_BUCKET, Key=STATE_KEY, Body=json.dumps(state).encode())

def build_url(from_dt: dt.datetime, to_dt: dt.datetime, page: int) -> str:
    params = {
        "pageNumber": str(page),
        "pageSize": str(PAGE_SIZE),
        "includeAccount": str(INCLUDE_ACCOUNT).lower(),
        "fromdate": from_dt.replace(microsecond=0).isoformat() + "Z",
        "todate": to_dt.replace(microsecond=0).isoformat() + "Z",
    }
    if TENANT_LIST:
        params["tenantList"] = TENANT_LIST
    return AUDIT_URL + "?" + urllib.parse.urlencode(params)

def fetch_page(url: str) -> dict:
    headers = {
        "Accept": "application/json",
        "Private-Token": PAT_TOKEN,
    }
    req = urllib.request.Request(url, headers=headers)
    with _http(req) as r:
        return json.loads(r.read())

def write_chunk(items: list[dict], ts: dt.datetime) -> str:
    key = f"{S3_PREFIX}{ts:%Y/%m/%d}/swimlane-audit-{uuid.uuid4()}.json.gz"
    buf = io.BytesIO()
    with gzip.GzipFile(fileobj=buf, mode="w") as gz:
        for rec in items:
            gz.write((json.dumps(rec) + "n").encode())
    buf.seek(0)
    s3.upload_fileobj(buf, S3_BUCKET, key)
    return key

def lambda_handler(event=None, context=None):
    state = get_state()

    # determine window
    to_dt = _now()
    from_dt = to_dt - dt.timedelta(minutes=WINDOW_MINUTES)
    if (prev := state.get("last_to_dt")):
        try:
            from_dt = dt.datetime.fromisoformat(prev.replace("Z", "+00:00"))
        except Exception:
            pass

    page = int(state.get("page", 1))
    total_written = 0

    while True:
        url = build_url(from_dt, to_dt, page)
        resp = fetch_page(url)
        items = resp.get("auditlogs", []) or []
        if items:
            write_chunk(items, _now())
            total_written += len(items)
        next_path = resp.get("next")
        if not next_path:
            break
        page += 1
        state["page"] = page

    # advance state window
    state["last_to_dt"] = to_dt.replace(microsecond=0).isoformat() + "Z"
    state["page"] = 1
    put_state(state)

    return {"ok": True, "written": total_written, "from": from_dt.isoformat() + "Z", "to": to_dt.isoformat() + "Z"}

if __name__ == "__main__":
    print(lambda_handler())

[構成] > [環境変数] に移動します。
[編集>新しい環境変数を追加] をクリックします。

次の環境変数を入力し、実際の値に置き換えます。

キー	値の例
`S3_BUCKET`	`swimlane-audit`
`S3_PREFIX`	`swimlane/audit/`
`STATE_KEY`	`swimlane/audit/state.json`
`SWIMLANE_BASE_URL`	`https://eu.swimlane.app`
`SWIMLANE_ACCOUNT_ID`	`xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx`
`SWIMLANE_TENANT_LIST`	`tenantA,tenantB`（オプション）
`INCLUDE_ACCOUNT`	`true`
`PAGE_SIZE`	`100`
`WINDOW_MINUTES`	`15`
`SWIMLANE_PAT_TOKEN`	`<your-personal-access-token>`
`TIMEOUT`	`30`

関数が作成されたら、そのページにとどまるか、[Lambda] > [関数] > [your-function] を開きます。
[CONFIGURATION] タブを選択します。
[全般設定] パネルで、[編集] をクリックします。
[Timeout] を [5 minutes (300 seconds)] に変更し、[Save] をクリックします。

注: タイムアウトは上限です。つまり、タイマーが 5 分に達したときに関数がまだ実行されている場合、AWS Lambda は関数を終了します。実行時間が短くなった場合は、通常どおり終了します。料金は実際の実行時間に対してのみ課金されます。

EventBridge スケジュールを作成する

[Amazon EventBridge] > [Scheduler] > [スケジュールの作成] に移動します。
次の構成の詳細を入力します。
- 定期的なスケジュール: レート（15 min）
- ターゲット: Lambda 関数 swimlane_audit_to_s3
- 名前: swimlane-audit-schedule-15min
[スケジュールを作成] をクリックします。

省略可: Google SecOps 用の読み取り専用の IAM ユーザーと鍵を作成する

AWS コンソールで、[IAM] > [Users] > [Add users] に移動します。
[ユーザーを追加] をクリックします。
次の構成の詳細を入力します。
- ユーザー: secops-reader
- アクセスタイプ: アクセスキー - プログラムによるアクセス
[ユーザーを作成] をクリックします。
最小限の読み取りポリシー（カスタム）を関連付ける: [ユーザー] > [secops-reader] > [権限] > [権限を追加] > [ポリシーを直接関連付ける] > [ポリシーを作成]。

JSON エディタで、次のポリシーを入力します。

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

名前を secops-reader-policy に設定します。
[ポリシーの作成> 検索/選択> 次へ> 権限を追加] に移動します。
[セキュリティ認証情報] > [アクセスキー] > [アクセスキーを作成] に移動します。
CSV をダウンロードします（これらの値はフィードに入力されます）。

Swimlane Platform のログを取り込むように Google SecOps でフィードを構成する

[SIEM 設定] > [フィード] に移動します。
[+ 新しいフィードを追加] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: Swimlane Platform logs）。
[ソースタイプ] として [Amazon S3 V2] を選択します。
[ログタイプ] として [Swimlane Platform] を選択します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
- S3 URI: s3://swimlane-audit/swimlane/audit/
- Source deletion options: 必要に応じて削除オプションを選択します。
- ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
- アクセスキー ID: S3 バケットにアクセスできるユーザーアクセスキー。
- シークレットアクセスキー: S3 バケットにアクセスできるユーザーのシークレットキー。
- アセットの名前空間: アセットの名前空間。
- Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

さらにサポートが必要な場合 コミュニティメンバーや Google SecOps のプロフェッショナルから回答を得ることができます。