以前、SoftetherVPNの認証ログ集計スクリプトを公開していました。
当初は深夜にその日の認証ログを集計する想定だったのですが、期間を指定して集計したいという状況になり、Claudeに更新してもらいました。
分析するデータが増えるとスクリプト動いているのかわからないので、プログレスバーを表示したいと伝えたところ、Pythonで書き換えることになりました。
それで出来上がったのは以下のコード。
auth_log_aggregator.py
#!/usr/bin/env python3
# ============================================================
# auth_log_aggregator.py
#
# SoftEther VPN の認証ログ(認証成功/認証失敗)を集計し、
# 結果をコンソールに表示、またはオプションで
# Google Chat の Webhook に通知するツールです。
#
# 使い方・オプション一覧・攻撃サマリー機能の説明・cron運用例・注意事項は
# 以下を実行して確認してください:
# python3 auth_log_aggregator.py --help
#
# 事前設定: すぐ下の「--- 設定 ---」セクションで
# VPN_SERVER_DIR / VIRTUAL_HUB / GCHAT_WEBHOOK_URL を
# 環境に合わせて編集してください。
# ============================================================
import argparse
import glob
import json
import os
import re
import socket
import sys
import urllib.error
import urllib.request
from datetime import datetime, timedelta
# --- 設定 ---------------------------------------------------
# SoftEther VPN Server のインストールディレクトリ（環境に合わせて変更）
#
# ここは環境によって2パターンあります。使う方だけ残し、もう片方はコメントアウトしてください。
#
# [パターンA] softether-automator.sh 等でOSに直接インストールしている場合 ← デフォルトはこちら
# 通常通り /usr/vpnserver がインストール先になります。
#
# [パターンB] softether-docker (Docker運用) の場合
# ホストから見えるセキュリティログの実体は /usr/vpnserver ではなく、
# docker-compose.yml でbind mountしているデータ領域になります。
# 本スクリプトはホストで実行する前提のため、Docker運用の場合はこちらに切り替えてください。
#
VPN_SERVER_DIR = "/usr/vpnserver" # [パターンA] OSに直接インストールの場合
# VPN_SERVER_DIR = "/opt/softether-docker/data" # [パターンB] Docker運用の場合
# 集計対象の仮想HUB名（環境に合わせて変更）
VIRTUAL_HUB = "YourVirtualHubName"
# セキュリティログの格納ディレクトリは VPN_SERVER_DIR と 仮想HUB名 から組み立てる。
# 仮想HUB名は --hub で実行時に上書きできるため、モジュールレベルの定数にはせず、
# 実際に使う箇所 (find_target_files) で都度組み立てる。
# ログファイル名のプレフィックス（通常は変更不要, 例: sec_20240101.log）
LOG_PREFIX = "sec_"
# 認証成功/失敗を判定するためのキーワード（通常は変更不要）
SUCCESS_KEYWORD = "Successfully authenticated as user"
FAIL_KEYWORD_RE = re.compile(r"(User authentication failed|User .* not found)")
# Google Chat Webhook URL
# -w オプション指定時に集計結果を送信します。
# 未設定（デフォルト値のまま）の場合は送信をスキップします。
# 取得方法: Google Chat のスペース > 設定 > 「Webhookを管理」
GCHAT_WEBHOOK_URL = "your_webhook_url"
# 認証失敗詳細を表示する際、1ユーザーあたり最大何件表示するか
MAX_FAIL_DETAIL_LINES = 20
# --- 設定終わり -----------------------------------------------
# ログファイル名から日付部分 (YYYYMMDD) を抽出する正規表現
FILENAME_RE = re.compile(re.escape(LOG_PREFIX) + r"(\d{8})\.log$")
# 認証失敗ログからユーザー名を抽出するための正規表現パターン群
# （SoftEtherのログ形式のバリエーションに対応するため複数用意）
FAIL_USER_PATTERNS = [
re.compile(r'The user name that has been provided was "([^"]+)"'),
re.compile(r'user "([^"]+)"'),
re.compile(r"User (\S+) not found"),
]
# 接続ID（例: "CID-12345"）を抽出するための正規表現。
# SoftEtherのログでは、認証成功/失敗の行にはIPアドレスが
# 含まれない代わりに、接続試行時の行と共通の接続IDが振られている。
CID_RE = re.compile(r'"(CID-\d+)"')
# 「接続試行」ログから 接続ID と 接続元IPアドレス を抽出する正規表現。
# 例:
# The connection "CID-2324" (IP address: 126.103.173.27, Host name: ...)
# is attempting to connect to the Virtual Hub. ...
ATTEMPT_RE = re.compile(
r'The connection "(CID-\d+)" \(IP address: (\d{1,3}(?:\.\d{1,3}){3})'
)
def extract_cid(line):
"""ログ行から接続ID（CID-xxxxx）を抽出する。
見つからない場合は None を返す。
"""
m = CID_RE.search(line)
return m.group(1) if m else None
# ユーザー名の正規化で除去する文字。
# - _ . : メールアドレスやログインIDで実際によく使われる区切り文字
# , : アカウント名として正規に使われることはまずないが、
# キーボード上で "." のすぐ隣にあるため、"." の打ち間違いで
# 誤って入力されるケースが多い（例: "taro.yamada" のつもりが
# "taro,yamada" になる）。そのためタイプミスとして扱う。
USERNAME_SEPARATOR_RE = re.compile(r"[-_.,]")
def normalize_username(username):
"""ユーザー名から区切り文字およびタイプミスされやすい記号
（- _ . ,）を除去し、小文字化して正規化する。
タイプミス判定（同一アカウントかどうかの比較）に使う。
"""
return USERNAME_SEPARATOR_RE.sub("", username).lower()
def count_distinct_accounts(usernames):
"""
同一IPから試行された複数のユーザー名について、区切り文字や
打ち間違いされやすい記号の違い（例: "toshiaki.yoshida" と
"toshiaki,yoshida"、"yuki" と "yuki.nomura"）を同一アカウントの
タイプミスとみなしてグルーピングし、実質的に異なるアカウントの
数を返す。
正規化後の文字列が完全一致、またはどちらかがもう一方の部分文字列
であれば「同じアカウントへのタイプミス」とみなして1グループに
まとめる。これにより、単なる入力ミスが「複数アカウント試行」の
誤検知につながるのを防ぐ。
"""
usernames = list(usernames)
n = len(usernames)
if n <= 1:
return n
normalized = [normalize_username(u) for u in usernames]
parent = list(range(n))
def find(x):
while parent[x] != x:
parent[x] = parent[parent[x]]
x = parent[x]
return x
def union(a, b):
ra, rb = find(a), find(b)
if ra != rb:
parent[ra] = rb
for i in range(n):
for j in range(i + 1, n):
a, b = normalized[i], normalized[j]
if a == b or a in b or b in a:
union(i, j)
return len({find(i) for i in range(n)})
def parse_args():
"""コマンドライン引数を解析する。"""
epilog = """\
■ 前提条件
- Python 3.7 以上（標準ライブラリのみで動作、pip install 不要）
- security_log が有効な SoftEther VPN Server
- 本スクリプトを実行するユーザーがログディレクトリを読み取れる権限を持っていること
■ 事前設定（このファイル冒頭の「--- 設定 ---」セクションを編集）
VPN_SERVER_DIR : SoftEther VPN Server のインストールディレクトリ
(OS直接インストール/Docker運用の2パターンをコメントで用意。使う方を選ぶ)
VIRTUAL_HUB : 集計対象の仮想HUB名（--hub で実行時にも上書き可）
GCHAT_WEBHOOK_URL : Google Chat の Webhook URL（-w/--webhook 使用時のみ必要。
--webhook <URL> で実行時にも上書き可。秘密情報として扱うこと）
■ 使用例
auth_log_aggregator.py
当日分を集計してコンソールに表示（デフォルト）
auth_log_aggregator.py -d 30
過去30日分を集計
auth_log_aggregator.py -d 7 -q -w
cron向け: 進捗バー等を抑制しつつGoogle Chat Webhookへ通知
auth_log_aggregator.py --details
認証失敗ログの詳細（日時・ユーザー名等）も表示
auth_log_aggregator.py --caution-threshold 15 --warning-threshold 30
攻撃判定の閾値を変更（デフォルト: 注意10回/警告20回）
auth_log_aggregator.py --hub AnotherHub
仮想HUB名を実行時に指定（複数HUBを運用している場合など）
auth_log_aggregator.py -w --webhook https://chat.googleapis.com/v1/spaces/xxx/messages?key=...
Webhook送信先URLを実行時に指定
■ 攻撃サマリー機能について
SoftEtherのセキュリティログでは、接続元IPアドレスは「接続試行」ログにのみ記録され、
認証成功/失敗そのもののログ行には含まれません。両者は同じ接続ID(CID)で紐付けられて
いるため、本ツールはCID経由でIPを逆引きして特定します。
以下いずれかに該当すると「攻撃の疑いあり」として表示します（どちらも該当しなければ
「該当期間に攻撃の疑いのあるログはありませんでした」と表示）。
- 失敗回数が --warning-threshold 以上 → 🚨警告 / --caution-threshold 以上 → ⚠️注意
（ブルートフォース攻撃の疑い）
- 試行された実質的に異なるアカウント数が --multi-user-threshold 以上
→ 🎯複数アカウント試行（アカウント総当たり/クレデンシャルスタッフィングの疑い）
※ 区切り文字( - _ . , )の違いはタイプミスとみなして同一アカウント扱いにする
（例: "taro.yamada" と "taro,yamada" は同一人物の打ち間違いとみなす）
CIDが解決できず送信元IPを特定できなかった失敗は「IP不明」として攻撃判定の対象外です
（--details 指定時のみ件数を注記表示）。
■ cron運用例（毎朝9時に過去1日分をGoogle Chatへ自動通知）
0 9 * * * /usr/bin/python3 /path/to/auth_log_aggregator.py -q -w
■ 注意事項
- 本スクリプトはログファイルを読み取り専用で扱い、ログの削除・変更は行いません。
- Webhook通知内容には --details 指定時、ユーザー名等の個人情報が含まれる場合が
あります。通知先のアクセス範囲にご注意ください。
"""
p = argparse.ArgumentParser(
description="SoftEther VPN 認証ログ集計ツール - "
"認証成功/失敗を集計してコンソール表示、"
"またはGoogle Chat Webhookへ通知する",
epilog=epilog,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
p.add_argument(
"-d", "--days", type=int, default=1,
help="集計対象の日数（過去n日、当日を含む。デフォルト: 1）",
)
p.add_argument(
"-w", "--webhook", nargs="?", const="", default=None, metavar="URL",
help="集計結果をGoogle Chat Webhookに送信する。"
"URLを省略した場合はスクリプト内のデフォルト値(GCHAT_WEBHOOK_URL)を使う。"
"値を指定した場合はそのURLへ送信する（デフォルト値を上書き）",
)
p.add_argument(
"--hub", type=str, default=None, metavar="HUB_NAME",
help="集計対象の仮想HUB名を指定する。"
"省略した場合はスクリプト内のデフォルト値(VIRTUAL_HUB)を使う",
)
p.add_argument(
"-q", "--quiet", action="store_true",
help="進捗バーや対象ファイルの詳細表示を抑制する（cron実行等向け）",
)
p.add_argument(
"--details", action="store_true",
help="認証失敗ログの詳細（日時・元ログ行）も表示する（通常は失敗回数のみ）",
)
p.add_argument(
"--caution-threshold", type=int, default=10,
help="同一IPからの1日あたり失敗回数がこの値以上で「⚠️注意」表示する"
"（デフォルト: 10回。想定運用: 1日1回の夜間集計）",
)
p.add_argument(
"--warning-threshold", type=int, default=20,
help="同一IPからの1日あたり失敗回数がこの値以上で「🚨警告」表示する"
"（デフォルト: 20回）",
)
p.add_argument(
"--multi-user-threshold", type=int, default=5,
help="同一IPが試行した「実質的に異なるアカウント数」"
"（区切り文字違いのタイプミスは同一アカウントとみなして"
"除外した後の数）がこの値以上の場合、失敗回数に関わらず"
"複数アカウント総当たりの疑いとして表示する（デフォルト: 5）",
)
return p.parse_args()
def find_target_files(days, quiet, virtual_hub):
"""
LOG_DIR (VPN_SERVER_DIR/security_log/virtual_hub) 配下から、
指定した日数分の対象ログファイルを探して日付の古い順にソートした
パスのリストを返す。
対象ファイルが1件も見つからない場合はエラー終了する。
"""
log_dir = os.path.join(VPN_SERVER_DIR, "security_log", virtual_hub)
all_files = glob.glob(os.path.join(log_dir, f"{LOG_PREFIX}*.log"))
cutoff_date = datetime.now().date() - timedelta(days=days - 1)
dated_files = []
for path in all_files:
m = FILENAME_RE.search(os.path.basename(path))
if not m:
continue
file_date = datetime.strptime(m.group(1), "%Y%m%d").date()
if file_date >= cutoff_date:
dated_files.append((file_date, path))
dated_files.sort(key=lambda x: x[0])
if not dated_files:
print(
f"🚨 過去{days}日間の対象ログファイルが見つかりませんでした。"
f"LOG_DIR ({log_dir}) の設定を確認してください。",
file=sys.stderr,
)
sys.exit(1)
if not quiet:
first_date = dated_files[0][0].strftime("%Y-%m-%d")
last_date = dated_files[-1][0].strftime("%Y-%m-%d")
print(f"--- 🔰 SoftEther VPN 認証ログ集計ツール (過去 {days} 日間) 🔰 ---")
print(f"サーバー名: {socket.gethostname()}")
print(f"仮想HUB名: {virtual_hub}")
print(
f"--- 🔍 集計対象ログファイル: {len(dated_files)} 件 / 全{len(all_files)}件中 "
f"(最初の日: {first_date} 〜 最後の日: {last_date}) ---"
)
return [p for _, p in dated_files]
def print_progress(current, total, quiet, width=40):
"""ログ集計の進捗をバー表示する（--quiet時は何もしない）。"""
if quiet:
return
ratio = (current / total) if total else 1.0
filled = int(width * ratio)
bar = "█" * filled + "-" * (width - filled)
sys.stdout.write(f"\r集計中: |{bar}| {ratio * 100:5.1f}% ({current}/{total} バイト)")
sys.stdout.flush()
if current >= total:
sys.stdout.write("\n")
def extract_fail_username(line):
"""認証失敗ログの1行からユーザー名を抽出する。
どのパターンにも一致しない場合は 'UNKNOWN' を返す。
"""
for pattern in FAIL_USER_PATTERNS:
m = pattern.search(line)
if m:
return m.group(1)
return "UNKNOWN"
def aggregate(target_files, quiet):
"""
対象ログファイル群を1行ずつ読み、認証成功/失敗を集計する。
戻り値の辞書:
user_success : {ユーザー名: 成功回数}
user_fail_counts : {ユーザー名: 失敗回数}
user_fail_details : {ユーザー名: [失敗ログの詳細行, ...]}
ip_fail_counts : {送信元IP: 失敗回数}（外部攻撃検出用。
IPが特定できなかった失敗は "IP不明" に集計）
ip_fail_usernames : {送信元IP: {試行されたユーザー名の集合}}
total_success : 成功の合計回数
total_fail : 失敗の合計回数
"""
user_success = {}
user_fail_counts = {}
user_fail_details = {}
ip_fail_counts = {}
ip_fail_usernames = {}
total_success = 0
total_fail = 0
# 接続ID(CID) → 接続元IPアドレス の対応表。
# 「接続試行」ログを読んだ時点でここに登録し、後続の認証失敗
# ログが来たときに同じCIDでIPを逆引きする。
cid_to_ip = {}
total_bytes = sum(os.path.getsize(p) for p in target_files) or 1
processed_bytes = 0
print_progress(0, total_bytes, quiet)
for path in target_files:
fname = os.path.basename(path)
try:
with open(path, "r", encoding="utf-8", errors="replace") as f:
for line in f:
# --- 接続試行ログ: CID と 接続元IP の対応を記録 ---
# （認証成功/失敗より必ず先に出現する行）
m = ATTEMPT_RE.search(line)
if m:
cid_to_ip[m.group(1)] = m.group(2)
if SUCCESS_KEYWORD in line:
total_success += 1
# SoftEtherのログはダブルクォート区切りでユーザー名を含む
parts = line.split('"')
username = parts[3] if len(parts) > 3 and parts[3] else "UNKNOWN"
user_success[username] = user_success.get(username, 0) + 1
elif FAIL_KEYWORD_RE.search(line):
total_fail += 1
username = extract_fail_username(line)
user_fail_counts[username] = user_fail_counts.get(username, 0) + 1
timestamp = " ".join(line.split()[:2])
user_fail_details.setdefault(username, []).append(
f"{fname} {timestamp}: {line.strip()}"
)
# --- 外部攻撃検出用: 同じCIDから接続元IPを逆引きし、
# IPごとに失敗回数と試行されたユーザー名を集計する ---
cid = extract_cid(line)
ip = cid_to_ip.get(cid) if cid else None
ip_key = ip if ip else "IP不明"
ip_fail_counts[ip_key] = ip_fail_counts.get(ip_key, 0) + 1
ip_fail_usernames.setdefault(ip_key, set()).add(username)
except OSError as e:
print(f"\n⚠️ ファイルを読み込めませんでした: {path} ({e})", file=sys.stderr)
processed_bytes += os.path.getsize(path)
print_progress(processed_bytes, total_bytes, quiet)
return {
"user_success": user_success,
"user_fail_counts": user_fail_counts,
"user_fail_details": user_fail_details,
"ip_fail_counts": ip_fail_counts,
"ip_fail_usernames": ip_fail_usernames,
"total_success": total_success,
"total_fail": total_fail,
}
def build_attack_summary(result, caution_threshold, warning_threshold,
multi_user_threshold, show_details):
"""
ip_fail_counts / ip_fail_usernames から「外部からの攻撃」の
疑いがある送信元IPのサマリー行リストを組み立てる。
判定は独立した2つの観点で行う（どちらか一方でも該当すれば「攻撃の疑いあり」）:
1. 失敗回数（ブルートフォース攻撃の疑い）
count >= warning_threshold → 🚨警告
count >= caution_threshold → ⚠️注意
2. 試行された「実質的に異なるアカウント数」（アカウント総当たり/
クレデンシャルスタッフィングの疑い）
distinct_accounts >= multi_user_threshold → 🎯複数アカウント試行
※ ここでの distinct_accounts は、区切り文字（- _ . ,）違いの
タイプミス（例: "toshiaki.yoshida" と "toshiaki,yoshida"）を
同一アカウントとみなして除外した後の数（count_distinct_accounts
参照）。単純にタイプミスを2〜3回しただけの正当な利用者を
誤検知しないようにするための工夫。
CIDが解決できず送信元IPを特定できなかった失敗は「IP不明」として
別枠で扱い、攻撃疑いの判定対象からは除外する
（複数の異なる送信元が混在している可能性があるため）。
表示は3パターン:
A. 認証失敗ログが1件もない → その旨だけ表示
B. 認証失敗はあるが、該当するIPが0件 → その旨だけ表示
C. 該当するIPが1件以上ある → ⚠️注意/🚨警告/🎯複数アカウント試行
それぞれの該当件数を表示
（--details指定時は該当IPの内訳も表示）
"""
lines = []
lines.append("## 🛡️ 外部からの攻撃サマリー (Suspicious Source IPs)")
ip_fail_counts = result["ip_fail_counts"]
ip_fail_usernames = result["ip_fail_usernames"]
# --- パターンA: 認証失敗ログが1件もない ---
if not ip_fail_counts:
lines.append("該当期間に認証失敗ログはありませんでした。")
return lines
known_ip_counts = {ip: c for ip, c in ip_fail_counts.items() if ip != "IP不明"}
unknown_count = ip_fail_counts.get("IP不明", 0)
def judge(ip, count):
distinct_accounts = count_distinct_accounts(ip_fail_usernames.get(ip, set()))
flags = []
if count >= warning_threshold:
flags.append("🚨警告")
elif count >= caution_threshold:
flags.append("⚠️注意")
if distinct_accounts >= multi_user_threshold:
flags.append("🎯複数アカウント試行")
return distinct_accounts, flags
# IPごとに判定結果を先に計算しておく
judged = {ip: judge(ip, c) for ip, c in known_ip_counts.items()}
caution_ips = {ip for ip, (_, flags) in judged.items() if "⚠️注意" in flags}
warning_ips = {ip for ip, (_, flags) in judged.items() if "🚨警告" in flags}
multi_ips = {ip for ip, (_, flags) in judged.items() if "🎯複数アカウント試行" in flags}
suspicious_ips = {ip for ip, (_, flags) in judged.items() if flags}
# --- パターンB: 認証失敗はあるが、該当する（攻撃の疑いがある）IPが0件 ---
if not suspicious_ips:
lines.append("該当期間に攻撃の疑いのあるログはありませんでした。")
if show_details and unknown_count:
lines.append(
f"（参考: 送信元IPを特定できなかった失敗が別途 {unknown_count} 件あります）"
)
return lines
# --- パターンC: 該当するIPが1件以上ある ---
lines.append(f"⚠️注意: 失敗{caution_threshold}回以上 : {len(caution_ips)} 件")
lines.append(f"🚨警告: 失敗{warning_threshold}回以上 : {len(warning_ips)} 件")
lines.append(f"🎯複数アカウント試行 : {len(multi_ips)} 件")
if show_details:
if unknown_count:
lines.append(
f"（参考: 送信元IPを特定できなかった失敗が別途 {unknown_count} 件"
f"あります。複数の送信元が混在している可能性があるため、"
f"上記の集計には含めていません）"
)
lines.append("")
lines.append("```")
lines.append(f"{'送信元IP':<17} | {'失敗回数':>8} | {'異なるアカウント数':>14} | 判定")
lines.append("-" * 17 + " | " + "-" * 8 + " | " + "-" * 14 + " | " + "-" * 6)
for ip in sorted(suspicious_ips, key=lambda x: known_ip_counts[x], reverse=True):
count = known_ip_counts[ip]
distinct_accounts, flags = judged[ip]
flag = " ".join(flags)
lines.append(f"{ip:<17} | {count:>8} | {distinct_accounts:>14} | {flag}")
lines.append("```")
return lines
def build_report(days, result, show_details, virtual_hub,
caution_threshold=10, warning_threshold=20, multi_user_threshold=5):
"""集計結果からMarkdown形式のレポート文字列を組み立てる。"""
lines = []
lines.append(f"## 📊 認証サマリー (過去 {days} 日間) 📊")
lines.append(f"### サーバー: {socket.gethostname()} / HUB: {virtual_hub}")
lines.append("---")
lines.append(f"✅ 成功合計: **{result['total_success']}** 回")
lines.append(f"❌ 失敗合計: **{result['total_fail']}** 回")
lines.append("---")
lines.append("")
lines.append("## 👤 ユーザーごとの認証成功回数 (Success Count)")
if result["user_success"]:
lines.append("```")
lines.append(f"{'ユーザー名':<35} | {'成功回数'}")
lines.append("-" * 37 + " | " + "-" * 8)
for username, count in sorted(
result["user_success"].items(), key=lambda x: x[1], reverse=True
):
lines.append(f"{username:<35} | {count:>8}")
lines.append("```")
else:
lines.append("該当期間に認証成功ログは見つかりませんでした。")
lines.append("")
lines.append("## ❌ ユーザーごとの認証失敗回数 (Fail Count)")
if result["user_fail_counts"]:
lines.append("```")
lines.append(f"{'ユーザー名':<35} | {'失敗回数'}")
lines.append("-" * 37 + " | " + "-" * 8)
for username, count in sorted(
result["user_fail_counts"].items(), key=lambda x: x[1], reverse=True
):
lines.append(f"{username:<35} | {count:>8}")
lines.append("```")
else:
lines.append("該当期間に認証失敗ログは見つかりませんでした。")
lines.append("")
lines.extend(build_attack_summary(
result, caution_threshold, warning_threshold, multi_user_threshold, show_details
))
lines.append("")
if show_details:
lines.append("## ⚠️ ユーザーごとの認証失敗ログ詳細 (Fail Details)")
if result["user_fail_details"]:
for username, detail_lines in result["user_fail_details"].items():
lines.append(f"### ユーザー: {username} (失敗件数: {len(detail_lines)})")
lines.append("```")
for dl in detail_lines[:MAX_FAIL_DETAIL_LINES]:
lines.append(dl)
if len(detail_lines) > MAX_FAIL_DETAIL_LINES:
lines.append(
f"... (以降 {len(detail_lines)} 件中、上位"
f"{MAX_FAIL_DETAIL_LINES}件のみ表示)..."
)
lines.append("```")
lines.append("")
else:
lines.append("該当期間に認証失敗ログは見つかりませんでした。")
lines.append("--- 🏁 集計完了 🏁 ---")
return "\n".join(lines)
def send_webhook(message, webhook_url):
"""
レポート文字列を Google Chat の Webhook に POST する。
webhook_url が未設定（ダミー値のまま）の場合は送信をスキップする。
"""
if webhook_url in ("your_webhook_url", "", None):
print(
"⚠️ Webhookオプションが指定されましたが、送信先URLが"
"未設定（ダミー値）のままです。通知をスキップします。"
"\n → スクリプト冒頭の設定セクションでGCHAT_WEBHOOK_URLを設定するか、"
"実行時に --webhook <URL> を指定してください。",
file=sys.stderr,
)
return
payload = json.dumps({"text": message}).encode("utf-8")
req = urllib.request.Request(
webhook_url,
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
body = resp.read().decode("utf-8", errors="replace")
if '"error"' in body:
print(
f"⚠️ Webhook送信に成功しましたが、Google Chat APIからエラー応答が"
f"ありました: {body}",
file=sys.stderr,
)
except urllib.error.URLError as e:
print(f"🚨 エラー: Webhook送信に失敗しました ({e})", file=sys.stderr)
def main():
args = parse_args()
virtual_hub = args.hub if args.hub else VIRTUAL_HUB
target_files = find_target_files(args.days, args.quiet, virtual_hub)
result = aggregate(target_files, args.quiet)
report = build_report(
args.days, result, args.details, virtual_hub,
args.caution_threshold, args.warning_threshold, args.multi_user_threshold,
)
print()
print(report)
if args.webhook is not None:
webhook_url = args.webhook if args.webhook else GCHAT_WEBHOOK_URL
send_webhook(report, webhook_url)
if __name__ == "__main__":
main()
基本の実行例（-d 7）
$ ./auth_log_aggregator.py -d 7
--- 🔰 SoftEther VPN 認証ログ集計ツール (過去 7 日間) 🔰 ---
サーバー名: vpn01.example.com
仮想HUB名: Sample-Hub
--- 🔍 集計対象ログファイル: 7 件 / 全92件中 (最初の日: 2026-07-23 〜 最後の日: 2026-07-29) ---
集計中: |████████████████████████████████████████| 100.0% (48213899/48213899 バイト)
## 📊 認証サマリー (過去 7 日間) 📊
### サーバー: vpn01.example.com / HUB: Sample-Hub
---
✅ 成功合計: **342** 回
❌ 失敗合計: **187** 回
---
## 👤 ユーザーごとの認証成功回数 (Success Count)
ユーザー名 | 成功回数
------------------------------------- | --------
taro.yamada | 58
hanako.suzuki | 47
ichiro.tanaka | 39
misaki.sato | 33
kenji.watanabe | 28
yui.kobayashi | 24
## ❌ ユーザーごとの認証失敗回数 (Fail Count)
ユーザー名 | 失敗回数
------------------------------------- | --------
admin | 62
root | 41
test | 28
taro.yamada | 3
taro,yamada | 1
## 🛡️ 外部からの攻撃サマリー (Suspicious Source IPs)
⚠️注意: 失敗10回以上 : 1 件
🚨警告: 失敗20回以上 : 1 件
🎯複数アカウント試行 : 1 件
送信元IP | 失敗回数 | 異なるアカウント数 | 判定
----------------- | -------- | -------------- | ------
203.0.113.45 | 98 | 12 | 🚨警告 🎯複数アカウント試行
198.51.100.7 | 15 | 2 | ⚠️注意
--- 🏁 集計完了 🏁 ---
攻撃サマリー欄の3パターン（記事で判定ロジックを説明する際に）
認証失敗が0件の場合
## 🛡️ 外部からの攻撃サマリー (Suspicious Source IPs)
該当期間に認証失敗ログはありませんでした。
失敗はあるが、どの基準にも該当しない場合（タイプミス程度）
## 🛡️ 外部からの攻撃サマリー (Suspicious Source IPs)
該当期間に攻撃の疑いのあるログはありませんでした。
該当するIPがある場合（上の基本例と同じパターン）
## 🛡️ 外部からの攻撃サマリー (Suspicious Source IPs)
⚠️注意: 失敗10回以上 : 1 件
🚨警告: 失敗20回以上 : 1 件
🎯複数アカウント試行 : 1 件
Webhook送信時のメッセージ例（-w実行時にターミナルへ出る補足）
$ ./auth_log_aggregator.py -d 7 -q -w
⚠️ Webhookオプションが指定されましたが、送信先URLが未設定（ダミー値）のままです。通知をスキップします。
→ スクリプト冒頭の設定セクションでGCHAT_WEBHOOK_URLを設定するか、実行時に --webhook <URL> を指定してください。
実際にURLを設定していれば、この警告は出ずにそのままGoogle Chatへ投稿されます。