logwatch-ai.py

#!/usr/bin/env python3
"""
Logwatch AI Analyzer
Analyzes logwatch output using OpenAI API and sends alerts only when issues are detected
"""

import os
import re
import json
import subprocess
import logging
import smtplib
import time
import fcntl
import socket
import secrets
import tempfile
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional, Dict, Any
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

try:
    from openai import OpenAI
except ImportError:
    print("Error: OpenAI library not installed. Run: pip install openai")
    exit(1)

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/var/log/logwatch-ai.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# --- Deterministic severity-ceiling constants -------------------------------
# LLM-reported severity is advisory only; these patterns match the *raw*
# logwatch text (never LLM-produced fields, which can be wrong/hallucinated -
# e.g. 2026-08-24's report claimed disk_usage_percent=0 while `df` showed 58%).
SENSITIVE_HTTP_PATH_RE = re.compile(
    r'(wp-admin|wp-login|\.env|phpinfo|eval-stdin|filemanager|/admin|credentials|wp-config|\.git)',
    re.IGNORECASE
)
HTTP_STATUS_HEADER_RE = re.compile(r'^\s*(\d{3})\s+[A-Za-z]')

CRASH_SIGNAL_PATTERNS = [
    re.compile(r'[Oo]ut of memory', re.IGNORECASE),
    re.compile(r'oom[-_]?kill', re.IGNORECASE),
    re.compile(r'[Kk]ernel panic'),
    re.compile(r'segfault'),
    re.compile(r'EXT4-fs error'),
    re.compile(r'[Cc]orrupt(ed)? filesystem'),
    re.compile(r'I/O error'),
    re.compile(r'InnoDB:.*[Cc]orrupt'),
]

DISK_CEILING_THRESHOLD_PERCENT = 85

# --- HTTP-probe live-verification constants ---------------------------------
# See _verify_http_probe_hits(): live-checks a "successful probe" hit against
# the actual vhosts before trusting it, to filter out SPA/catch-all routes
# that return 200 for literally any path (2026-08-29: /.htpasswd flagged
# high, actual cause was uptime.gdev.fun's catch-all index.html, not a leak).
HTTP_PROBE_MAX_PATHS = 10  # hits beyond this many are left unverified -> confirmed
HTTP_PROBE_MAX_VHOSTS = 15
HTTP_PROBE_TIME_BUDGET_SECONDS = 60.0
HTPASSWD_CONTENT_MARKERS = (
    b'$apr1$', b'$2y$', b'$1$', b':{SHA}', b':{SSHA}', b':{PLAIN}', b':{CRYPT}'
)

