Вас тоже достаёт спам и реклама? Рекламу я блокирую через свой DNS сервер и локальными CSS фильтрами, а вот для почты пришлось придумать что-то другое.
Почта у меня устроена немного необычно (я так думаю). Все мои ящики — Яндекс, Mail, старые адреса — настроены на пересылку на мой домен. С домена всё летит в Gmail. Протоколом IMAP не пользуюсь много лет (да ещё какие то платные мутки в РФ недавно придумали). А вот интерфейс Google меня полностью устраивает, поиск мгновенный, удобные метки и сортировка. Есть мелочи, которые не нравятся, но лучше варианта, который закрывал бы мои потребности, пока не встречал. В итоге получается единая точка, где видна вся переписка.
Схема многоступенчатая. Первыми срабатывают фильтры Яндекса и Mail — что-то они отсеивают сами, ещё до пересылки. То, что прошло через них, падает на мой сервер, где стоит SpamAssassin. Ловит ещё часть. Но после двух уровней всё равно что-то просачивается, спамеры же не сидят без дела. И вот этот остаток доезжает до Gmail и что-то оседает в папке Спам, а что-то попадает во входящие и приходит раздражающее уведомление. Хотелось, чтобы со временем не накапливался мусор в папках, который надо разгребать вручную. Особенно важно заблокировать то, что не является полностью спамом: приглашения на конференции, партнёрские предложения, кредиты — формально не нарушение, поэтому байесовский фильтр такие вещи плохо ловит.
Сначала думал взять LLM через API. Llama или GPT отлично разбирают текст, с ними гибко настраиваются критерии. Но это внешняя зависимость. В наше нестабильное сетевое время желательно уменьшать такие зависимости. Сервис может поменять условия, отключиться временно, геоблокировки, да и много чего ещё.
Локальная BERT-модель закрыла обе проблемы. Взял ruBert-base-antispam с HuggingFace — файн-тюн на базе DeepPavlov/rubert-base-cased-conversational. 177 миллионов параметров, 12 слоёв трансформера, 768 hidden size. Физически не принимает больше 512 токенов на вход. В памяти занимает около 550 МБ, ответ приходит за 100-200 миллисекунд. Бинарный классификатор — текст на входе, 0 или 1 на выходе, никаких промптов и reasoning. Идеально!
Встроить вызов скрипта прямо в SMTP-конвейер Exim оказалось непросто. Механизмы вроде ACL или transport_filter имеют свои особенности, особенно с мудреной панелью сервера, которая сама обновляет многие файлы. А то будет так: где-то письмо недоступно целиком, где-то прав не хватает, где-то модифицированный текст не передаётся дальше, где-то файл обновился. Пошёл другим путём — post-delivery обработкой.
Получилась трёхуровневая серверная защита. Первый уровень — pre-rules, набор регулярок для очевидного спама. Срабатывают за миллисекунду, без обращения к нейросети. Второй уровень — сама BERT-модель, обрабатывает всё, что не отсекли регулярки, это около 70% потока. Третий уровень работает параллельно — SpamAssassin со своим байесовским фильтром. Если он ставит score выше порога, Dovecot через Sieve перемещает письмо в .Spam. Это страховка на случай, если первые два уровня что-то пропустили. При этом AI-скрипт всё равно проверяет письмо: если SpamAssassin ошибся и пометил нормальное письмо, а AI считает его чистым, оно всё равно уйдёт на Gmail. Fail-open стратегия в AI-скрипте спасает от параноидального SpamAssassin и без сильных заморочек с коэффициентами.
Схема работает так. Exim принимает письмо и складывает его в Maildir, в папку new/. Отдельный скрипт каждые полминуты обходит все директории. Из MIME вытаскивает текст и заголовки — поддерживает и plain, и HTML, теги чистит стандартным HTMLParser. Сначала гонит через pre-rules, причём смотрит не только в текст, но и в заголовки. Что не отсеялось — идёт в постоянно работающий daemon через Unix-сокет. Спам перекладывается в .Spam/, нормальные письма идут в cur/ и параллельно улетают на Gmail через sendmail.
Задержка на доставку письма в 30–60 секунд меня не беспокоит. В почте я в реальном времени не сижу, это ж не чат, да и Gmail подтягивает не прям мгновенно. Почта нужна для спокойных переписок, а не для гонок кто быстрее ответил.
Привожу пример daemon (/opt/spam_filter/daemon.py):
#!/usr/bin/env python3
import socket, sys, os, re, signal, logging
import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification
MODEL_NAME = "assskelad/ruBert-base-antispam" # любая BertForSequenceClassification
SOCKET_PATH = "/tmp/spam_filter.sock"
MAX_LEN = 512 # лимит BERT
MAX_TEXT = 6000 # символы из письма
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("spam_daemon")
torch.set_num_threads(1)
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
model = AutoModelForSequenceClassification.from_pretrained(MODEL_NAME)
model.eval()
def predict(text: str) -> int:
inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=MAX_LEN)
with torch.inference_mode():
out = model(**inputs)
return int(torch.argmax(out.logits, dim=1).item())
def handle(conn):
try:
data = conn.recv(1024 * 1024)
text = data.decode("utf-8", errors="replace")[:MAX_TEXT]
result = predict(text) if text else 0
conn.sendall(str(result).encode())
except Exception as e:
logger.error(f"Error: {e}")
conn.sendall(b"0")
finally:
conn.close()
def shutdown(signum, frame):
try:
os.unlink(SOCKET_PATH)
except:
pass
sys.exit(0)
signal.signal(signal.SIGTERM, shutdown)
signal.signal(signal.SIGINT, shutdown)
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
srv.bind(SOCKET_PATH)
srv.listen(10)
while True:
conn, _ = srv.accept()
handle(conn)
Pre-rules (/opt/spam_filter/pre_rules.py) — паттерны:
#!/usr/bin/env python3
import re
def check_pre_rules(text, headers=None):
text_lower = text.lower()
headers = headers or {}
# паттерны
spam_patterns = [
r'рассылк[аи]s+поs+баз|миллионs+адрес',
r'внедрени.{0,15}crm|ботs+продвижени.{0,20}(?:telegram|whatsapp)',
r'диплом.{0,20}(?:профпереподготовк|допобразован)',
r'бухгалтерск.{0,30}сопровожден|нулевой.{0,20}ндс',
r'спарсить.{0,30}базу|парсинг.{0,30}клиентов',
r'подтвердите.{0,30}данны.{0,200}предложениеs+истекает',
r'(?:[а-я]s){4,}',
r'деньгиs+подs+залог.{0,20}(?:авто|машин)',
r'сеоs+оптимизац.{0,20}недорого',
]
for pattern in spam_patterns:
if re.search(pattern, text_lower):
return "SPAM"
# рассылки
bulk_headers = any(
k.lower() in ['list-unsubscribe', 'precedence', 'feedback-id', 'x-rpcampaign']
for k in headers
)
if bulk_headers and re.search(r'предлагаем|приглашаем|бесплатн.{0,20}консультац', text_lower):
return "SPAM"
# спам
if re.search(r'кето|метаболизм|инсулин|жиросжигани[ею]|жкт', text_lower) and
re.search(r'тыs+устал|энерги[яи]s+наs+нуле|мыs+да[её]мs+систему', text_lower):
return "SPAM"
# нужные письма
ham_patterns = [
r'кодs+(?:подтвержден|дляs+вход|доступа):s*d{4,6}',
r'вашs+заказs+№?d+|orders+#?d+',
r'сбросs+парол|passwords+reset',
r'платежs+прошел|payments+received',
r'встреч[аи]s+(?:перенесен|назначен|состоится)',
]
for pattern in ham_patterns:
if re.search(pattern, text_lower):
return "HAM"
return None
Правило про массовые рассылки оказалось очень полезным. Нейросеть часто пропускает такие письма, потому что они грамматически корректны и не содержат явного мусора. А связка заголовок List-Unsubscribe + слово приглашаем в тексте даёт 100% точности без ложных срабатываний на личные письма. Правило про медицинские термины — пример составной регулярки: по отдельности слова могут употребляться в переписке, но в сочетании с некоторыми это однозначный спам. Так файл pre_rules.py со временем обрастает специфичными правилами под ваш поток, который иногда можно пополнять.
Сокращенный Post-delivery скрипт:
#!/usr/bin/env python3
import os, sys, socket, datetime, time, subprocess
from email import policy
from email.parser import BytesParser
from html.parser import HTMLParser
import re
SOCKET = "/tmp/spam_filter.sock"
MAIL_BASE = "/home/admin/mail" # путь
FORWARD_TO = "your@gmail.com" # куда присылать
WAIT_SECONDS = 5 # ожидание
LOCK_FILE = "/tmp/ai_spam_check.lock"
def check_spam(email_bytes):
text, headers = extract_text_and_headers(email_bytes)
pre_verdict = check_pre_rules(text, headers)
if pre_verdict == "SPAM":
return True, text[:100], "pre_rule"
elif pre_verdict == "HAM":
return False, text[:100], "pre_rule"
text_bytes = text.encode('utf-8', errors='ignore')
try:
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.settimeout(2)
s.connect(SOCKET)
s.settimeout(5)
s.sendall(text_bytes)
s.shutdown(socket.SHUT_WR)
r = s.recv(8).decode("ascii", errors="ignore").strip()
s.close()
return r == "1", text[:100], "ai"
except Exception as e:
log(f"DAEMON_ERROR: {e}")
return False, text[:100], "error"
def forward_to_gmail(email_bytes, filename=""):
modified = b"X-Forwarded-By: AI-Filtern" + email_bytes
p = subprocess.Popen(
["/usr/sbin/sendmail", "-i", FORWARD_TO],
stdin=subprocess.PIPE,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
p.stdin.write(modified)
p.stdin.close()
p.wait()
def process_new_dir(new_dir, spam_new_dir, cur_dir):
for filename in os.listdir(new_dir):
filepath = os.path.join(new_dir, filename)
if time.time() - os.path.getmtime(filepath) < WAIT_SECONDS:
continue
with open(filepath, "rb") as f:
email_bytes = f.read()
if b"X-Forwarded-By: AI-Filter" in email_bytes:
os.rename(filepath, os.path.join(cur_dir, filename + ":2,"))
continue
is_spam, preview, method = check_spam(email_bytes)
if is_spam:
os.rename(filepath, os.path.join(spam_new_dir, filename))
log(f"SPAM [{method}]: {filename} | {preview}")
else:
os.rename(filepath, os.path.join(cur_dir, filename + ":2,"))
forward_to_gmail(email_bytes, filename)
log(f"NOT_SPAM [{method}]: {filename} | {preview}")
Самое интересное — гонки с MTA. Если читать файл сразу после создания, можно получить недописанное письмо пока оно сохраняется на носитель. Отсюда WAIT_SECONDS: скрипт пропускает файлы моложе заданного возраста. Это плата за post-delivery.
Петли пересылки тоже пришлось учесть. Если случайно отправить самому себе или замкнуть цепочку, начнётся бесконечный цикл. Защита — заголовок X-Forwarded-By: скрипт видит его и пропускает такие письма, просто перемещая в cur/.
По RAM памяти: модель занимает около 550 МБ, скрипт проверки при запуске ещё около 50 МБ, Exim 25 МБ, Dovecot 10 МБ, SpamAssassin 80 МБ. Итого около 716 МБ на всю систему. У меня сервер на 2 гига, в итоге занято около одного. Всё стабильно, без свопа. Для более слабых машин подошёл бы DistilBERT, он вдвое легче, но у меня запаса хватает.
Если daemon упал, сокет недоступен или произошла любая ошибка — письмо считается нормальным и идёт дальше. Лучше получить пару спам-писем, чем пропустить важное.
Для мониторинга есть bash-скрипт SPAM, который показывает статус всех компонентов, статистику за день, состояние ящиков и trend за последний час прямо в терминале. Команда SPAM -w даёт live-обновление каждые 3 секунды, SPAM -t тестирует модель на 10 примерах, SPAM -r — pre-rules на 21 примере. Для админа почтового сервера удобный CLI-мониторинг оказывается полезнее любого веб-интерфейса. Логи пишу в /var/log/ai_spam.log с пометками [pre_rule] или [ai], чтобы было видно, какой уровень сработал.
Спам перестал доезжать до Gmail. Что раньше оседало в папке и требовало периодической чистки, теперь остаётся на сервере в .Spam/. На Gmail попадают только письма, прошедшие все три фильтра. Иногда модель ошибается — на редких темах или специфическом сленге. В таких случаях добавляю паттерн в pre_rules.py, и проблема больше не повторяется. Раз в месяц можно глянуть в .Spam/, просматриваю заголовки — не потерялось ли чего. Обычно нет. А если что и попадет по ошибке — пара строк в конфиге, и проблема закрыта.
Нагрузка на сервер минимальная, при сотне писем в день daemon большую часть времени спит, CPU почти не ест. Раз настроил и работает, иногда только добавляю новые регулярки, когда замечаю в логах повторяющийся мусор. Проблема со спамом закрыта!
А как вы боретесь со спамом на своих серверах? Если будут нужны исходники, то выложу на GitHub.
Автор: ArtemErykov
