I needed to implement SpamAssasin directly inside my Python code (to control the entire flow) so I implemented a client using spamd (the SpamAssassin daemon) that I connect via socket. Here's the code, with an example of how to use it.

If you have any suggestions on how to improve it (keeping exotic libraries out is better), I'm happy to get your opinion!

Note: I haven't find something similar, that's why I did it. I used a lot of code from https://github.com/slimta/python-slimta/blob/master/slimta/policy/spamassassin.py though.

# -*- config:utf-8 -*-

import socket, select, re, logging
from io import BytesIO

divider_pattern = re.compile(br'^(.*?)\r?\n(.*?)\r?\n\r?\n', re.DOTALL)
first_line_pattern = re.compile(br'^SPAMD/[^ ]+ 0 EX_OK$')

# @see https://github.com/slimta/python-slimta/blob/master/slimta/policy/spamassassin.py
class SpamAssassin(object):
    def __init__(self, message, timeout=3):
        self.score = None
        self.symbols = None

        # Connecting
        client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        client.connect(('', 783))

        # Sending

        # Reading
        resfp = BytesIO()
        while True:
            ready = select.select([client], [], [], timeout)
            if ready[0] is None:
                # Kill with Timeout!
                logging.info('[SpamAssassin] - Timeout ({0}s)!'.format(str(timeout)))

            data = client.recv(4096)
            if data == b'':


        # Closing
        client = None


    def _build_message(self, message):
        reqfp = BytesIO()
        data_len = str(len(message)).encode()
        reqfp.write(b'SYMBOLS SPAMC/1.2\r\n')
        reqfp.write(b'Content-Length: ' + data_len + b'\r\n')
        reqfp.write(b'User: cx42\r\n\r\n')
        return reqfp.getvalue()

    def _parse_response(self, response):
        if response == b'':
            logging.info("[SPAM ASSASSIN] Empty response")
            return None

        match = divider_pattern.match(response)
        if not match:
            logging.error("[SPAM ASSASSIN] Response error:")
            return None

        first_line = match.group(1)
        headers = match.group(2)
        body = response[match.end(0):]

        # Checking response is good
        match = first_line_pattern.match(first_line)
        if not match:
            logging.error("[SPAM ASSASSIN] invalid response:")
            return None

        self.symbols = [s.decode('ascii').strip() for s in body.strip().split(',')]

        headers = headers.replace(' ', '').replace(':', ';').replace('/', ';').split(';')
        self.score = float(headers[2])

    def get_score(self):
        return self.score

    def get_symbols(self):
        return self.symbols

    def is_spam(self, level=5):
        return self.score is None or self.score > level

And here's how to use it:

from spamassassin import SpamAssassin

# message contains the DATA part
assassin = SpamAssassin(message)
if assassin.is_spam():
    print "The received message is considered spam with a score of {0}".format(assassin.get_score())

That's all :) Let me know


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.