Home / KUSANAGI / WordPress(KUSANAGI)のログ事項解析・自動防御

WordPress(KUSANAGI)のログ事項解析・自動防御

LinuxサーバーやWordPress(KUSANAGI)を運用していると、/var/log/messages や Nginx のアクセスログ・エラーログに日々大量の記録が蓄積されます。しかし、すべてのログを目視で追いかけ、ブルートフォース攻撃や脆弱性スキャン、5xxエラーの兆候をいち早く見つけ出すのは容易ではありません。

本記事では、AlmaLinux 9 + KUSANAGI環境において、Gemini API(google-genai SDK) を活用したログ自動解析・自動防御システムをゼロから構築する手順を解説します。

本記事で構築するシステムの全体像

  1. 複数ログの差分抽出(logrotate対応): /var/log/messages と Nginx の各種ログから「前回読み取った位置以降の差分」のみを抽出。
  2. トークン消費を劇的に減らす事前フィルタ: 画像やCSSなどの正常アクセスをPython側で除外し、エラーや不審な通信がある時だけAPIを呼び出す省エネ設計。
  3. firewalld (ipset) による自動IP遮断: Geminiが「明らかな攻撃元」と判定したIPアドレスを、ホワイトリスト保護付きで即座にブラックリストへ自動投入。
  4. Googleドライブ/スプレッドシートへの日次保存: サーバー上にGoogleのOAuthトークンを置かず、Google Apps Script (GAS) のWebhook経由で安全に日次レポートをクラウド保存。
  5. systemdタイマーによる15分間隔の自動実行: APIサーバーの一時的な混雑(503エラー等)にも自動でリトライする堅牢なサービス構成。

1. Gemini APIキーの発行とセキュリティ対策(IP制限)

サーバー上にAPIキーを配置する以上、万が一キーが漏洩しても第三者に悪用されないよう、Google Cloudコンソール側で送信元IPアドレス制限とAPI制限をかけておきます。

1-1. サーバーのグローバルIPアドレスを確認する

GoogleのAPIサーバーはIPv6通信に対応しています。サーバーでIPv6が有効な場合、IPv6経由でリクエストが飛ぶことが多いため、IPv4とIPv6の両方を確認して控えておきます。

