Cloudflare R2 に Logpush したログを ALog Cloud に連携する

# 参考：ALog Cloud とは
無料トライアルを活用させていただきました。

https://www.amiya.co.jp/lp/alog_cloud/

>ALogはオンプレからクラウドまで、あらゆるシステムのログデータを収集・分析・保管する純国産のSIEM製品です。
>“わかりやすさ”を追及したALogは、あらゆるITシステムのログを自動で収集、一元的に管理するSIEM製品です。
「セキュリティの難しいをカンタンに」を、コンセプトに専門知識やノウハウがなくとも、誰でもカンタンに高度なログ分析を実現。
企業がかかえる課題に沿って、幅広いシーンで活用できるのが特徴です。

# R2 に Logpush したデータをマウントしてフォワードする
ALog Cloud のログ取り込み方式を考慮し、以下のような構成を実装します。

必ずしも Linux サーバーを別に立てる必要はありませんが、切り分けのために別立てします。

```mermaid
%%{init: {'themeCSS': 'img {width: 48px; height:48px;}'}}%%
graph LR
subgraph Cloudflare
    Logpush["<img src='https://cf-icons.pages.dev/logs.svg' width='48' height='48'><br>#emsp;Logs#emsp;"]
    R2-1["<img src='https://cf-icons.pages.dev/r2.svg' width='48' height='48'><br>#emsp;R2#emsp;<br>*.log.gz"]
    R2-2["<img src='https://cf-icons.pages.dev/r2.svg' width='48' height='48'><br>#emsp;R2#emsp;<br>*.csv"]
    Queue[<img src='https://cf-icons.pages.dev/queues.svg' width='48' height='48'><br>Queue]
    Workers[<img src='https://cf-icons.pages.dev/workers.svg' width='48' height='48'><br>Workers]
    Logpush --> |Logpush|R2-1
    R2-1 --> |Event notification|Queue --> |Trigger|Workers --> |gunzip|R2-2
end

subgraph Linux
    Directory["`${HOME}/alog-cloud`"] --> |"`s3fs`" mount|R2-2
end

subgraph Windows
    Forwarder[ALog Forwarder] --> |SCP Pull|Directory
end

Forwarder ---> |"`*.csv`" upload|ALog["<img src="https://www.amiya.co.jp/wp-content/themes/Ami-CorporateWeb/assets/old/images/ja/logos/alog.svg" width='48' height='48'><br>#emsp;#emsp;ALog Cloud#emsp;#emsp;"]
```

## R2 バケットに Event notification を設定
以下の記事を参考にして進めます。

https://developers.cloudflare.com/r2/examples/upload-logs-event-notifications/

必要となる R2, Queue, Workers を作成します。

```shell:
# R2 バケットを作成
export LOGPUSH_BUCKET_NAME='alog-gzip'
export MOUNT_BUCKET_NAME='alog-cloud'
wrangler r2 bucket create $LOGPUSH_BUCKET_NAME
wrangler r2 bucket create $MOUNT_BUCKET_NAME

# Queue を作成
export QUEUE_NAME='alog-gunzip-queue'
wrangler queues create $QUEUE_NAME

# Workers を作成
npm create cloudflare@latest -- alog-gunzip-worker
cd alog-gunzip-worker
```

オブジェクト書き込みのイベント通知ごとに処理を起動し、`*.log.gz` の `gzip` ファイル形式から `*.csv` にすることで Forwarder からのアップロードに対応します。 

https://zenn.dev/takoyaki3/articles/354f5f6a6eee61

https://github.com/kyouheicf/alog-gunzip-worker

```toml:wrangler.toml
name = "alog-gunzip-worker"
main = "src/index.ts"
compatibility_date = "2024-08-15"
compatibility_flags = ["nodejs_compat"]

[[queues.consumers]]
queue = "alog-gunzip-queue"
max_batch_size = 100
max_batch_timeout = 5

[[r2_buckets]]
binding = "LOGPUSH_BUCKET"
bucket_name = "alog-gzip"

[[r2_buckets]]
binding = "MOUNT_BUCKET"
bucket_name = "alog-cloud"
```

