Anti Spam 杂谈

Posted 2023-03-08 sp42a

本文是一篇随笔，将email的anti spam技术和论坛的防灌水结合在一起讨论。从技术层面出发。不涉及其它。

据说德国有这样一句谚语：没有泡沫的啤酒不是好啤酒。推而广知，可以得到：没人灌水的论坛不是好论坛，没有垃圾邮件的邮件系统不是好系统（至少是不知名的系统／电邮地址），没有病毒骚扰的 OS 不是好的 OS，等等。但是，只有泡沫的啤酒也不是什么好啤酒吧？关键是将不需要的内容控制在可以允许的范围内。单就开论坛、维护垃圾邮件的角度出发，审核技术还是很有用，很有必要的。否则，其地盘很快就会淹没在垃圾广告的汪洋大海之中。自己的论坛，自己发广告是为了维持网站开销，但是不请自来的广告是无法容忍的。

垃圾邮件有两个特点：

• 大量。一封两封垃圾邮件，个人用户可能比较在意，但是对于服务器来说，邮件数以百万计。在这样大的分母下，如果偶然有一两封垃圾邮件被判为合法邮件，或者合法邮件被误判为垃圾，实属正常。
• 不需要。需要与否，取决于用户的主观判断。大家都认为 porn 和 drug 内容是 spam，但是也不排除有人将这类邮件标为 ham 的。

Anti-Spam

常用的反垃圾邮件有以下几种方法：

• 静态关键词列表。如果邮件头（标题，收发件人，电邮地址，正文）含某些关键词（例如 Viagra），就将邮件标为 Spam 或 Ham。这是最简单的方式，为各大邮件厂商所采用，包括 Gmail。它最大的优点是快。缺点多多，一是需要维护（增／改／删 关键词），二是准确率不高（含黑名单中关键词的邮件未必全是 Spam，反之亦然），三是不能识别含有干扰因素的邮件。例如，V1agra，发*漂，含这种关键词的邮件，人眼立即能识别它是垃圾邮件，但是静态法就傻了。
• 正则表达式规则表。与静态关键词列表相反，它速度稍慢，但是极其强大和灵活。对于邮件头的扫描，效果尤佳：邮件头是有规律可循的，尤其是对于大量的垃圾邮件而言，不可能不在邮件头中留下蛛丝马迹。但是这种方法也有其短板。与上述方法类似，它也需要专人来维护，而且无论从配置难度到维护成本，都远高于前者。对于邮件的正文，正则表达式的扫描速度比较缓慢，尤其是对于精心设计了干扰因素的垃圾邮件。有相当大的一部分邮件，人眼看上去确实也是垃圾邮件，但是使用正则表达式也不好写规则。一个新的规则写手，可能要在准确率与查杀总量的折哀上花费很长一段时间才能掌握其规律。
• 贝叶斯概率法。若已知某些字词经常出现在垃圾邮件中，却很少出现在合法邮件中，当一封邮件含有这些字词时，那么link它是垃圾邮件的可能性就很大。参考此文。贝叶斯过滤技术。它最大的优点是，只要有足够健壮的算法，足够的样本空间，其准确率是非常高的。同时，它主要依赖于机器学习，而不需要后期大量的人工干预。

国内有的网站，其内容过滤系统极其简单粗暴，只要出现单个汉字“日”，“操”，“干”等等字眼，就当作垃圾邮件／评论对待，而不分析具体上下文，实在令人又好气又好笑。又有，《百家姓》的常见姓氏用字本身不是垃圾字眼或违禁词汇，如果将其加入静态列表，就会导致连萝卜也无法搜索。其实，用一点点正则表达式（环视）或贝叶斯的技术（条件概率），就能提高过滤质量，皆大欢喜。当然，如果要扫描亿万级的网页，速度的要求肯定要优先于准确度，某些情况下只能做到大致靠谱罢了。然而，频频出错的系统，即使快一点点又有何用（成语：南辕北辙）？ 不过，从来都是宁枉勿纵的。

由于静态法的特点，注定了列表只能向管理员开放，而对普通游客讳莫如深。这导致了另一种现象：该贴无法显示，是因为含有某关键词。至于哪些词是关键词，不好意思，不能告诉你，怕这个列表一旦公开，想发类似内容的人就能轻易绕过。那就有劳管理员们从严自省，并用心地揣摩圣意。

动静结合，人、机结合，系统才能越用越新。@chunzi 说得很形象：反垃圾邮件的过程，不是拼耐力的马拉松赛跑，而是适者生存的进化。总是魔已先高一丈，道才一尺尺增高，并最终压住魔。同时有新的魔即将出现。

一些工具

如果你说自己的邮箱里其实没多少垃圾邮件，或者即使有也已经自动被转入垃圾箱了，那么有两种可能。一是你所用的邮箱系统本身的反垃圾邮件系统做得不错（有太多太多各式各样的明显的垃圾邮件在进入你的邮箱之前，已经被服务器端给 block了）。二是你的邮址没有被爬虫抓到或算出来。

SpamAssassin 是一套不错的反垃圾邮件系统。免费，与 Apache 紧密结合，强大的正则式支持。国内有一个组织专门动态维护一个中文的规则表，在这里，可以参考。SpamAssassin 其实也有贝叶斯模块，只是它以正则知名罢了。

WordPress 有个 Akismet 插件是用来 block 博客上的垃圾评论的。这个设置起来比较傻瓜（只需要申请一个API）即可，效果比较智能，完全不用用户再手工添加任何规则。对于出错的判断，用户有义务提交给 Akismet 官方，方便它学习新的变种。应该说，用户提交的漏判或误判，是必不可少的语料库。没有用户提交，Akismet 就会一根筋地按照既定的思路继续犯同样的错误。