class LogwatchAIAnalyzer:
    """Analyzes logwatch output using AI and sends notifications"""

    def __init__(self, config_path: str = "/etc/logwatch-ai/config.json"):
        """Initialize with configuration"""
        self.config = self.load_config(config_path)
        self.client = OpenAI(api_key=self.config['openai_api_key'])
        self.rate_limit_file = Path('/var/log/logwatch-ai-ratelimit.json')
        self.lock_file = Path('/var/lock/logwatch-ai.lock')
        # Populated by _has_sensitive_http_success() each run; surfaced in the
        # email via format_email_body() regardless of whether the severity
        # ceiling ended up applying.
        self._probe_verification_notes = []

    def load_config(self, config_path: str) -> Dict[str, Any]:
        """Load configuration from JSON file"""
        config_file = Path(config_path)

        # Default configuration
        default_config = {
            "openai_api_key": os.getenv("OPENAI_API_KEY", ""),
            "openai_model": "gpt-4o-mini",
            "smtp_host": "localhost",
            "smtp_port": 25,
            "smtp_user": "",
            "smtp_password": "",
            "smtp_use_tls": False,
            "from_email": "logwatch-ai@localhost",
            "to_emails": ["root@localhost"],
            "alert_threshold": "medium",
            "logwatch_output_file": "/var/log/logwatch_output.txt",
            "always_send_summary": False,
            "verify_http_probes": True,
            "known_local_users": [],
            "trusted_ownership_paths": [],
            "max_requests_per_hour": 10,
            "max_requests_per_day": 50,
            "min_interval_minutes": 5,
            "max_retries": 3,
            "retry_delay_seconds": 30
        }

        if config_file.exists():
            try:
                with open(config_file, 'r') as f:
                    user_config = json.load(f)
                    default_config.update(user_config)
            except Exception as e:
                logger.warning(f"Failed to load config from {config_path}: {e}")

        return default_config

    def get_disk_usage(self) -> str:
        """Get actual disk usage information using df command"""
        try:
            result = subprocess.run(
                ['df', '-h'],
                capture_output=True,
                text=True,
                timeout=10
            )
            if result.returncode == 0:
                return result.stdout
            return ""
        except Exception as e:
            logger.warning(f"Failed to get disk usage: {e}")
            return ""

    def run_logwatch(self) -> str:
        """Execute logwatch and capture output"""
        try:
            result = subprocess.run(
                ['/usr/sbin/logwatch', '--output', 'stdout', '--format', 'text',
                 '--range', 'yesterday', '--detail', '10'],
                capture_output=True,
                text=True,
                timeout=60
            )

            if result.returncode != 0:
                logger.error(f"Logwatch failed with code {result.returncode}: {result.stderr}")
                return ""

            # Save raw output for debugging
            output_file = Path(self.config['logwatch_output_file'])
            output_file.parent.mkdir(parents=True, exist_ok=True)
            output_file.write_text(result.stdout)

            return result.stdout

        except Exception as e:
            logger.error(f"Failed to run logwatch: {e}")
            return ""

    def check_rate_limits(self) -> bool:
        """Check if we're within rate limits to prevent API abuse"""
        now = datetime.now()

        # Load existing rate limit data
        rate_data = {"requests": []}
        if self.rate_limit_file.exists():
            try:
                with open(self.rate_limit_file, 'r') as f:
                    rate_data = json.load(f)
            except Exception as e:
                logger.warning(f"Failed to load rate limit data: {e}")

        # Clean up old entries (older than 24 hours)
        cutoff_time = (now - timedelta(days=1)).isoformat()
        rate_data["requests"] = [
            req for req in rate_data["requests"]
            if req > cutoff_time
        ]

        # Check minimum interval since last request
        if rate_data["requests"]:
            last_request = datetime.fromisoformat(rate_data["requests"][-1])
            time_since_last = (now - last_request).total_seconds() / 60

            if time_since_last < self.config["min_interval_minutes"]:
                remaining = self.config["min_interval_minutes"] - time_since_last
                logger.warning(f"Rate limit: minimum interval not met. Wait {remaining:.1f} more minutes.")
                return False

        # Check hourly limit
        hour_ago = (now - timedelta(hours=1)).isoformat()
        hour_requests = sum(1 for req in rate_data["requests"] if req > hour_ago)

        if hour_requests >= self.config["max_requests_per_hour"]:
            logger.warning(f"Rate limit: hourly limit ({self.config['max_requests_per_hour']}) reached")
            return False

        # Check daily limit
        day_requests = len(rate_data["requests"])

        if day_requests >= self.config["max_requests_per_day"]:
            logger.warning(f"Rate limit: daily limit ({self.config['max_requests_per_day']}) reached")
            return False

        # Add current request to rate limit data
        rate_data["requests"].append(now.isoformat())

        # Save updated rate limit data
        try:
            self.rate_limit_file.parent.mkdir(parents=True, exist_ok=True)
            with open(self.rate_limit_file, 'w') as f:
                json.dump(rate_data, f)
        except Exception as e:
            logger.error(f"Failed to save rate limit data: {e}")

        return True

    def acquire_lock(self) -> Optional[Any]:
        """Acquire a file lock to prevent concurrent runs"""
        try:
            self.lock_file.parent.mkdir(parents=True, exist_ok=True)
            lock_fd = open(self.lock_file, 'w')
            fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            return lock_fd
        except (IOError, OSError):
            logger.error("Another instance is already running. Exiting to prevent duplicate API calls.")
            return None

    def release_lock(self, lock_fd):
        """Release the file lock"""
        if lock_fd:
            try:
                fcntl.flock(lock_fd, fcntl.LOCK_UN)
                lock_fd.close()
            except Exception as e:
                logger.warning(f"Failed to release lock: {e}")

    def _prompt_known_users_clause(self) -> str:
        """AI-prompt ignore-list line for routine sudo by configured known local
        users. Empty (line omitted) when known_local_users is empty, so the
        prompt never names users that may not exist on this host.
        """
        users = self.config.get('known_local_users') or []
        if not users:
            return ''
        return f"- 正規ユーザー（{'、'.join(users)}）による通常のsudo操作\n"

    def _prompt_trusted_ownership_clause(self) -> str:
        """AI-prompt ignore-list line for www-data ownership/permission changes
        under configured trusted paths. Empty (line omitted) when
        trusted_ownership_paths is empty.
        """
        paths = self.config.get('trusted_ownership_paths') or []
        if not paths:
            return ''
        return f"- www-dataへの所有権・権限変更（{'、'.join(paths)}等）\n"

    def _prompt_known_users_exclusion_clause(self) -> str:
        """AI-prompt parenthetical excluding configured known local users from
        the "unexpected root/admin login" critical-issue rule. Empty (no
        parenthetical) when known_local_users is empty, since there is
        nothing concrete to name as excluded - every login stays reportable.
        """
        users = self.config.get('known_local_users') or []
        if not users:
            return ''
        return f"（正規ユーザー{'/'.join(users)}は除外）"

    def analyze_with_ai(self, log_content: str) -> Dict[str, Any]:
        """Analyze log content using OpenAI API with rate limiting and retries"""

        if not log_content:
            return {
                "severity": "error",
                "issues_found": True,
                "summary": "分析するlogwatch出力がありません",
                "details": [],
                "recommendations": []
            }

        # Check rate limits before making API call
        if not self.check_rate_limits():
            return {
                "severity": "error",
                "issues_found": True,
                "summary": "レート制限超過 - API過剰利用防止のためスキップしました",
                "critical_issues": ["レート制限保護が作動しました"],
                "warnings": [],
                "statistics": {},
                "recommendations": ["次回実行まで待つか、設定でレート制限を調整してください"]
            }

        prompt = f"""あなたはLinuxシステムセキュリティの専門家です。以下のlogwatch出力を分析し、構造化された評価を日本語で提供してください。

【最重要】本当に対応が必要な問題だけを報告してください。インターネット公開サーバーで日常的に発生する事象は全て無視してください。

以下は【完全に無視】してください（critical_issuesやwarningsに含めない）：
- 失敗したSSHログイン試行（ブロック済みの攻撃）
- SSH接続エラー（"banner exchange"、"invalid format"、"Bad protocol"、"Connection closed"等）
  → これらは失敗した攻撃であり、ログイン成功ではない。絶対に「未知のIPからのログイン成功」として報告しないこと
- 404/400/401エラーを返したHTTPリクエスト（スキャンボットは日常的）
- /.env、/.git/config、/phpMyAdmin等への脆弱性スキャン（全て失敗している）
- "Attempts to use known hacks"の報告（攻撃試行は失敗している）
- mod_proxyへの接続試行
- fail2banによるブロック
- ディスク使用率85%未満
- 通常のサービス再起動
- 定期的なcronジョブ実行（chown、chmod、tar、バックアップスクリプトなど）
- パッケージの更新・インストール
- 通常のメール送受信
{self._prompt_known_users_clause()}{self._prompt_trusted_ownership_clause()}- cronによるrootユーザーでのメンテナンスタスク
- SECURITYメール送信（sudoコマンド実行の通知）

以下の【本当に重大な問題のみ】をcritical_issuesに含めてください：
- 認証成功後の不審なアクティビティ（ただし、cron/sudo経由の定期タスクは除外）
  * 未知のIPアドレスからのログイン成功
    【重要】「ログイン成功」の判定基準：
    - 成功: "Accepted publickey"、"Accepted password"、"session opened" のみ
    - 失敗（無視すべき）: "banner exchange"、"invalid format"、"Failed password"、
      "Invalid user"、"Connection closed"、"Disconnected"、"Bad protocol"
    → 失敗した接続試行は攻撃者がブロックされた証拠であり、問題ではない
  * 通常と異なる時間帯での予期しないログイン
  * 設定ファイルやシステムファイルへの不正な変更（cronタスク以外）
- rootや管理者での予期しないログイン成功{self._prompt_known_users_exclusion_clause()}
  【重要】成功したログインのみを報告。接続試行の失敗やエラーは無視すること。
- ディスク使用率85%超過
- サービスの異常停止・クラッシュ（再起動ではなく停止）
- カーネルパニックやOOMキラー発動
- データベースの破損やクラッシュ
- ファイルシステムエラー

severity判定基準：
- "none": 問題なし（日常的なスキャンのみ）
- "low": 軽微な注意事項のみ
- "medium": 確認が必要だが緊急ではない
- "high": 24時間以内の対応が必要。以下のいずれかの【具体的な成功・実害の証拠】がある場合のみ使用すること：
  未知ユーザーでの認証成功、機微パスへの200/301/302応答、ディスク使用率85%超過、サービスのクラッシュ/OOM/カーネルパニック
- "critical": 即時対応が必要（上記の実害が広範囲・進行中の場合）
【重要】ブロック済みの探査・fail2banの重複ブロック・403/404のみのスキャンは、それがどれだけ大量でも上記の「具体的な成功・実害の証拠」がない限り"high"にしないこと（"low"または"medium"にとどめる）

JSON形式で日本語で回答してください：
{{
    "severity": "none|low|medium|high|critical",
    "issues_found": true|false,
    "summary": "簡潔な一行サマリー",
    "critical_issues": ["問題1", "問題2"],
    "warnings": ["警告1", "警告2"],
    "statistics": {{
        "ssh_attempts": 数値,
        "blocked_ips": 数値,
        "disk_usage_percent": 数値,
        "errors_count": 数値
    }},
    "recommendations": ["推奨アクション1", "推奨アクション2"],
    "log_excerpts": ["重要な問題がある場合のみ、関連するログの抜粋を含める（severity が medium 以上の場合）"]
}}

Logwatch出力:
{log_content[:8000]}"""  # Limit to avoid token limits

        # Retry logic with exponential backoff
        last_error = None
        for attempt in range(self.config['max_retries']):
            try:
                response = self.client.chat.completions.create(
                    model=self.config['openai_model'],
                    messages=[
                        {"role": "system", "content": "あなたはLinuxセキュリティの専門家です。全ての回答を必ず日本語で出力してください。英語は絶対に使用しないでください。summary、critical_issues、warnings、recommendationsの全てを日本語で記述してください。"},
                        {"role": "user", "content": prompt}
                    ],
                    # temperature=0,  # Removed - not supported by gpt-4o-mini
                    max_completion_tokens=1000,  # Changed from max_tokens to max_completion_tokens
                    response_format={"type": "json_object"},
                    timeout=30  # 30 second timeout per request
                )

                result = json.loads(response.choices[0].message.content)
                logger.info(f"AI Analysis complete. Severity: {result.get('severity', 'unknown')}")
                return result

            except Exception as e:
                last_error = e
                logger.warning(f"API call attempt {attempt + 1}/{self.config['max_retries']} failed: {e}")

                if attempt < self.config['max_retries'] - 1:
                    delay = self.config['retry_delay_seconds'] * (2 ** attempt)  # Exponential backoff
                    logger.info(f"Retrying in {delay} seconds...")
                    time.sleep(delay)

        # All retries failed
        logger.error(f"All API retry attempts failed. Last error: {last_error}")
        return {
            "severity": "error",
            "issues_found": True,
            "summary": f"AI analysis failed after {self.config['max_retries']} attempts: {str(last_error)}",
            "critical_issues": ["Failed to analyze logs with AI after multiple retries"],
            "warnings": [],
            "statistics": {},
            "recommendations": ["Check OpenAI API key, connectivity, and rate limits"]
        }

    def review_analysis(self, analysis: Dict[str, Any], log_content: str) -> Dict[str, Any]:
        """Second-pass AI review: verify every concrete claim against the raw log verbatim.

        Guards against hallucinated critical_issues/log_excerpts (e.g. a fabricated IP or
        file path that never appears in the source log) surviving to the alert email.
        """

        review_prompt = f"""あなたはLinuxシステムセキュリティのレビュー担当です。以下は別のAIがlogwatch出力を分析した結果(JSON)です。

【最重要】この分析結果の critical_issues・warnings・log_excerpts に書かれた具体的な記述(IPアドレス、ユーザー名、ファイルパス、ログの引用文など)が、下記の「元のlogwatch出力」に一字一句(逐語で)実在するか、または元ログの記述から直接読み取れる事実かを検証してください。

判定基準:
- IPアドレスは元ログ中の該当行に実在する形式・値であること。"xyz.xxx.xxx.xxx"のようなプレースホルダー形式や、元ログに存在しないIPは即座に不合格とすること
- ファイルパス・ユーザー名・引用文も元ログの記述と完全一致すること。似ているが完全一致しない場合(例:別々の行にある内容を混ぜて作った文)は不合格
- 元ログに存在しない、または裏付けが取れない項目は、分析結果から削除すること
- 全てのcritical_issues/warningsが削除された場合は severity を "none" に、issues_found を false にすること
- statistics(ssh_attempts等)も元ログの記述と矛盾する場合は元ログに基づいて修正すること

出力は元の分析結果と同じJSON形式に加えて、"review_notes"(検証で削除・修正した項目とその理由を日本語で簡潔に。削除がなければ空文字)を含めてください：
{{
    "severity": "none|low|medium|high|critical",
    "issues_found": true|false,
    "summary": "簡潔な一行サマリー",
    "critical_issues": ["問題1", "問題2"],
    "warnings": ["警告1", "警告2"],
    "statistics": {{
        "ssh_attempts": 数値,
        "blocked_ips": 数値,
        "disk_usage_percent": 数値,
        "errors_count": 数値
    }},
    "recommendations": ["推奨アクション1", "推奨アクション2"],
    "log_excerpts": ["元ログから逐語で確認できたものだけ"],
    "review_notes": "削除・修正した項目とその理由(無ければ空文字)"
}}

【検証対象の分析結果】
{json.dumps(analysis, ensure_ascii=False)}

【元のlogwatch出力】
{log_content[:8000]}"""

        last_error = None
        for attempt in range(self.config['max_retries']):
            try:
                response = self.client.chat.completions.create(
                    model=self.config['openai_model'],
                    messages=[
                        {"role": "system", "content": "あなたはLinuxセキュリティのレビュー担当です。元ログに実在しない記述は必ず除去してください。全ての回答を必ず日本語で出力してください。英語は使用しないでください。"},
                        {"role": "user", "content": review_prompt}
                    ],
                    max_completion_tokens=2000,
                    response_format={"type": "json_object"},
                    timeout=30
                )

                reviewed = json.loads(response.choices[0].message.content)
                reviewed['review_failed'] = False
                logger.info(
                    f"AI Review complete. Severity: {analysis.get('severity', 'unknown')} -> "
                    f"{reviewed.get('severity', 'unknown')}"
                )
                return reviewed

            except Exception as e:
                last_error = e
                logger.warning(f"Review API call attempt {attempt + 1}/{self.config['max_retries']} failed: {e}")

                if attempt < self.config['max_retries'] - 1:
                    delay = self.config['retry_delay_seconds'] * (2 ** attempt)
                    logger.info(f"Retrying review in {delay} seconds...")
                    time.sleep(delay)

        # Review failed after all retries - fail open with the unreviewed analysis,
        # but flag it so the email/log make clear the hallucination check did not run.
        logger.error(f"All review retry attempts failed. Using unreviewed analysis. Last error: {last_error}")
        analysis['review_failed'] = True
        analysis['review_notes'] = f"レビューAPI呼び出しに失敗したため未検証です: {last_error}"
        return analysis

    @staticmethod
    def _extract_section(log_content: str, begin_marker: str, end_marker: str) -> str:
        """Slice out one logwatch section (e.g. between 'httpd Begin' and 'httpd End')"""
        start = log_content.find(begin_marker)
        if start == -1:
            return ''
        end = log_content.find(end_marker, start)
        return log_content[start:end] if end != -1 else log_content[start:]

    @staticmethod
    def _max_disk_usage_percent(disk_info: str) -> float:
        """Parse our own `df -h` output (not the LLM's statistics field) for the worst Use%"""
        max_pct = 0.0
        for line in disk_info.splitlines():
            parts = line.split()
            if len(parts) >= 5 and parts[0].startswith('/dev/'):
                try:
                    max_pct = max(max_pct, float(parts[4].rstrip('%')))
                except ValueError:
                    continue
        return max_pct

    def _has_verified_ssh_success(self, log_content: str) -> (bool, str):
        """Detect a real login by anyone other than the configured known local users.

        Matches logwatch's own aggregated formats:
          pam_unix: "alice(uid=1002) by alice: 16 Time(s)"
          SSHD:     "Users logging in through sshd:\n    alice:\n       1.2.3.4: 1 Time"
        (NOT raw sshd syslog "Accepted publickey for ..." - logwatch never emits that.)

        config['known_local_users'] defaults to empty, i.e. every login is
        treated as unknown (safe default) until the host's real local users
        are configured.
        """
        reasons = []
        known_users = set(self.config.get('known_local_users') or [])

        pam_section = self._extract_section(
            log_content, '--------------------- pam_unix Begin', '---------------------- pam_unix End')
        for m in re.finditer(r'^\s*(\S+)\(uid=\d+\)\s+by\s+\S+:', pam_section, re.MULTILINE):
            user = m.group(1)
            if user not in known_users:
                reasons.append(f"pam_unixで未知ユーザー'{user}'のセッション開始を検出")

        sshd_section = self._extract_section(
            log_content, '--------------------- SSHD Begin', '---------------------- SSHD End')
        in_login_block = False
        for line in sshd_section.splitlines():
            if 'Users logging in through sshd:' in line:
                in_login_block = True
                continue
            if in_login_block:
                m = re.match(r'^\s{4}(\S+):\s*$', line)
                if m and m.group(1) not in known_users:
                    reasons.append(f"SSHDで未知ユーザー'{m.group(1)}'のログイン成功を検出")

        return (len(reasons) > 0, '; '.join(reasons))

    def _has_sensitive_http_success(self, log_content: str) -> (bool, str):
        """Detect a genuine successful hit, in either of logwatch's two httpd formats:

        (A) "A total of N possible successful probes were detected ... (URLs matching a
            known exploit pattern that got a non-error response)", listed inline as
            "<path> HTTP Response 200" - this is logwatch's OWN success determination,
            already restricted to exploit-pattern paths, so any 2xx/301/302 here counts
            regardless of path text (real example: 2026-08-21's day3 report, 7 hits
            including /etc/passwd and .aws/credentials - neither contains our keyword
            list below, which is why format (A) cannot be reduced to format (B)).
        (B) paths itemized under an explicit status-code header ("200 OK" etc, mirroring
            how 401/403/404 are itemized in normal operation) - kept as a defensive
            fallback in case logwatch's format changes; filtered by sensitive keywords
            since this format is not pre-filtered to exploit patterns.

        Raw hits collected above are advisory only: a "200" here can be a SPA/catch-all
        vhost returning its index.html for literally any path (2026-08-29: /.htpasswd
        flagged high; the actual cause was uptime.gdev.fun's catch-all route, not a real
        leak). Unless config['verify_http_probes'] is disabled, hits are run through
        _verify_http_probe_hits(), which live-probes the local vhosts to tell a real hit
        apart from a catch-all response. Verification fails closed on any error/timeout/
        budget overrun - it can only remove a hit from the confirmed set when it has
        positive evidence of a catch-all, never by default.
        """
        httpd_section = self._extract_section(log_content, '--------------------- httpd Begin', '---------------------- httpd End')
        hits = []  # [{'raw': str, 'status': str}]

        for m in re.finditer(r'^(.*\bHTTP Response ((?:2\d\d|301|302))\s*)$', httpd_section, re.MULTILINE):
            hits.append({'raw': m.group(1).strip(), 'status': m.group(2)})

        current_code = None
        for line in httpd_section.splitlines():
            header = HTTP_STATUS_HEADER_RE.match(line)
            if header:
                current_code = header.group(1)
                continue
            if current_code and (current_code.startswith('2') or current_code in ('301', '302')):
                if SENSITIVE_HTTP_PATH_RE.search(line):
                    hits.append({'raw': line.strip(), 'status': current_code})

        self._probe_verification_notes = []

        if not hits:
            return (False, '')

        if self.config.get('verify_http_probes', True):
            try:
                confirmed, fp_notes = self._verify_http_probe_hits(hits)
            except Exception as e:
                # Verification itself must never be the reason a real hit gets buried.
                logger.warning(
                    f"HTTP probe verification raised an unexpected exception; "
                    f"treating all hits as confirmed (fail-closed): {e}"
                )
                confirmed = [h['raw'] for h in hits]
                fp_notes = [f"検証処理で予期しない例外が発生したため、全ヒットを確定扱いとしました: {e}"]
        else:
            confirmed = [h['raw'] for h in hits]
            fp_notes = []

        self._probe_verification_notes = fp_notes
        return (len(confirmed) > 0, '; '.join(confirmed[:5]))

    def _verify_http_probe_hits(self, hits: list) -> (list, list):
        """Live-verify raw HTTP-success hits against the real local vhosts.

        hits: [{'raw': str, 'status': str}, ...] as collected by
        _has_sensitive_http_success().

        Returns (confirmed_raw_hits, fp_notes) where confirmed_raw_hits is the
        subset of hits['raw'] that could NOT be explained away as a catch-all
        response, and fp_notes is a list of human-readable Japanese notes
        explaining every hit's disposition (both filtered and confirmed), for
        display in the alert email.
        """
        confirmed = []
        fp_notes = []
        to_verify = []  # [(raw, path, status)]

        for hit in hits:
            raw = hit['raw']
            status = hit['status']

            # Self-probe exclusion: a hit line containing our own verification
            # marker is this script's own prior probe, echoed back by logwatch.
            # Never treat it as evidence of anything, and never re-probe it -
            # doing so would create an infinite self-reinforcing loop.
            if 'lwai-verify' in raw:
                fp_notes.append(f"'{raw}' は検証プローブ自身が過去に残したアクセスのため除外しました")
                continue

            if ' HTTP Response ' in raw:
                path = raw.split(' HTTP Response ', 1)[0].strip()
            else:
                tokens = raw.split()
                path = tokens[0].rstrip(':') if tokens else ''

            if not path.startswith('/'):
                # Couldn't parse a path out of this line - err toward reporting it.
                confirmed.append(raw)
                fp_notes.append(f"'{raw}' はパスを特定できなかったため確定扱いとしました")
                continue

            to_verify.append((raw, path, status))

        if not to_verify:
            return confirmed, fp_notes

        if len(to_verify) > HTTP_PROBE_MAX_PATHS:
            overflow, to_verify = to_verify[HTTP_PROBE_MAX_PATHS:], to_verify[:HTTP_PROBE_MAX_PATHS]
            confirmed.extend(raw for raw, _, _ in overflow)
            fp_notes.append(
                f"検証対象パスが{HTTP_PROBE_MAX_PATHS}件を超えたため、"
                f"超過分{len(overflow)}件は無検証のまま確定扱いとしました"
            )

        vhosts = self._get_nginx_vhosts()
        if not vhosts:
            # Enumeration failure is indistinguishable from "no vhosts" only in
            # effect, never in meaning: fail closed and confirm everything.
            confirmed.extend(raw for raw, _, _ in to_verify)
            fp_notes.append(
                "nginx vhost列挙に失敗したため、ライブ検証をスキップし全ヒットを確定扱いとしました(fail-closed)"
            )
            return confirmed, fp_notes

        if len(vhosts) > HTTP_PROBE_MAX_VHOSTS:
            vhosts = vhosts[:HTTP_PROBE_MAX_VHOSTS]

        deadline = time.monotonic() + HTTP_PROBE_TIME_BUDGET_SECONDS

        # Phase 1: exactly one canary probe per vhost, done once up front and
        # shared across every hit path below - this is what keeps the number
        # of brand-new "exploit-pattern" log lines we create to a minimum.
        canary_results = {}
        for vhost in vhosts:
            if time.monotonic() > deadline:
                break
            canary_path = f'/lwai-verify-canary-{secrets.token_hex(8)}?lwai-verify=1'
            canary_results[vhost] = self._fetch_probe_response(vhost, canary_path)

        for raw, path, status in to_verify:
            if time.monotonic() > deadline:
                confirmed.append(raw)
                fp_notes.append(f"'{raw}' は検証の時間予算(60秒)超過のため未検証のまま確定扱いとしました")
                continue

            try:
                is_fp, note = self._verify_single_hit(raw, path, status, canary_results, deadline)
            except Exception as e:
                is_fp, note = False, f"'{raw}' は検証中に予期しない例外が発生したため確定扱いとしました: {e}"

            if is_fp:
                fp_notes.append(note)
            else:
                confirmed.append(raw)
                fp_notes.append(note)

        return confirmed, fp_notes

    def _verify_single_hit(self, raw: str, path: str, status_str: str,
                            canary_results: Dict[str, Optional[Dict[str, Any]]],
                            deadline: float) -> (bool, str):
        """Decide whether one hit (path, status) is a catch-all false positive.

        Only probes vhosts whose canary response already matched the hit's
        status class (2xx, or 301/302) - i.e. vhosts that are themselves
        catch-all candidates. If none exist, or none of them explains the hit
        when actually probed, the hit is confirmed (fail-closed): a past 200
        that nothing currently reproduces is not something we can wave away.
        """
        is_redirect = status_str in ('301', '302')

        candidates = [
            vhost for vhost, canary in canary_results.items()
            if canary is not None and (
                canary['status'] in (301, 302) if is_redirect
                else 200 <= canary['status'] < 300
            )
        ]

        if not candidates:
            return False, (
                f"'{raw}' はcatch-all候補vhostが無かった"
                f"(どのvhostもcanaryへ2xx/3xxを返さなかった)ため確定扱いとしました"
            )

        query_path = path + ('&lwai-verify=1' if '?' in path else '?lwai-verify=1')

        any_matched = False
        explain_vhosts = []
        for vhost in candidates:
            if time.monotonic() > deadline:
                return False, f"'{raw}' は検証の時間予算超過のため確定扱いとしました"

            resp = self._fetch_probe_response(vhost, query_path)
            if resp is None:
                continue

            matches_class = (resp['status'] in (301, 302)) if is_redirect else (200 <= resp['status'] < 300)
            if not matches_class:
                continue

            any_matched = True

            # Real-leak guard takes priority over any catch-all match: never
            # let a body-diff heuristic wave away actual credential content.
            if self._looks_like_real_leak(resp):
                return False, (
                    f"'{raw}' は{vhost}で機微データらしき応答内容"
                    f"(HTML以外またはhtpasswd様の内容)が確認されたため確定扱いとしました"
                )

            canary = canary_results[vhost]
            is_catch_all = (
                self._redirect_matches_catch_all(canary, resp) if is_redirect
                else self._body_matches_catch_all(canary, resp)
            )
            if not is_catch_all:
                return False, (
                    f"'{raw}' は{vhost}でcanaryと異なる応答内容が確認された"
                    f"(catch-allでは説明できない)ため確定扱いとしました"
                )

            explain_vhosts.append(vhost)

        if not any_matched:
            return False, f"'{raw}' はcatch-all候補vhostのいずれも本番同様の応答を返さなかったため確定扱いとしました"

        vhost_list = '、'.join(explain_vhosts)
        return True, (
            f"'{raw}' は{vhost_list}のSPA catch-all応答(canary比較一致)のため誤検知と判定しました"
        )

    @staticmethod
    def _looks_like_real_leak(resp: Dict[str, Any]) -> bool:
        """Independent safety net: even a body-diff catch-all match is overridden
        if the response isn't HTML, or contains htpasswd-style credential hashes.
        """
        body = resp.get('body') or b''
        if not body.lstrip().startswith(b'<'):
            return True
        return any(marker in body for marker in HTPASSWD_CONTENT_MARKERS)

    @staticmethod
    def _body_matches_catch_all(canary: Optional[Dict[str, Any]], resp: Optional[Dict[str, Any]]) -> bool:
        """True if a 2xx response to the sensitive path looks like the same
        catch-all page the canary (guaranteed-nonexistent) path got: either an
        exact match on the first 512 bytes, or both HTML documents of nearly
        (+/-10%) the same size.
        """
        if not canary or not resp or canary['status'] != resp['status']:
            return False

        cbody, rbody = canary.get('body') or b'', resp.get('body') or b''
        if cbody[:512] == rbody[:512]:
            return True

        def _is_html_start(b: bytes) -> bool:
            head = b.lstrip()[:15].lower()
            return head.startswith(b'<!doctype html') or head.startswith(b'<html')

        if _is_html_start(cbody) and _is_html_start(rbody):
            c_ct = (canary.get('content_type') or '').lower()
            r_ct = (resp.get('content_type') or '').lower()
            if 'text/html' in c_ct and 'text/html' in r_ct and cbody:
                if abs(len(cbody) - len(rbody)) / len(cbody) <= 0.10:
                    return True

        return False

    @staticmethod
    def _redirect_matches_catch_all(canary: Optional[Dict[str, Any]], resp: Optional[Dict[str, Any]]) -> bool:
        """True if a 3xx response to the sensitive path redirects to the same
        origin (scheme+host[:port]) as the canary's redirect - i.e. a blanket
        redirect rule (https-upgrade, canonical-host, etc), not path-specific.
        """
        if not canary or not resp or canary['status'] != resp['status']:
            return False

        def _origin(url: str) -> str:
            m = re.match(r'^[a-zA-Z][a-zA-Z0-9+.-]*://[^/]+', url)
            return m.group(0) if m else url

        return _origin(canary.get('location', '')) == _origin(resp.get('location', ''))

    def _get_nginx_vhosts(self) -> Optional[list]:
        """Enumerate configured nginx server_name values.

        Returns None (never an empty list) when enumeration itself failed, so
        callers can tell "no vhosts configured" (impossible on this host, but
        would mean nothing to verify against) apart from "couldn't find out"
        (must fail closed and skip verification entirely).
        """
        vhosts = []
        try:
            result = subprocess.run(['nginx', '-T'], capture_output=True, text=True, timeout=10)
            if result.returncode == 0 and result.stdout:
                vhosts = self._parse_server_names(result.stdout)
        except Exception as e:
            logger.warning(f"'nginx -T' failed, falling back to sites-enabled scan: {e}")

        if not vhosts:
            try:
                result = subprocess.run(
                    ['grep', '-rhoE', r'server_name[[:space:]]+[^;]+;', '/etc/nginx/sites-enabled/'],
                    capture_output=True, text=True, timeout=10
                )
                if result.stdout:
                    vhosts = self._parse_server_names(result.stdout)
            except Exception as e:
                logger.warning(f"sites-enabled fallback scan failed: {e}")

        return vhosts if vhosts else None

    @staticmethod
    def _parse_server_names(text: str) -> list:
        """Parse `server_name a b c;` directives out of nginx config text,
        dropping the catch-all `_` placeholder, comments, and duplicates.
        """
        names = []
        seen = set()
        for line in text.splitlines():
            stripped = line.strip()
            if stripped.startswith('#'):
                continue
            m = re.match(r'server_name\s+([^;]+);?', stripped)
            if not m:
                continue
            for token in m.group(1).split():
                token = token.strip().rstrip(';')
                if not token or token == '_':
                    continue
                if token not in seen:
                    seen.add(token)
                    names.append(token)
        return names

    def _fetch_probe_response(self, host: str, path: str) -> Optional[Dict[str, Any]]:
        """Issue one GET for `host` + `path` against the local nginx (127.0.0.1),
        trying https then falling back to http. Returns None on any failure
        (curl missing, timeout, unparseable response) - callers must treat a
        None response as "couldn't verify" (i.e. lean toward confirming the
        original hit), never as "got a 404" or any other positive signal.
        """
        for scheme, port in (('https', 443), ('http', 80)):
            body_path = None
            try:
                fd, body_path = tempfile.mkstemp(prefix='lwai-verify-')
                os.close(fd)
                result = subprocess.run(
                    ['curl', '-sk', '-D', '-', '-o', body_path,
                     '--resolve', f'{host}:{port}:127.0.0.1',
                     '--max-time', '5',
                     f'{scheme}://{host}{path}'],
                    capture_output=True, text=True, timeout=8
                )
                if result.returncode != 0:
                    continue

                status_match = re.search(r'^HTTP/\S+\s+(\d{3})', result.stdout, re.MULTILINE)
                if not status_match:
                    continue
                status = int(status_match.group(1))

                ct_match = re.search(r'^content-type:\s*(.+?)\r?$', result.stdout, re.MULTILINE | re.IGNORECASE)
                content_type = ct_match.group(1).strip() if ct_match else ''

                loc_match = re.search(r'^location:\s*(.+?)\r?$', result.stdout, re.MULTILINE | re.IGNORECASE)
                location = loc_match.group(1).strip() if loc_match else ''

                body = Path(body_path).read_bytes()
                return {'status': status, 'content_type': content_type, 'location': location, 'body': body}

            except Exception as e:
                logger.warning(f"Probe fetch failed for {scheme}://{host}{path}: {e}")
                continue
            finally:
                if body_path:
                    try:
                        os.unlink(body_path)
                    except OSError:
                        pass
        return None

    @staticmethod
    def _has_crash_signal(log_content: str) -> (bool, str):
        for pattern in CRASH_SIGNAL_PATTERNS:
            if pattern.search(log_content):
                return True, pattern.pattern
        return False, ''

    @staticmethod
    def _is_duplicate_ban_noise(log_content: str) -> bool:
        """True when fail2ban is only re-blocking IPs it has already banned before."""
        return 'Duplicate Ban attempts' in log_content

    def apply_severity_ceiling(self, analysis: Dict[str, Any], log_content: str, disk_info: str) -> Dict[str, Any]:
        """Deterministic backstop over the LLM's severity.

        The AI prompt already instructs the model to ignore blocked scans / known-hack
        attempts / fail2ban activity, but it does not reliably follow that instruction
        (observed 2026-08-24: rated HIGH on a report containing only blocked scans and
        duplicate fail2ban bans, zero SSH attempts). Rather than trust the LLM's
        judgment call every run, cap "high"/"critical" back down unless a concrete,
        regex-verifiable signal is present. Never trust LLM-produced fields here
        (e.g. analysis['statistics']['disk_usage_percent']) - only the raw log and our
        own `df` output.
        """
        severity_levels = {'none': 0, 'low': 1, 'medium': 2, 'high': 3, 'critical': 4, 'error': 5}
        severity = analysis.get('severity', 'none')

        # 'error' (API failure / rate limit) must stay visible - never ceiling it.
        if severity == 'error' or severity_levels.get(severity, 0) < severity_levels['high']:
            return analysis

        disk_pct = self._max_disk_usage_percent(disk_info)
        ssh_success, ssh_reason = self._has_verified_ssh_success(log_content)
        http_success, http_reason = self._has_sensitive_http_success(log_content)
        crash, crash_pattern = self._has_crash_signal(log_content)

        # Surface the HTTP-probe live-verification outcome regardless of
        # whether it ended up changing anything, so a human can see why a
        # given probe hit was (or wasn't) treated as real.
        if self._probe_verification_notes:
            analysis['probe_verification_notes'] = list(self._probe_verification_notes)

        if disk_pct >= DISK_CEILING_THRESHOLD_PERCENT or ssh_success or http_success or crash:
            # A concrete hard signal is present - leave the AI's severity as-is.
            logger.info(
                f"Severity ceiling not applied: disk={disk_pct}% ssh_success={ssh_success} "
                f"http_success={http_success} crash={crash}"
            )
            return analysis

        dup_ban_only = self._is_duplicate_ban_noise(log_content)
        new_severity = 'low' if dup_ban_only else 'medium'

        analysis['severity_ceiling_applied'] = True
        analysis['severity_ceiling_reason'] = (
            f"元のAI判定は「{severity}」でしたが、決定論チェック(SSH/HTTPの成功痕跡・"
            f"ディスク{DISK_CEILING_THRESHOLD_PERCENT}%超過・クラッシュ兆候のいずれも未検出)により「{new_severity}」に自動格下げしました。"
            + ("成功痕跡のない失敗探査のみで、fail2banの重複ブロック実績も確認されています。" if dup_ban_only
               else "ブロック済みでない新規アクセス元が含まれるため、mediumで可視性は維持しています。")
        )
        logger.info(
            f"Severity ceiling applied: {severity} -> {new_severity} (dup_ban_only={dup_ban_only})"
        )
        analysis['severity'] = new_severity
        return analysis

    def should_send_alert(self, analysis: Dict[str, Any]) -> bool:
        """Determine if an alert should be sent based on analysis"""
        if self.config['always_send_summary']:
            return True

        severity = analysis.get('severity', 'none')
        threshold = self.config['alert_threshold']

        severity_levels = {
            'none': 0,
            'low': 1,
            'medium': 2,
            'high': 3,
            'critical': 4,
            'error': 4
        }

        return severity_levels.get(severity, 0) >= severity_levels.get(threshold, 2)

    def format_email_body(self, analysis: Dict[str, Any], disk_info: str = "", html: bool = True) -> str:
        """Format analysis results for email"""
        severity = analysis.get('severity', 'unknown').upper()
        severity_ja = {
            'NONE': '正常',
            'LOW': '低',
            'MEDIUM': '中',
            'HIGH': '高',
            'CRITICAL': '緊急',
            'ERROR': 'エラー'
        }
        emoji_map = {
            'NONE': '✅',
            'LOW': '📋',
            'MEDIUM': '⚠️',
            'HIGH': '🔴',
            'CRITICAL': '🚨',
            'ERROR': '❌'
        }
        emoji = emoji_map.get(severity, '❓')
        severity_text = severity_ja.get(severity, severity)
        hostname = socket.gethostname()

        if html:
            body = f"""<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <style>
        body {{ font-family: 'Hiragino Sans', 'Yu Gothic', Arial, sans-serif; line-height: 1.6; }}
        .header {{ background: {'#d4edda' if severity == 'NONE' else '#f8d7da' if severity in ['HIGH', 'CRITICAL'] else '#fff3cd'};
                   padding: 15px; border-radius: 5px; margin-bottom: 20px; }}
        .severity {{ font-size: 24px; font-weight: bold; }}
        .section {{ margin: 20px 0; }}
        .issues {{ background: #f8f9fa; padding: 10px; border-left: 4px solid #dc3545; }}
        .warnings {{ background: #f8f9fa; padding: 10px; border-left: 4px solid #ffc107; }}
        .stats {{ background: #e9ecef; padding: 10px; border-radius: 5px; }}
        ul {{ margin: 10px 0; }}
    </style>
</head>
<body>
    <div class="header">
        <div class="severity">{emoji} 重要度: {severity_text}</div>
        <div>ホスト: {hostname}</div>
        <div>日時: {datetime.now().strftime('%Y年%m月%d日 %H:%M:%S')}</div>
    </div>

    <div class="section">
        <h2>📝 概要</h2>
        <p>{analysis.get('summary', '概要情報がありません')}</p>
    </div>
"""

            if analysis.get('review_failed'):
                body += """
    <div class="section" style="background:#f8d7da;padding:10px;border-left:4px solid #dc3545;">
        <strong>⚠️ AIによる二次レビュー(逐語照合によるハルシネーション検証)がAPI障害のため実行できませんでした。この内容は未検証です。以下は1段目AIの判定のみに基づくため、人の目での確認を推奨します。</strong>
    </div>"""

            if analysis.get('severity_ceiling_applied'):
                body += f"""
    <div class="section" style="background:#d1ecf1;padding:10px;border-left:4px solid #0c5460;">
        <strong>ℹ️ 重要度を自動調整しました:</strong> {analysis.get('severity_ceiling_reason', '')}
    </div>"""

            if analysis.get('probe_verification_notes'):
                body += """
    <div class="section" style="background:#d1ecf1;padding:10px;border-left:4px solid #0c5460;">
        <strong>ℹ️ 機微パスプローブのライブ検証結果:</strong>
        <ul>"""
                for note in analysis['probe_verification_notes']:
                    body += f"\n            <li>{note}</li>"
                body += """
        </ul>
    </div>"""

            if analysis.get('critical_issues'):
                body += """
    <div class="section issues">
        <h3>🚨 緊急対応が必要な問題</h3>
        <ul>"""
                for issue in analysis['critical_issues']:
                    body += f"\n            <li>{issue}</li>"
                body += """
        </ul>
    </div>"""

            if analysis.get('warnings'):
                body += """
    <div class="section warnings">
        <h3>⚠️ 警告</h3>
        <ul>"""
                for warning in analysis['warnings']:
                    body += f"\n            <li>{warning}</li>"
                body += """
        </ul>
    </div>"""

            if analysis.get('statistics'):
                body += """
    <div class="section stats">
        <h3>📊 統計情報</h3>
        <ul>"""
                stats_ja = {
                    'ssh_attempts': 'SSH試行回数',
                    'blocked_ips': 'ブロックされたIP数',
                    'disk_usage_percent': 'ディスク使用率(%)',
                    'errors_count': 'エラー数'
                }
                for key, value in analysis['statistics'].items():
                    label = stats_ja.get(key, key.replace('_', ' ').title())
                    body += f"\n            <li><strong>{label}:</strong> {value}</li>"
                body += """
        </ul>
    </div>"""

            if analysis.get('recommendations'):
                body += """
    <div class="section">
        <h3>💡 推奨対応</h3>
        <ul>"""
                for rec in analysis['recommendations']:
                    body += f"\n            <li>{rec}</li>"
                body += """
        </ul>
    </div>"""

            # ログ詳細（重要な問題がある場合のみ）
            if analysis.get('log_excerpts') and severity in ['MEDIUM', 'HIGH', 'CRITICAL', 'ERROR']:
                body += """
    <div class="section">
        <h3>📄 関連ログ詳細</h3>
        <pre style="background: #f4f4f4; padding: 10px; border-radius: 5px; overflow-x: auto; font-size: 12px;">"""
                for excerpt in analysis['log_excerpts']:
                    import html as html_module
                    body += html_module.escape(str(excerpt)) + "\n"
                body += """</pre>
    </div>"""

            # ディスク容量情報（毎回表示）
            if disk_info:
                body += """
    <div class="section stats">
        <h3>💾 ディスク容量</h3>
        <pre style="background: #e9ecef; padding: 10px; border-radius: 5px; overflow-x: auto; font-size: 12px;">"""
                import html as html_module
                body += html_module.escape(disk_info)
                body += """</pre>
    </div>"""

            body += """
</body>
</html>"""
        else:
            # Plain text version
            body = f"""{emoji} LOGWATCH AI 分析結果 - {datetime.now().strftime('%Y年%m月%d日 %H:%M:%S')}
{'=' * 60}
ホスト: {hostname}
重要度: {severity_text}
概要: {analysis.get('summary', '概要情報がありません')}

"""
            if analysis.get('review_failed'):
                body += "⚠️ AIによる二次レビュー(逐語照合によるハルシネーション検証)がAPI障害のため実行できませんでした。この内容は未検証です。人の目での確認を推奨します。\n\n"

            if analysis.get('severity_ceiling_applied'):
                body += f"ℹ️ 重要度を自動調整しました: {analysis.get('severity_ceiling_reason', '')}\n\n"

            if analysis.get('probe_verification_notes'):
                body += "ℹ️ 機微パスプローブのライブ検証結果:\n"
                for note in analysis['probe_verification_notes']:
                    body += f"  • {note}\n"
                body += "\n"

            if analysis.get('critical_issues'):
                body += "🚨 緊急対応が必要な問題:\n"
                for issue in analysis['critical_issues']:
                    body += f"  • {issue}\n"
                body += "\n"

            if analysis.get('warnings'):
                body += "⚠️ 警告:\n"
                for warning in analysis['warnings']:
                    body += f"  • {warning}\n"
                body += "\n"

            if analysis.get('statistics'):
                body += "📊 統計情報:\n"
                stats_ja = {
                    'ssh_attempts': 'SSH試行回数',
                    'blocked_ips': 'ブロックされたIP数',
                    'disk_usage_percent': 'ディスク使用率(%)',
                    'errors_count': 'エラー数'
                }
                for key, value in analysis['statistics'].items():
                    label = stats_ja.get(key, key.replace('_', ' ').title())
                    body += f"  • {label}: {value}\n"
                body += "\n"

            if analysis.get('recommendations'):
                body += "💡 推奨対応:\n"
                for rec in analysis['recommendations']:
                    body += f"  • {rec}\n"
                body += "\n"

            # ログ詳細（重要な問題がある場合のみ）
            if analysis.get('log_excerpts') and severity in ['MEDIUM', 'HIGH', 'CRITICAL', 'ERROR']:
                body += "📄 関連ログ詳細:\n"
                body += "-" * 40 + "\n"
                for excerpt in analysis['log_excerpts']:
                    body += f"{excerpt}\n"
                body += "-" * 40 + "\n\n"

            # ディスク容量情報（毎回表示）
            if disk_info:
                body += "💾 ディスク容量:\n"
                body += disk_info + "\n"

        return body

    def send_email(self, analysis: Dict[str, Any], disk_info: str = "") -> bool:
        """Send email notification"""
        try:
            severity = analysis.get('severity', 'unknown').upper()
            severity_ja = {
                'NONE': '正常',
                'LOW': '低',
                'MEDIUM': '中',
                'HIGH': '高',
                'CRITICAL': '緊急',
                'ERROR': 'エラー'
            }
            emoji_map = {
                'NONE': '✅',
                'LOW': '📋',
                'MEDIUM': '⚠️',
                'HIGH': '🔴',
                'CRITICAL': '🚨',
                'ERROR': '❌'
            }
            emoji = emoji_map.get(severity, '❓')
            severity_text = severity_ja.get(severity, severity)

            msg = MIMEMultipart('alternative')
            hostname = socket.gethostname()
            msg['Subject'] = f"{emoji} [{hostname}] Logwatch AI レポート - 重要度: {severity_text} - {datetime.now().strftime('%Y年%m月%d日')}"
            msg['From'] = self.config['from_email']
            msg['To'] = ', '.join(self.config['to_emails'])

            # Add both plain text and HTML versions
            text_part = MIMEText(self.format_email_body(analysis, disk_info, html=False), 'plain')
            html_part = MIMEText(self.format_email_body(analysis, disk_info, html=True), 'html')

            msg.attach(text_part)
            msg.attach(html_part)

            # Send email
            # Port 465 uses SSL, not STARTTLS
            if self.config['smtp_port'] == 465:
                smtp = smtplib.SMTP_SSL(self.config['smtp_host'], self.config['smtp_port'])
            else:
                smtp = smtplib.SMTP(self.config['smtp_host'], self.config['smtp_port'])
                if self.config['smtp_use_tls']:
                    smtp.starttls()

            if self.config['smtp_user'] and self.config['smtp_password']:
                smtp.login(self.config['smtp_user'], self.config['smtp_password'])

            smtp.send_message(msg)
            smtp.quit()

            logger.info(f"Email sent successfully to {self.config['to_emails']}")
            return True

        except Exception as e:
            logger.error(f"Failed to send email: {e}")
            return False

    def run(self) -> None:
        """Main execution method with concurrency protection"""
        logger.info("Starting Logwatch AI analysis")

        # Acquire lock to prevent concurrent runs
        lock_fd = self.acquire_lock()
        if not lock_fd:
            logger.error("Could not acquire lock - another instance may be running")
            return

        try:
            # Run logwatch
            logger.info("Running logwatch...")
            log_content = self.run_logwatch()

            if not log_content:
                logger.error("No logwatch output to analyze")
                return

            # Get disk usage information
            logger.info("Getting disk usage information...")
            disk_info = self.get_disk_usage()

            # Analyze with AI
            logger.info("Analyzing logs with AI...")
            analysis = self.analyze_with_ai(log_content)

            # Second-pass review: verify claims against the raw log to catch hallucinations
            logger.info("Reviewing analysis with AI (hallucination check)...")
            analysis = self.review_analysis(analysis, log_content)

            # Deterministic backstop: cap high/critical severity absent a concrete,
            # regex-verifiable signal (the LLM doesn't reliably follow its own
            # "ignore blocked scans" instructions - see apply_severity_ceiling docstring)
            analysis = self.apply_severity_ceiling(analysis, log_content, disk_info)

            # Save analysis results
            analysis_file = Path('/var/log/logwatch-ai-analysis.json')
            analysis_file.write_text(json.dumps(analysis, indent=2))

            # Send alert if needed
            if self.should_send_alert(analysis):
                logger.info(f"Sending alert email (severity: {analysis.get('severity', 'unknown')})")
                self.send_email(analysis, disk_info)
            else:
                logger.info(f"No alert needed (severity: {analysis.get('severity', 'unknown')})")

            logger.info("Logwatch AI analysis complete")

        finally:
            # Always release lock
            self.release_lock(lock_fd)

def main():
    """Main entry point"""
    try:
        analyzer = LogwatchAIAnalyzer()
        analyzer.run()
    except Exception as e:
        logger.error(f"Fatal error: {e}")
        exit(1)

if __name__ == "__main__":
    main()