# 外向きIPv4アドレスの確認
curl -4 -s [https://ifconfig.me](https://ifconfig.me) && echo ""

# 外向きIPv6アドレスの確認
curl -6 -s [https://ifconfig.me](https://ifconfig.me) && echo ""

1-2. Google Cloudコンソールで制限を設定する

  1. Google AI Studio でAPIキーを発行します。
  2. Google Cloud コンソールの「認証情報」画面 を開き、対象のプロジェクト(Default Gemini Project 等)を選択します。
  3. 発行したAPIキーをクリックし、「アプリケーションの制限」 で 「IP アドレス」 を選択します。
  4. 先ほど確認したサーバーのIPv4アドレス(例: 203.0.113.10)とIPv6アドレス(例: 2001:db8:abcd:1234::1)を登録します。
  5. 「API の制限」 で 「キーを制限」 を選択し、Generative Language API のみにチェックを入れて保存します。

ポイント: これにより、万が一APIキー文字列が外部に流出しても、指定したAlmaLinuxサーバー以外のIPからは一切APIを利用できなくなります。また、Google Cloudの「予算とアラート」から月額1,000円などの少額アラートを設定しておくとさらに安心です。


2. Python 3.12仮想環境とGoogle GenAI SDKの準備

AlmaLinux 9のシステム標準Pythonは 3.9 ですが、Googleの認証ライブラリ(google-auth)でPython 3.9を使用するとサポート期限切れの警告(FutureWarning)が出力されます。

OS標準の dnf 等に影響を与えないよう、AppStreamリポジトリから Python 3.12 を追加インストールし、スクリプト専用の仮想環境(venv)を構築します。

# 1. ディレクトリの作成
sudo mkdir -p /opt/gemini-log-analyzer
sudo mkdir -p /var/lib/gemini-log-analyzer
sudo mkdir -p /var/log/gemini-log-analyzer

# 2. Python 3.12のインストールと仮想環境(venv)の作成
sudo dnf install -y python3.12
sudo python3.12 -m venv /opt/gemini-log-analyzer/venv

# 3. 新公式SDK(google-genai)のインストール
sudo /opt/gemini-log-analyzer/venv/bin/pip install --upgrade pip
sudo /opt/gemini-log-analyzer/venv/bin/pip install google-genai

# バージョン確認(Python 3.12.x と表示されればOK)
/opt/gemini-log-analyzer/venv/bin/python3 --version

3. firewalld(ipset)によるブラックリスト基盤の作成

攻撃元IPを1件ずつ rich-rule で追加していくと、件数が増えた際に firewalld の処理負荷が高くなります。そのため、ipset(blacklist)を作成して drop ゾーンに紐付ける高速な遮断基盤を作ります。

# blacklist という名前の ipset を作成し、dropゾーンに紐付ける
sudo firewall-cmd --permanent --get-ipsets | grep -qw blacklist || {
  sudo firewall-cmd --permanent --new-ipset=blacklist --type=hash:net
  sudo firewall-cmd --permanent --zone=drop --add-source=ipset:blacklist
  sudo firewall-cmd --reload
}

# 設定確認(ipset:blacklist と表示されればOK)
sudo firewall-cmd --zone=drop --list-sources

4. Google Apps Script (GAS) によるクラウド保存の準備

サーバー上にGoogleドライブのOAuth認証トークンを配置するリスクを避けるため、GASでWebhook(受け皿)を作成し、PythonからJSONをPOST送信して日次ログファイル(およびスプレッドシート)へ記録します。

4-1. GAS側のコード(日次ファイル作成・排他制御対応)

  1. Googleドライブで保存先フォルダを開き、URL末尾のフォルダID(folders/【ここの文字列】)を控えます。
  2. Google Apps Script で新規プロジェクトを作成し、以下のコードを貼り付けます。
// ===== 設定項目 =====
const SECRET_TOKEN = "your-random-secret-token"; // サーバー側と合わせる合言葉
const DRIVE_FOLDER_ID = "YOUR_GOOGLE_DRIVE_FOLDER_ID"; // 保存先フォルダID
const SPREADSHEET_ID = ""; // スプレッドシートにも記録する場合はIDを入力(不要なら空欄)
// ====================

function doPost(e) {
  const lock = LockService.getScriptLock();
  try {
    // 同時書き込みによる競合を防ぐためロックを取得(最大10秒待機)
    lock.waitLock(10000);

    const data = JSON.parse(e.postData.contents);

    // 1. トークン認証
    if (data.token !== SECRET_TOKEN) {
      return ContentService.createTextOutput("Forbidden").setMimeType(ContentService.MimeType.TEXT);
    }

    const now = new Date();
    const timestamp = data.timestamp || Utilities.formatDate(now, "JST", "yyyy-MM-dd HH:mm:ss");
    const hostname = data.hostname || "server";
    const severity = data.severity || "不明";
    const report = data.report || "";

    // 2. Googleスプレッドシートへの追記(SPREADSHEET_ID 設定時のみ)
    if (SPREADSHEET_ID) {
      const ss = SpreadsheetApp.openById(SPREADSHEET_ID);
      const sheet = ss.getSheets()[0];
      if (sheet.getLastRow() === 0) {
        sheet.appendRow(["解析日時", "ホスト名", "重要度判定", "解析レポート詳細"]);
      }
      sheet.appendRow([timestamp, hostname, severity, report]);
    }

    // 3. Googleドライブへ1日単位のファイル(analysis_ホスト名_yyyy-MM-dd.log)で保存・追記
    if (DRIVE_FOLDER_ID) {
      const folder = DriveApp.getFolderById(DRIVE_FOLDER_ID);
      const dateStr = Utilities.formatDate(now, "JST", "yyyy-MM-dd");
      const fileName = `analysis_${hostname}_${dateStr}.log`;

      const logEntry = report.startsWith("===")
        ? report
        : `========================================\n解析日時: ${timestamp} (${hostname})\n========================================\n${report}\n\n`;

      const files = folder.getFilesByName(fileName);
      if (files.hasNext()) {
        const file = files.next();
        const currentContent = file.getBlob().getDataAsString("UTF-8");
        file.setContent(currentContent + logEntry);
      } else {
        folder.createFile(fileName, logEntry);
      }
    }

    return ContentService.createTextOutput("OK").setMimeType(ContentService.MimeType.TEXT);
  } catch (err) {
    return ContentService.createTextOutput("Error: " + err.message).setMimeType(ContentService.MimeType.TEXT);
  } finally {
    lock.releaseLock();
  }
}
  1. 右上の 「デプロイ」>「新しいデプロイ」 から種類 「ウェブアプリ」 を選択し、次のユーザーとして実行を 「自分」、アクセスできるユーザーを 「全員」 にしてデプロイします。
  2. 発行された ウェブアプリのURL(https://script.google.com/macros/s/.../exec) を控えます。
    (※後からGASのコードを修正した場合は、「デプロイを管理」>「編集」>「新バージョン」としてデプロイするとURLを変えずに反映できます)

5. 環境変数ファイルの作成

AlmaLinux側で /etc/sysconfig/gemini-log-analyzer を作成し、APIキー、GASのWebhook情報、および誤って自動遮断しないための管理者ホワイトリストIPを設定します。パーミッションは 600 に設定し、root 以外の閲覧を遮断します。

sudo bash -c 'cat << "EOF" > /etc/sysconfig/gemini-log-analyzer
GEMINI_API_KEY="AIzaSyYourGeminiApiKeyHere"
GAS_WEBHOOK_URL="[https://script.google.com/macros/s/YOUR_GAS_DEPLOYMENT_ID/exec](https://script.google.com/macros/s/YOUR_GAS_DEPLOYMENT_ID/exec)"
GAS_SECRET_TOKEN="your-random-secret-token"
WHITELIST_IPS="127.0.0.1,198.51.100.10"
EOF'

sudo chmod 600 /etc/sysconfig/gemini-log-analyzer

※ 198.51.100.10 の部分は、ご自身の管理用固定IPアドレス等に置き換えてください。


6. ログ解析&自動遮断Pythonスクリプトの作成(完成版)

/opt/gemini-log-analyzer/analyze_messages.py を作成します。このスクリプトには以下の工夫を組み込んでいます。

  • 複数ファイルのinode・offset管理: logrotate でログファイルが切り替わっても自動で検知し、先頭から読み直します。また、API通信が成功した後にだけ読み取り位置を更新するため、通信エラー時にログを取りこぼしません。
  • トークン削減フィルタ: Nginxのアクセスログから、画像・CSS・JS等の静的ファイルへの正常アクセス(HTTP 200/304)を除外し、4xx/5xxエラーやPOST通信、wp-login.php 等の重要パスのみを抽出します。さらに、重要な差分が1行もない時間帯はAPI通信自体をスキップします。
  • 思考バジェットの抑制: thinking_budget を制限し、内部推論による出力トークンの過剰消費を抑えます。
  • ランタイム&永続の同時遮断: firewall-cmd --reload による既存通信の瞬断を避けるため、--permanent とランタイムの両方に ipset エントリを直接追加します。
sudo bash -c 'cat << "EOF" > /opt/gemini-log-analyzer/analyze_messages.py
#!/opt/gemini-log-analyzer/venv/bin/python3
import ipaddress
import json
import os
import re
import socket
import subprocess
import sys
import urllib.error
import urllib.request
from datetime import datetime
from pathlib import Path
from google import genai
from google.genai import types

# 監視対象のログファイル一覧
LOG_FILES = [
    Path("/var/log/messages"),
    Path("/home/kusanagi/kusanagi_html/log/nginx/access.log"),
    Path("/home/kusanagi/kusanagi_html/log/nginx/error.log"),
    Path("/home/kusanagi/kusanagi_html/log/nginx/ssl_access.log"),
    Path("/home/kusanagi/kusanagi_html/log/nginx/ssl_error.log"),
]

STATE_FILE = Path("/var/lib/gemini-log-analyzer/state.json")
REPORT_FILE = Path(f"/var/log/gemini-log-analyzer/analysis_{datetime.now().strftime('%Y-%m-%d')}.log")
BLOCK_LOG_FILE = Path("/var/log/gemini-log-analyzer/auto_blocked.log")

MAX_CHARS_PER_FILE = 8000
MAX_TOTAL_CHARS = 25000
MODEL_ID = "gemini-3.8-flash"
IPSET_NAME = "blacklist"

# 静的アセットの拡張子(正常アクセス時は除外)
STATIC_EXT_RE = re.compile(
    r"\.(?:jpe?g|png|gif|webp|svg|ico|css|js|woff2?|ttf|eot|map)(?:\?.*)?\s+HTTP",
    re.IGNORECASE
)
# 重要なパス(これらはHTTP 200でも監視対象に残す)
CRITICAL_PATH_RE = re.compile(
    r"(?:wp-login|xmlrpc|wp-admin|php|\.env|\.git|setup|admin)",
    re.IGNORECASE
)

SYSTEM_PROMPT = """
あなたはLinuxシステム(AlmaLinux)およびKUSANAGI(Nginx / WordPress環境)のセキュリティ・障害解析アシスタントです。
提供されたログ差分を解析し、以下の形式で日本語で簡潔にレポートしてください。

1. 【重要度判定】: 「異常なし」「要確認(Warning)」「緊急(Critical)」のいずれか
2. 【システムログ (/var/log/messages) の主な事象】: 要約(異常がない場合は「異常なし」)
3. 【Web / Nginxログの主な事象】: 5xxエラー、wp-login/xmlrpc攻撃、脆弱性スキャンの要約(異常がない場合は「異常なし」)
4. 【推奨アクション】: 確認すべきコマンドや対処法(異常がない場合は「特になし」)
5. 【自動ブロック対象IP】: 明らかなブルートフォース攻撃、脆弱性スキャン、悪意あるアクセスを行っている送信元IPアドレスのみをカンマ区切りで列挙。該当なしは「なし」と記述
"""


def load_state() -> dict:
    if STATE_FILE.exists():
        try:
            data = json.loads(STATE_FILE.read_text(encoding="utf-8"))
            if isinstance(data, dict):
                return data
        except Exception:
            pass
    return {}


def save_state(state: dict):
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    STATE_FILE.write_text(
        json.dumps(state, ensure_ascii=False, indent=2),
        encoding="utf-8"
    )


def filter_access_log_lines(raw_text: str) -> str:
    """アクセスログから静的アセット正常アクセスを除外し、重要行のみ抽出"""
    filtered_lines = []
    for line in raw_text.splitlines():
        if not line.strip():
            continue

        # 静的ファイルへの正常リクエスト(200 / 304)は除外
        if STATIC_EXT_RE.search(line) and (\x27 " 200 \x27 in line or \x27 " 304 \x27 in line or \x27 200 \x27 in line):
            continue

        # POSTリクエスト、エラー(4xx/5xx)、または攻撃頻出パスを含むものは残す
        is_suspicious_or_error = (
            \x27"POST \x27 in line
            or re.search(r"\s+[45]\d{2}\s+", line)
            or CRITICAL_PATH_RE.search(line)
        )

        if is_suspicious_or_error:
            # User-Agent等が極端に長い場合は末尾を切り詰めてトークン節約
            if len(line) > 300:
                line = line[:300] + "..."
            filtered_lines.append(line)

    return "\n".join(filtered_lines)


def get_all_logs_diff():
    state = load_state()
    new_state = dict(state)
    combined_sections = []

    for log_path in LOG_FILES:
        key = str(log_path)
        if not log_path.exists():
            continue

        stat = log_path.stat()
        current_inode = stat.st_ino
        current_size = stat.st_size

        file_state = state.get(key, {})
        saved_inode = file_state.get("inode")
        saved_offset = file_state.get("offset", 0)

        if saved_inode != current_inode or current_size < saved_offset:
            saved_offset = 0

        if current_size == saved_offset:
            new_state[key] = {"inode": current_inode, "offset": current_size}
            continue

        with log_path.open("r", encoding="utf-8", errors="replace") as f:
            f.seek(saved_offset)
            diff_text = f.read()
            new_offset = f.tell()

        new_state[key] = {"inode": current_inode, "offset": new_offset}

        if not diff_text.strip():
            continue

        # access.log / ssl_access.log はノイズを除去して圧縮
        if "access.log" in log_path.name:
            diff_text = filter_access_log_lines(diff_text)
            if not diff_text.strip():
                continue

        if len(diff_text) > MAX_CHARS_PER_FILE:
            diff_text = (
                f"...(重要ログ多数のため先頭省略。直近 {MAX_CHARS_PER_FILE} 文字抽出)...\n"
                + diff_text[-MAX_CHARS_PER_FILE:]
            )

        combined_sections.append(f"=== [{log_path}] ===\n{diff_text.strip()}")

    full_diff = "\n\n".join(combined_sections)
    if len(full_diff) > MAX_TOTAL_CHARS:
        full_diff = (
            f"...(全体上限超過のため先頭省略。直近 {MAX_TOTAL_CHARS} 文字解析)...\n"
            + full_diff[-MAX_TOTAL_CHARS:]
        )

    return full_diff, new_state


def analyze_with_gemini(log_text: str) -> str:
    client = genai.Client()
    config = types.GenerateContentConfig(
        system_instruction=SYSTEM_PROMPT,
        temperature=0.2,
        thinking_config=types.ThinkingConfig(thinking_budget=1024),
    )
    response = client.models.generate_content(
        model=MODEL_ID,
        contents=f"以下のサーバーログ差分を解析してください:\n\n{log_text}",
        config=config,
    )
    return response.text


def extract_severity(report_text: str) -> str:
    if "緊急" in report_text or "Critical" in report_text:
        return "緊急(Critical)"
    if "要確認" in report_text or "Warning" in report_text:
        return "要確認(Warning)"
    if "異常なし" in report_text:
        return "異常なし"
    m = re.search(r"【重要度判定】[::\s]*(.+)", report_text)
    return m.group(1).strip() if m else "不明"


def get_current_ipset_entries() -> set:
    try:
        res = subprocess.run(
            ["firewall-cmd", "--ipset=" + IPSET_NAME, "--get-entries"],
            capture_output=True, text=True, check=True
        )
        return set(res.stdout.split())
    except Exception:
        return set()


def auto_block_malicious_ips(report_text: str) -> list:
    m = re.search(r"【自動ブロック対象IP】[::\s]*(.+)", report_text)
    if not m:
        return []

    target_text = m.group(1).split("\n")[0]
    if "なし" in target_text or "特になし" in target_text:
        return []

    raw_candidates = re.findall(r"\b(?:\d{1,3}\.){3}\d{1,3}(?:/\d{1,2})?\b", target_text)
    if not raw_candidates:
        return []

    whitelist = {"127.0.0.1", "::1"}
    env_wl = os.environ.get("WHITELIST_IPS", "")
    for ip in env_wl.split(","):
        ip = ip.strip()
        if ip:
            whitelist.add(ip)

    try:
        my_ip = socket.gethostbyname(socket.gethostname())
        whitelist.add(my_ip)
    except Exception:
        pass

    existing_entries = get_current_ipset_entries()
    blocked_ips = []

    for item in set(raw_candidates):
        try:
            net = ipaddress.ip_network(item, strict=False)
            if net.is_private or net.is_loopback or net.is_link_local:
                continue
            if str(net.network_address) in whitelist or item in whitelist:
                continue
            if item in existing_entries:
                continue

            subprocess.run(
                ["firewall-cmd", "--permanent", f"--ipset={IPSET_NAME}", f"--add-entry={item}"],
                check=True, capture_output=True, text=True
            )
            subprocess.run(
                ["firewall-cmd", f"--ipset={IPSET_NAME}", f"--add-entry={item}"],
                check=True, capture_output=True, text=True
            )
            blocked_ips.append(item)
        except Exception:
            pass

    if blocked_ips:
        now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_line = f"[{now_str}] 自動遮断完了: {\x27, \x27.join(blocked_ips)}\n"
        BLOCK_LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
        with BLOCK_LOG_FILE.open("a", encoding="utf-8") as f:
            f.write(log_line)

    return blocked_ips


def send_to_gas(timestamp: str, report_entry: str, raw_result: str):
    webhook_url = os.environ.get("GAS_WEBHOOK_URL", "").strip()
    secret_token = os.environ.get("GAS_SECRET_TOKEN", "").strip()

    if not webhook_url:
        return

    payload = {
        "token": secret_token,
        "timestamp": timestamp,
        "hostname": socket.gethostname(),
        "severity": extract_severity(raw_result),
        "report": report_entry,
    }

    req = urllib.request.Request(
        webhook_url,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )

    try:
        with urllib.request.urlopen(req, timeout=30) as res:
            pass
    except urllib.error.URLError:
        pass


def main():
    diff_text, new_state = get_all_logs_diff()
    if not diff_text.strip():
        print("解析対象となる重要ログなし。API通信をスキップします。")
        save_state(new_state)
        return

    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    result = analyze_with_gemini(diff_text)

    blocked_ips = auto_block_malicious_ips(result)
    block_notice = f"\n【自動遮断実行】: {\x27, \x27.join(blocked_ips) if blocked_ips else \x27遮断対象なし\x27}"

    REPORT_FILE.parent.mkdir(parents=True, exist_ok=True)
    report_entry = (
        f"========================================\n"
        f"解析日時: {timestamp}\n"
        f"========================================\n"
        f"{result}\n{block_notice}\n\n"
    )

    with REPORT_FILE.open("a", encoding="utf-8") as f:
        f.write(report_entry)

    send_to_gas(timestamp, report_entry, result)
    save_state(new_state)
    print(f"解析完了: {REPORT_FILE} に出力しました。")


if __name__ == "__main__":
    main()
EOF'

sudo chmod 700 /opt/gemini-log-analyzer/analyze_messages.py

7. systemdサービスと15分タイマーの設定

Gemini API側が一時的に高負荷(503 UNAVAILABLE)になった場合でも自動的に60秒後に再試行するよう、Restart=on-failure を組み込んだsystemdユニットを作成します。また、ローカルに保存された日次ログが肥大化しないよう、30日経過したファイルを自動削除する処理も入れています。

7-1. サービスファイルの作成

sudo bash -c 'cat &lt;&lt; "EOF" > /etc/systemd/system/gemini-log-analyzer.service
[Unit]
Description=Analyze System and Nginx Logs with Gemini API
After=network-online.target
Wants=network-online.target

[Service]
Type=oneshot
User=root
EnvironmentFile=/etc/sysconfig/gemini-log-analyzer
ExecStart=/opt/gemini-log-analyzer/venv/bin/python3 /opt/gemini-log-analyzer/analyze_messages.py
ExecStartPost=/usr/bin/find /var/log/gemini-log-analyzer/ -name "analysis_*.log" -mtime +30 -delete
Restart=on-failure
RestartSec=60
StartLimitIntervalSec=300
StartLimitBurst=3
StandardOutput=journal
StandardError=journal
EOF'

7-2. タイマーファイルの作成(15分間隔)

sudo bash -c 'cat &lt;&lt; "EOF" > /etc/systemd/system/gemini-log-analyzer.timer
[Unit]
Description=Run Gemini Log Analyzer every 15 minutes

[Timer]
OnCalendar=*:0/15
RandomizedDelaySec=30
Persistent=true

[Install]
WantedBy=timers.target
EOF'

7-3. テスト実行とタイマーの有効化

# 設定の再読み込み
sudo systemctl daemon-reload

# テスト用ログを1行出力して手動実行
logger "Gemini Log Analyzer test execution"
sudo systemctl start gemini-log-analyzer.service

# 実行結果とステータスの確認
sudo journalctl -u gemini-log-analyzer.service -n 20 --no-pager

# 問題なければタイマーを有効化
sudo systemctl enable --now gemini-log-analyzer.timer
sudo systemctl list-timers gemini-log-analyzer.timer

8. 運用時のトラブルシューティングとコスト管理の知見

実際に運用する中で遭遇しやすいエラーとその対策をまとめました。

① 403 PERMISSION_DENIED(IPアドレスブロック)が出る場合

Google CloudコンソールでAPIキーにIPv4アドレスしか登録していない場合、サーバーがIPv6で通信を行ってブロックされるケースがあります。エラーメッセージ内に表示される送信元IPv6アドレスを確認し、Google Cloudの「認証情報」の許可IPリストへ追加してください。

② 429 RESOURCE_EXHAUSTED(1日の無料枠リクエスト上限超過)が出る場合

最新世代のモデルでは、無料枠(Free Tier)の1日あたりリクエスト上限(RPD)が20回程度に制限されている場合があります。15分間隔(1日最大96回)で常時稼働させる場合は、以下の2つを組み合わせるのが効果的です。

  • スクリプト側のフィルタリング(実装済み): 平時の正常アクセスを除外し、エラーや不審アクセスがない時間帯はAPIを呼ばないようにする。
  • Paid Tier(従量課金)への移行: Google Cloudコンソールで請求先アカウントをリンクするとRPD制限が解除されます。なお、Google Developer Program特典などのGCPクレジットがある場合、有効な支払い方法が設定されていればクレジット残高から優先的に自動消費されます。

③ 入力・出力トークン数を抑えて月額数百円以内に収めるコツ

Webサーバーの access.log を無加工で送り続けると、画像やCSSの読み込みログだけで1日あたり約60万トークン以上を消費してしまうことがあります。

本記事のスクリプトのように 「静的アセット(200/304)の除外」「長いUser-Agentの切り詰め」「thinking_budget の制限」 を入れることで、監視精度を維持したままトークン消費量を 約70%〜90%削減 でき、Paid Tierでも月額数百円程度の低コストで安全に運用できます。

④ 誤って遮断されたIPを解除したい場合

万が一、特定のIPアドレスをブロック解除したい場合は以下のコマンドで即座に除外できます。

# 現在ブロックされているIP一覧の確認
sudo firewall-cmd --ipset=blacklist --get-entries

# 特定IPのブロック解除
sudo firewall-cmd --permanent --ipset=blacklist --remove-entry="203.0.113.50"
sudo firewall-cmd --ipset=blacklist --remove-entry="203.0.113.50"

まとめ

Gemini APIと systemd、firewalld(ipset)、そしてGoogle Apps Scriptを組み合わせることで、単なる「ログの要約」にとどまらず、不審な通信の検知からファイアウォールでの自動遮断、クラウドへの日次レポート保管までを一気通貫で自動化できました。

サーバーの自己防衛力を高めつつ、日々のログ監視業務を大幅に効率化できるので、ぜひ導入してみてください。

Leave a Reply

Your email address will not be published. Required fields are marked *

CAPTCHA