メールに jev で迷惑度スコアを計算させて、新手の迷惑メールも検知する

はじめに

迷惑メールが後を絶ちません。最近のAI性能向上もあって、より精巧で自然で既存のスパムフィルターもすり抜けるようになってきました。
メールサーバー側で postscreen や SpamAssassin のようなプラグインを導入したり、外部サービスの MXToolbox や Spamhaus でブラックリストのチェックを入れたりしても、それでも間に合いません。

ここではメールフィルターにプロンプトの代わりにスコアを返す jev を追加することで、それでもすり抜けてやってくる迷惑メールに「迷惑度」という、AI判定結果が反映されるようにします。

このAI判定結果はカスタムヘッダーに付与されるため、あとはメールクライアント側のフィルター設定で閾値を設定して、一定の迷惑度を超えているものは迷惑フォルダに移動したり、直接削除したりして、適宜運用できればと思っています。

実装環境

• AlmaLinux 10.*
• postfix 3.*
• python 3.*
• typesafe-sdk 0.7.2

導入方法

filter.py

from __future__ import annotations

import hashlib
import json
import os
import re
import subprocess
import sys
import time
from email import policy
from email.header import decode_header, make_header
from email.message import Message
from email.parser import BytesParser
from pathlib import Path

import constants as C
from typesafe_sdk import Choice, Noul, TypeSafeClient

CATEGORY_CRITERIA = {
    "phishing": "フィッシングやブランド詐称の疑い",
    "scam": "詐欺・当選・緊急性の高い金銭誘導",
    "advertising": "広告・宣伝・セールス",
    "suspicious": "その他の不審なメール",
    "legitimate": "正常・業務上妥当なメール",
}


def _load_prompt(path: Path) -> str:
    return path.read_text(encoding="utf-8").strip()


def _decode_mime_header(value: str | None) -> str:
    if not value:
        return ""
    try:
        return str(make_header(decode_header(value)))
    except Exception:
        return value


def _domain_of(addr: str) -> str:
    addr = addr.strip().strip("<>")
    if "@" not in addr:
        return ""
    return addr.rsplit("@", 1)[-1].lower()


def _clip(text: str, limit: int = C.HEADER_FIELD_MAX_CHARS) -> str:
    text = re.sub(r"\s+", " ", text).strip()
    if len(text) <= limit:
        return text
    return text[: limit - 1] + "…"


def _parse_authentication_results(msg: Message) -> dict[str, str]:
    result = {"spf": "none", "dkim": "none", "dmarc": "none", "header_from": ""}
    headers = msg.get_all("Authentication-Results", [])
    if not headers:
        return result

    for raw in headers:
        line = " ".join(str(raw).split())
        for key in ("spf", "dkim", "dmarc"):
            m = re.search(
                rf"\b{key}=(pass|fail|softfail|neutral|none|temperror|permerror)\b",
                line,
                re.I,
            )
            if m:
                result[key] = m.group(1).lower()
        m = re.search(r"\bheader\.from=([^\s;]+)", line, re.I)
        if m:
            result["header_from"] = m.group(1).lower().strip("<>")
    return result


def extract_header_features(raw_email: str) -> dict[str, str]:
    msg = BytesParser(policy=policy.default).parsebytes(
        raw_email.encode("utf-8", errors="replace")
    )
    from_raw = msg.get("From", "") or ""
    from_display = _decode_mime_header(from_raw)
    from_addr = ""
    m = re.search(r"<([^>]+)>", from_raw)
    if m:
        from_addr = m.group(1).strip()
    elif "@" in from_raw:
        from_addr = from_raw.strip()

    auth = _parse_authentication_results(msg)
    return {
        "message_id": (msg.get("Message-ID") or msg.get("Message-Id") or "").strip(),
        "subject": _clip(_decode_mime_header(msg.get("Subject"))),
        "from_display": _clip(from_display),
        "from_addr": from_addr,
        "from_domain": _domain_of(from_addr),
        "return_path": _clip(msg.get("Return-Path") or ""),
        "reply_to": _clip(_decode_mime_header(msg.get("Reply-To"))),
        "spf": auth["spf"],
        "dkim": auth["dkim"],
        "dmarc": auth["dmarc"],
        "header_from": auth["header_from"],
    }