```typescript:src/index.ts
import { ungzip } from 'pako';

export interface Env {
	LOGPUSH_BUCKET: R2Bucket;
	MOUNT_BUCKET: R2Bucket;
}
  
export default {
	async queue(batch, env): Promise<void> {
		// A queue consumer can make requests to other endpoints on the Internet,
		// write to R2 object storage, query a D1 Database, and much more.
		for (const message of batch.messages) {
			// Process each message (we'll just log these)
			console.log(`Message: ${JSON.stringify(message.body)}`);
			const fileName = message.body["object"]["key"];
			console.log(`object.key: ${fileName}`);
			//gunzip data into decompressedData
			const response = await env.LOGPUSH_BUCKET.get(fileName);
			const decompressedData = ungzip(new Uint8Array(await response.arrayBuffer()));
			// Write the batch of messages to R2
			console.log(`object.target: ${fileName.split('.')[0]}`);
			const fileTargetName = fileName.split('.')[0] + ".csv";
			await env.MOUNT_BUCKET.put(fileTargetName, decompressedData, {
				httpMetadata: {
					contentType: "text/csv",
				},
			});
			// Explicitly acknowledge the message as delivered
			message.ack();
		}
	},
} satisfies ExportedHandler<Env>;
```

最後に Workers をデプロイして、R2 バケットに Event notification を設定します。

```shell:
wrangler deploy
wrangler r2 bucket notification create $LOGPUSH_BUCKET_NAME --event-type object-create --queue $QUEUE_NAME
```

## Logpush を CSV で R2 に出力
以下のコマンドで `gateway_http` データセットの Logpush を作成します。

`output_type` で `csv` を指定し、`destination_conf` で R2 を指定します。

R2 バケット内では日付ごとにフォルダ分けはせず、`timestamp_format` は `unix` にします。

```shell:
export EMAIL='YOUR_EMAIL'
export APIKEY='YOUR_APIKEY'
export ACCOUNT_ID='YOUR_ACCOUNT_ID'
export R2_ACCESS_KEY_ID='YOUR_R2_ACCESS_KEY_ID'
export R2_SECRET_ACCESS_KEY='YOUR_R2_SECRET_ACCESS_KEY'

# 全フィールドを指定
export FIELDS=$(curl -s \
-H "X-Auth-Email: $EMAIL" \
-H "X-Auth-Key: $APIKEY" \
"https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/logpush/datasets/gateway_http/fields" | jq -r '.result | keys | join("\",\"")')
echo $FIELDS

# Logpush ジョブを作成
curl -s "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/logpush/jobs" -X POST -d '
{                                              
  "name": "gateway-http-r2-alog",
  "output_options": {
    "field_names": ["'$FIELDS'"],
    "output_type": "csv",
    "timestamp_format": "unix"
  },
  "destination_conf": "r2://alog-gzip?account-id='$ACCOUNT_ID'&access-key-id='$R2_ACCESS_KEY_ID'&secret-access-key='$R2_SECRET_ACCESS_KEY'",
  "max_upload_bytes": 5000000,
  "max_upload_records": 1000,
  "dataset": "gateway_http",
  "enabled": true
}' \
-H "X-Auth-Email: $EMAIL" \
-H "X-Auth-Key: $APIKEY"
```

## s3fs を使って R2 を Linux にマウント
以下の記事を参考にします。

https://blog.serverworks.co.jp/mount-s3-as-local-drive

以下のコマンドで R2 をマウントできます。

```shell:
# s3fs インストール
sudo yum install epel-release -y
sudo yum install s3fs-fuse -y

# クルデンシャルファイル作成
export R2_ACCESS_KEY_ID='YOUR_R2_ACCESS_KEY_ID'
export R2_SECRET_ACCESS_KEY='YOUR_R2_SECRET_ACCESS_KEY'
echo $R2_ACCESS_KEY_ID:$R2_SECRET_ACCESS_KEY > ${HOME}/.passwd-s3fs
chmod 600 ${HOME}/.passwd-s3fs

# マウントポイント作成
mkdir alog-cloud

# R2 をマウント
export ACCOUNT_ID='YOUR_ACCOUNT_ID'
s3fs -o passwd_file=${HOME}/.passwd-s3fs -o endpoint=auto \
-o url="https://$ACCOUNT_ID.r2.cloudflarestorage.com" alog-cloud ${HOME}/alog-cloud
```