个人应用

国人开发的 Discuz 是一款不错的论坛程序。不过，它没有较好的反垃圾模块，我一直在颇厌其烦地删除 spam 和 spammer。看了一下Ak的官网，几乎国外知名的论坛程序都有 Ak 的插件了。我研究了 dz 的数据结构，使用 python 写了一个脚本，定时搜索新贴子，将其提交到 Ak 做判断。如果判为垃圾，则屏蔽贴子，并对该会员实施减分操作。刚开始试用，效果还可以。其实可以做成原生的 php 插件，集成到 dz 中的。

流程很直接：

• 定时搜索最新贴子；
• 对于每一个新贴子:
  - 提交给 Akismet作 测试。
  - 如果不是垃圾，忽视之。
  - 如果是垃圾，将该贴转为仅管理员可见。同时将该用户扣分。
• 总合一下应该执行的操作，执行 SQL, Commit。
• 生成报表，发邮件给管理员。

Ak 的开发者页面在这里 Akismet API Documentation。我用了其中的 Python 模块 将其封装为一个 class，只需要 init 和 check 即可：

#!/usr/bin/python
# -*- coding: utf-8 -*-
#
#author:         rex
#blog:           http://iregex.org
#filename        comment.py
#created:        2010-08-14 15:58

from akismet import Akismet

class Comment():
    def __init__(self):
        self.api=Akismet()

        if self.api is None or not self.api.verify_key():
            print 'No Valid Akismet API'
            exit(1)

    def init(self, comment, user='', ip='', email=''):
        self.user=user.encode("utf-8")
        self.comment=comment.encode("utf-8")
        self.ip=ip
        self.email=email

    def check(self):
        return self.api.comment_check(self.comment,
                    
                         'comment_author': self.user,
                         'comment_author_email':self.email,
                         'user_ip':self.ip,
                         'user_agent':"Mozilla/5.0 (X11; U; Linux i686; en-US; rv:1.9.2.8) Gecko/20100723 Ubuntu/10.04 (lucid) Firefox/3.6.8",
                    ,
                )

这是程序的主要部分

#!/usr/bin/python
# -*- coding: utf-8 -*-
#
#author:         rex
#blog:           http:#iregex.org
#filename        dzas.py
#created:        2010-08-14 15:20

#anti spam for discuz! bbs.

from comment import Comment as C
from eml import Email

#########################################################
#global settings
send_email_log =1

dbhost = 'yourhost.website.com';
dbuser = 'db_user';              
dbpw = 'yourpassword';            
dbname = 'dbname';    
dbpre='cdb_'


#punish : punish for the user if he/she publish spam
punish_score=2

#now is 2 hours; you may change
sql=
    'last_n_hr':'''select `%sposts`.pid, `%sposts`.author, `%sposts`.message, `%sposts`.useip, `%smembers`.email, authorid from `%sposts`, `%smembers` where dateline>UNIX_TIMESTAMP(now())-2*3600 and `%sposts`.author=`%smembers`.username order by pid desc;''' % (dbpre,dbpre,dbpre,dbpre,dbpre,dbpre,dbpre,dbpre,dbpre,),
    'hide':'update %sposts set `status`= 1 where pid in (%s);',
    'punish':'update %smembers set credits=credits-%s where uid=%s;',
    'find_hided':"SELECT * FROM  `%sposts` WHERE  `status`=1;"%(dbpre),


import mysqldb

def init_db():
    db=MySQLdb.connect(host=dbhost, user=dbuser, passwd=dbpw,db=dbname, charset='utf8')
    return db

def hide_spam(db, spam):
    c=db.cursor()
    sql_str=sql['hide'] % (dbpre,','.join(spam))
    c.execute(sql_str)
    db.commit()

def punish(db,score):
    c=db.cursor()

    for u in score.keys():
        s=score[u]
        sql_str=sql['punish'] % (dbpre,s*punish_score,u)
        c.execute(sql_str)
    db.commit()

def get_msg(db):
    c=db.cursor()
    c.execute(sql['last_n_hr'])
    records= c.fetchall()
    result=[]
    for r in records:
        result.append(
                
                    'pid':r[0],
                    'user':r[1],
                    'msg':r[2],
                    'ip':r[3],
                    'email':r[4],
                    'uid':r[5]
                )

    return result

#send log to admin. change the global variable
#send_email_log = 1 or 0 to enable/disable
def report(spam, score):

    sub="%s spams caputured" % len(spam)
    body="spammers including: %s\\n" % ', '.join( set([m['user'] for m in spam] ))
    body+="spam pids including: %s\\n" % ', '.join( [ str(m['pid']) for m in spam] )
 
    body+="useful sql: %s\\n\\n" % sql['find_hided']
    body+="spam preview: \\n%s" % "\\n".join( m['msg'].splitlines()[0] for m in spam)
    Email(sub.encode("utf-8"),body.encode("utf-8"))

#the core part

def anti_spam(db, msgs):
    c=C()
    spam=[]
    score=.fromkeys([m['uid'] for m in msgs], 0)
    for m in msgs:
        c.init(m['msg'], m['user'], m['ip'], m['email'])
        if c.check():
            score[m['uid']]+=1
            spam.append(m)

    for s in score.keys():
        if score[s]==0:
            del score[s]

    if score and spam:
        hide_spam(db, [str(m['pid']) for m in spam])
        punish(db, score)

        if send_email_log:
            report(spam, score)

def main():
    db=init_db()
    msg=get_msg(db)
    anti_spam(db, msg)

if __name__=='__main__':
    main()

参阅：https://blog.csdn.net/weixin_45583158/article/details/100143406