def build_ai_document(features: dict[str, str]) -> str:
    return "\n".join(
        [
            f"[AUTH] spf={features['spf']} dkim={features['dkim']} dmarc={features['dmarc']}",
            f"[DMARC header.from] {features['header_from'] or '(none)'}",
            f"[FROM] display={features['from_display']}",
            f"[FROM] addr={features['from_addr']} domain={features['from_domain']}",
            f"[RETURN-PATH] {features['return_path'] or '(none)'}",
            f"[REPLY-TO] {features['reply_to'] or '(none)'}",
            f"[SUBJECT] {features['subject'] or '(none)'}",
        ]
    )


def build_reasons(features: dict[str, str], category: str, noul: float) -> str:
    parts = [
        f"category={category}",
        f"noul={noul:.3f}",
        f"spf={features['spf']}",
        f"dkim={features['dkim']}",
        f"dmarc={features['dmarc']}",
    ]
    if features["from_domain"]:
        parts.append(f"from_domain={features['from_domain']}")
    if features["header_from"]:
        parts.append(f"header_from={features['header_from']}")
    return ";".join(parts)


def _cache_path(message_id: str) -> Path:
    digest = hashlib.sha256(message_id.encode("utf-8")).hexdigest()
    return C.CACHE_DIR / f"{digest}.json"


def cache_get(message_id: str) -> dict | None:
    if not C.CACHE_ENABLED or not message_id:
        return None
    path = _cache_path(message_id)
    if not path.is_file():
        return None
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        if time.time() - float(data["ts"]) > C.CACHE_TTL_SECONDS:
            path.unlink(missing_ok=True)
            return None
        return data
    except Exception:
        return None


def cache_put(message_id: str, score: int, reasons: str) -> None:
    if not C.CACHE_ENABLED or not message_id:
        return
    C.CACHE_DIR.mkdir(parents=True, exist_ok=True)
    path = _cache_path(message_id)
    tmp = path.with_suffix(".tmp")
    payload = {
        "ts": time.time(),
        "score": score,
        "reasons": reasons,
        "message_id": message_id,
    }
    tmp.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
    tmp.replace(path)


def classify_lane(_features: dict[str, str]) -> str:
    if C.SHORT_CIRCUIT_MODE == "always_gray":
        return "gray"
    return "gray"


def run_ai(features: dict[str, str]) -> tuple[int, str]:
    api_key = os.environ.get("TYPESAFE_API_KEY")
    if not api_key:
        raise RuntimeError("not found TYPESAFE_API_KEY")

    document = build_ai_document(features)
    with TypeSafeClient(api_key=api_key) as client:
        response = client.system_one(
            state={"document": document},
            questions={
                "is_spam": Noul(instructions=_load_prompt(C.PROMPT_IS_SPAM)),
                "category": Choice(
                    instructions=_load_prompt(C.PROMPT_CATEGORY),
                    criteria=CATEGORY_CRITERIA,
                ),
            },
        )

    noul = float(response.nouls["is_spam"].noul)
    score = max(0, min(C.SCORE_SCALE, int(noul * C.SCORE_SCALE)))
    category = "suspicious"
    if "category" in response.choices:
        category = response.choices["category"].choice
    return score, build_reasons(features, category, noul)


def deliver(output_email: str, argv: list[str]) -> None:
    if "--test" in argv:
        sys.stdout.write(output_email)
        return
    recipients = [a for a in argv[1:] if a != "--test"]
    subprocess.run(
        ["/usr/sbin/sendmail", "-i"] + recipients,
        input=output_email,
        text=True,
        check=False,
    )


