import imaplib
import sys
import re
import email
import time
import os
from datetime import datetime
from email import message_from_bytes
from email.header import decode_header, make_header
# ==================== 設定項目 ====================
IMAP_SERVER = "mail.university.ac.jp" # 大学のIMAPサーバー
IMAP_PORT = 993 # 通常はSSLの993
USER = "your_username"
PASSWORD = "your_password"
# 最初は True（削除しない）のまま数日間運用し、誤検知がないかログを確認してください。
# 安全が確認できたら False に変更すると、実際にサーバーからスパムを自動削除します。
DRY_RUN = True
# --- 非スパム新着メール通知設定 ---
# mail-notification-receiver.py とセットで機能します。
# <https://chronoflyer.ddo.jp/~trinary/misc/mail-notification-receiver.py.html>
ENABLE_NOTIFICATION = True # Trueで通知有効、Falseで無効
FIFO_PATH = "/var/log/mail-notify.fifo" # 通知連携用の一時ファイル
# 1. Fromヘッダー内の偽装用制御文字を含むメールを削除します。
# 例）差出人 【ス ビーサ送配 n ozamA】
# 2. 件名や送信元のブラックリスト（正規表現パターン）
# ※リスト内で r"..." を使用しているため、re.search にそのまま投入可能です。
SPAM_SUBJECTS = [
r"未承諾広告"
]
SPAM_ENVELOPE_FROM = [
r"<news@cyagenbiomail\.com>",
r"<akrso-.*@",
r"@car-marked\.homes>",
r"@night-flower\.pics>",
r"<JP-[A-Za-z]+-[a-z]+.*=.*@[a-z]{4}\.[a-z]+\.[a-z]{3}",
r"<noreply-.*=.*@tiw[0-9]{2}\.[a-z-]+\.cfd>"
]
SPAM_HEADER_FROM = [
r"<news@cyagenbiomail\.com>",
r"<akrso@",
r"@car-marked\.homes>",
r"@night-flower\.pics>",
r"<JP-[A-Za-z]+-[a-z]+@[a-z]{4}\.[a-z]+\.[a-z]{3}",
r"<noreply@tiw[0-9]{2}\.[a-z-]+\.cfd>",
r"^サイヤジェン株式会社 <",
r"】 <"
]
# 3. 一番下のReceivedヘッダーのブラックリスト（正規表現パターン）
SPAM_RECEIVED_KEYWORDS = [
r"cp\.nov-stells\.pt",
r"sendera[0-9]{2}\.zcsend\.net",
r"mail\.viste-ram\.ru",
r"tiw[0-9]{2}\.[a-z-]+\.cfd"
]
# ==================================================
# # インストール手順
# $ chmod +x /PATH/TO/YOUR/spam-filter-daemon.py
# $ sudo touch /var/log/spam-filter.log
# $ sudo vi /etc/logrotate.d/spam-filter
# /var/log/spam-filter.log {
# weekly
# rotate 4
# compress
# delaycompress
# missingok
# notifempty
# copytruncate
# su root syslog
# # send email for the weekly log to root
# postrotate
# grep "\[DELETED\]" /var/log/spam-filter.log.1
# endscript
# }
#
# # logrotate のテスト
# $ sudo logrotate -d /etc/logrotate.d/spam-filter
#
# $ sudo vi /etc/systemd/system/spam-filter.service
# [Unit]
# Description=IMAP IDLE Spam Filter Daemon
# After=network.target
#
# [Service]
# Type=simple
# ExecStart=/usr/bin/python3 /PATH/TO/YOUR/spam-filter-daemon.py
# Restart=always
# User=root
# Environment=PYTHONUNBUFFERED=1
# StandardOutput=append:/var/log/spam-filter.log
# StandardError=append:/var/log/spam-filter.log
#
# [Install]
# WantedBy=multi-user.target
#
#
# # 設定を反映
# $ sudo systemctl daemon-reload
# # サーバー起動時に自動実行されるようにする
# $ sudo systemctl enable spam-filter.service
# # 今すぐ起動する
# $ sudo systemctl start spam-filter.service
# ==================================================
def get_timestamp():
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def decode_mime_words(s):
if not s: return ""
try:
return str(make_header(decode_header(str(s))))
except Exception:
return str(s)
def send_desktop_notification(subject, header_from):
"""新着メールの情報を一時ファイルに書き込んで、ユーザー権限の通知スクリプトへ合図を送る"""
if not ENABLE_NOTIFICATION:
return
try:
# タブ区切りの1行テキストとして情報を末尾に追記
with open(FIFO_PATH, "a", encoding="utf-8") as f:
f.write(f"{subject}\t{header_from}\n")
except Exception as e:
print(f"[{get_timestamp()}] [WARNING] 通知ファイルへの書き込みに失敗: {e}")
def check_spam(msg):
subject = decode_mime_words(msg.get("Subject", ""))
envelope_from = msg.get("Return-Path", "")
header_from = decode_mime_words(msg.get("From", ""))
# 1. Fromヘッダー内の偽装用制御文字トラップ
hidden_chars_pattern = r"[\u200b\u200c\u200d\u202a\u202b\u202c\u202d\u202e\ufeff]"
if re.search(hidden_chars_pattern, header_from):
return True, "Fromヘッダー内に偽装用の非表示制御文字（ZWJ/RLO等）を検出"
# 2. ブラックリストチェック
for k in SPAM_SUBJECTS:
if re.search(k, subject):
return True, f"件名ブラックリストマッチ: {k}"
for l in SPAM_ENVELOPE_FROM:
if re.search(l, envelope_from):
return True, f"Return-Pathブラックリストマッチ: {l}"
for m in SPAM_HEADER_FROM:
if re.search(m, header_from):
return True, f"差出人ブラックリストマッチ: {m}"
# 3. 一番下の Received ヘッダーのブラックリスト照合
received_headers = msg.get_all('Received', [])
if received_headers:
oldest_received_str = str(received_headers[-1])
for keyword in SPAM_RECEIVED_KEYWORDS:
if re.search(keyword, oldest_received_str):
return True, f"一番下のReceivedヘッダーにブラックリストキーワードを検知: '{keyword}'"
return False, ""
def process_unseen_emails(mail):
"""未読メールをチェックしてスパムを処理し、正常メールなら通知シグナルを送る"""
mail.select("INBOX")
status, response = mail.search(None, "UNSEEN")
mail_ids = response[0].split()
if not mail_ids:
return
deleted_count = 0
for m_id in mail_ids:
status, msg_data = mail.fetch(m_id, "(BODY[HEADER])")
for response_part in msg_data:
if isinstance(response_part, tuple):
msg = message_from_bytes(response_part[1])
is_spam, reason = check_spam(msg)
msg_subject = decode_mime_words(msg.get("Subject", "(No Subject)"))
msg_header_from = decode_mime_words(msg.get("From", "(No From)"))
if is_spam:
msg_envelope_from = msg.get("Return-Path", "(No Return-Path)")
received_headers = msg.get_all('Received', [])
oldest_received = received_headers[-1].replace('\r', '').replace('\n', ' ') if received_headers else "(No Received Header)"
if DRY_RUN:
print(f"[{get_timestamp()}] [DRY-RUN MATCH] Reason: {reason} | Return-Path: {msg_envelope_from} | From: {msg_header_from} | Subject: {msg_subject} | Oldest Received: {oldest_received}")
mail.store(m_id, '-FLAGS', '\\Seen')
else:
print(f"[{get_timestamp()}] [DELETED] Reason: {reason} | Return-Path: {msg_envelope_from} | From: {msg_header_from} | Subject: {msg_subject} | Oldest Received: {oldest_received}")
mail.store(m_id, '+FLAGS', '\\Deleted')
deleted_count += 1
else:
# --- スパムではない正常な新着メールの場合 ---
# 一時ファイルに書き込んでUIプロセス側に通知を委託
send_desktop_notification(msg_subject, msg_header_from)
# 未読状態（UNSEEN）を維持する
mail.store(m_id, '-FLAGS', '\\Seen')
if not DRY_RUN and deleted_count > 0:
mail.expunge()
# imaplibのSSLソケットを確実に捕まえるための拡張クラス
class MyIMAP4_SSL(imaplib.IMAP4_SSL):
def open(self, host='', port=993, timeout=None):
super().open(host, port, timeout)
self.ssl_socket = self.sock
def main():
print(f"[{get_timestamp()}] Starting IMAP IDLE Spam Filter Daemon...")
while True:
try:
mail = MyIMAP4_SSL(IMAP_SERVER, IMAP_PORT)
mail.login(USER, PASSWORD)
mail.select("INBOX")
# 起動直後に、現在溜まっている未読メールを一度すべて処理する
process_unseen_emails(mail)
print(f"[{get_timestamp()}] Entering IDLE mode. Waiting for new messages...")
while True:
mail.send(f'{mail._new_tag()} IDLE\r\n'.encode())
idle_start = time.time()
has_new_mail = False
while time.time() - idle_start < 1200:
mail.ssl_socket.settimeout(10)
try:
resp = mail.ssl_socket.recv(4096).decode('utf-8', errors='ignore')
if "EXISTS" in resp or "RECENT" in resp or "FETCH" in resp:
has_new_mail = True
break
except socket.timeout:
continue
except Exception:
raise
mail.send(b'DONE\r\n')
mail.readline()
if has_new_mail:
print(f"[{get_timestamp()}] New mail notification received!")
process_unseen_emails(mail)
except (imaplib.IMAP4.abort, ConnectionResetError, socket.error) as e:
print(f"[{get_timestamp()}] Connection lost ({e}). Reconnecting in 30 seconds...", file=sys.stderr)
time.sleep(30)
except Exception as e:
print(f"[{get_timestamp()}] Unexpected error: {e}. Retrying in 60 seconds...", file=sys.stderr)
time.sleep(60)
if __name__ == "__main__":
import socket
main()