以下のように R2 に Logpush された `*.log.gz` ファイルを、Event notification によって連携した Workers で解凍した `*.csv` ファイルが確認できます。

```
$ tree alog-cloud/
alog-cloud/
|-- test.csv
|-- 20240821T054839Z_20240821T054839Z_06b7e0be.csv
|-- 20240821T054914Z_20240821T054914Z_8f641fd1.csv
|-- 20240821T055213Z_20240821T055213Z_81de4248.csv
|-- 20240821T055338Z_20240821T055338Z_bf3b5176.csv
...
```

## ALog Forwarder 設定
以下のガイドに従ってログ連携します。

https://www.amiya.co.jp/support/alog/product/raw/alogcloud/user_guide/source/startup.html#index-1

ALog Cloud の管理画面から Forwarder をダウンロードし、インストールします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/101361/9f8f4854-6b96-9ca1-daa5-15df61aea8ca.png)

ALog Cloud の管理画面からクルデンシャルを新規作成し、Forwarder でアクティベーションを実行します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/101361/8388adc0-c710-eb4d-f547-faa226c55574.png)

対象ホストの登録は以下のように設定できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/101361/e4a275de-0b08-00ea-593b-35ecf478e20c.png)


## ALog Cloud マッピング設定
管理＞対象ホストから以下のマッピング設定を行い、ログを見やすい形で取り込みます。

https://www.amiya.co.jp/support/alog/product/raw/alogcloudforwarder/user_guide/source/mapping.html#

`TimeStamp` は `unix` シリアル値の `DateTime` フィールドを参照し、`ConvSerialToDateTime` 関数で取り込むことができます。

その他の詳細フィールドは適宜見やすい形に定義できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/101361/4ba08fcb-7fc7-d0b6-2b05-16e7b53d0e31.png)

検索画面にて以下のようにログを取り込んで活用できる状態になったことが確認できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/101361/c43bdc4b-3f3a-4e7b-5b46-23afba259ed7.png)

# 参考：ALog（オンプレミス版）を利用した、よりシンプルな構成
ALog（オンプレミス版）を利用すると、gzip `.gz` ファイルを解凍しながらアップロードできるため、よりシンプルな構成を考えることができます。

```mermaid
%%{init: {'themeCSS': 'img {width: 48px; height:48px;}'}}%%
graph LR
subgraph Cloudflare
    Logpush["<img src='https://cf-icons.pages.dev/logs.svg' width='48' height='48'><br>#emsp;Logs#emsp;"]
    R2["<img src='https://cf-icons.pages.dev/r2.svg' width='48' height='48'><br>#emsp;R2#emsp;<br>*.log.gz"]
    Logpush --> |Logpush|R2
end

subgraph Linux
    Directory["`${HOME}/alog-cloud`"] --> |"`s3fs`" mount|R2
end

subgraph Windows
    Forwarder[ALog<br>On-premises<br>Edition] --> |"SCP Pull<br>(gunzip)"|Directory
end

Forwarder ---> |"`*.csv`" upload|ALog["<img src="https://www.amiya.co.jp/wp-content/themes/Ami-CorporateWeb/assets/old/images/ja/logos/alog.svg" width='48' height='48'><br>#emsp;#emsp;ALog Cloud#emsp;#emsp;"]
```

# まとめ

Cloudflare からの Logpush で直接の宛先として対応していない SIEM に対しても、R2 + Queue + Workers を使って柔軟にファイル形式を変換する処理を実現できました。

こうしたログ処理のパイプラインを一度構築してしまえば、後は自動で取り込みが実行されていくため、SIEM 側でのレポート・アラートを活用した運用に注力していくことができます。