def main() -> None:
    raw_email = sys.stdin.read()
    features = extract_header_features(raw_email)
    lane = classify_lane(features)

    if lane in ("ok", "ng"):
        deliver(raw_email, sys.argv)
        return

    score: int | None = None
    reasons: str | None = None
    error_msg: str | None = None

    cached = cache_get(features["message_id"])
    if cached is not None:
        score = int(cached["score"])
        reasons = str(cached["reasons"])
    else:
        try:
            score, reasons = run_ai(features)
            cache_put(features["message_id"], score, reasons)
        except Exception as e:
            error_msg = str(e)

    custom_headers = ""
    if error_msg:
        custom_headers += f"X-AI-SPAM-ERROR: {error_msg}\n"
    else:
        custom_headers += f"X-AI-SPAM-SCORE: {score}\n"
        custom_headers += f"X-AI-SPAM-REASONS: {reasons}\n"

    deliver(custom_headers + raw_email, sys.argv)


if __name__ == "__main__":
    main()

constants.py

from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent

# --- パス ---
PROMPT_IS_SPAM = BASE_DIR / "prompts" / "is_spam.txt"
PROMPT_CATEGORY = BASE_DIR / "prompts" / "category.txt"
CACHE_DIR = BASE_DIR / ".cache" / "results"

# --- キャッシュ（同一 Message-ID の複数宛先で AI 再利用） ---
CACHE_TTL_SECONDS = 6 * 60 * 60  # 6時間
CACHE_ENABLED = True

# 安全だということが分かっているホワイトリストや、
# 辞書を使った"rules" などに拡張できるように
# いったんグレーかどうかのみの判定
SHORT_CIRCUIT_MODE = "always_gray"

# --- スコア ---
# X-AI-SPAM-SCORE = int(noul * SCORE_SCALE)、0..SCORE_SCALE
SCORE_SCALE = 100

# --- AI 入力 ---
# 構造化ヘッダ要約に含める件名/表示名の最大文字数
HEADER_FIELD_MAX_CHARS = 200

prompts/is_spam.txt

次のメールヘッダ要約だけを見て、スパム・広告・フィッシング・ブランド詐称のいずれかに該当するかを判定してください。
本文はありません。表示名と From ドメインの不一致、件名の誘導・脅迫・偽通知っぽさ、認証結果（spf/dkim/dmarc）を重視してください。
認証が pass でも、表示上のブランドと送信ドメインが食い違う場合は危険寄りに判定してください。

prompts/category.txt

ヘッダ要約だけを見て、最も近い分類を1つ選んでください。本文はありません。

.env

TYPESAFE_API_KEY={あなたのAPIキー}

filter-wrapper.sh

#!/bin/bash
set -euo pipefail

DIR="$(cd "$(dirname "$0")" && pwd)"
ENV_FILE="${DIR}/.env"
PYTHON="/home/{あなたのアカウント}/.pyenv/versions/{あなたのバージョン}/bin/python3"
FILTER="${DIR}/filter.py"

if [[ ! -x "$PYTHON" ]]; then
  echo "filter-wrapper: python not found: $PYTHON" >&2
  exit 1
fi

if [[ ! -f "$FILTER" ]]; then
  echo "filter-wrapper: filter.py not found: $FILTER" >&2
  exit 1
fi

if [[ -f "$ENV_FILE" ]]; then
  set -a
  # shellcheck disable=SC1090
  source "$ENV_FILE"
  set +a
fi

exec "$PYTHON" "$FILTER" "$@"

/etc/postfix/master.cf

smtp      inet  n       -       n       -       -       smtpd
  -o content_filter=ai-spam-filter:dummy
(略)
ai-spam-filter unix -       n       n       -       5       pipe
  flags=Rq user={あなたのアカウント} argv=filter-wrapper.sh ${recipient}

運用方法

あとは postfix をリロードすれば、その後メールクライアントには X-AI-SPAM-SCORE という百分率の迷惑度が自動で付与されたメールが確認できるようになります。
「X-AI-SPAM-SCORE が "80" 以上なら迷惑フォルダに移動するようにクライアント側で設定する」等で、既存のソリューションでは防ぎきれなかった迷惑メールのブロック精度向上に寄与できるのではないかと思います。

あとがき

防御の非対称性 に負けず、できるところから積極手的に攻めの防御をしていきましょう。
我々には「頑張る」という選択肢しか残されていないのですから。

ちなみに、本当は「受信メールのスパムフィルターに jev を導入する」程度の弱いタイトルだったのですが、タイトルも頑張って強気めに変えました。