Skip to content

Instantly share code, notes, and snippets.



Last active Feb 28, 2021
What would you like to do?
DKIM signing and verification filters for OpenSMTPD

The OpenSMTPD documentation currently suggests using either opensmtpd-filter-dkimsign or opensmtpd-filter-rspamd for DKIM support. The former lacks functionality and requires you to compile code from some Austrian web server yourself. The latter is overdimensioned for my needs. So I’ve written my own fairly simple filters in Python.


These filters require Python 3 with dkimpy module installed. You can optionally install pyspf module as well, if you want to perform SPF verification as well.

Setting up

Your smtpd.conf file should contain directives like the following:

filter dkimverify proc-exec "/usr/local/bin/"
filter dkimsign proc-exec "/usr/local/bin/"

listen on eth0 tls filter dkimverify
listen on eth0 port 587 tls-require auth filter dkimsign

This sets up dkimverify filter for port 25 (incoming mail) and dkimsign filter for port 587 (outgoing mail). takes a single command line parameter: the host name to appear in the Authentication-Results email header. It will add a header like Authentication-Results:; dkim=pass; spf=fail (sender is to emails, this header can then be considered in further processing. takes one or multiple parameters of the form domain:selector:keyfile on the command line. Instead of configuring all domains on the command line, you can also pass this script -c /etc/mail/dkim/dkim.conf parameter, with the file /etc/mail/dkim/dkim.conf containing domain configurations in the same format, one per line. module

The module here allows implementing OpenSMTPD filters easily. It is used like following:

from opensmtpd import FilterServer

server = FilterServer()
server.register_handler('report', 'link-auth', handle_auth)
server.register_handler('filter', 'connect', handle_connect)

def handle_auth(session, username, result):
    if result == 'pass':
        print('Session {} authenticated'.format(session), file=sys.stderr)

def handle_connect(session, rdns, fcrdns, src, dest):
    if fcrdns == 'pass':
        return 'proceed'
        return 'junk'

See smtpd-filters man page for the description of the existing report events and filter requests and their parameters. The FilterServer class also exposes a convenience method register_message_filter() that allows filtering complete email messages:


  def handle_message(context, lines):
      return map(lambda line: line.replace('xyz', 'abc'), lines)

There is also method track_context(). If called during registration phase, the server will create a context object for each session and pass it to the handlers instead of the session ID.

#!/usr/bin/env python3
import argparse
import email
import re
import sys
from dkim import dkim_sign
from opensmtpd import FilterServer
def start():
parser = argparse.ArgumentParser(description='DKIM signing filter for OpenSMTPD.')
parser.add_argument('--config', '-c', metavar='config_path', help='Config file listing domain configurations (one per line)')
parser.add_argument('domains', nargs='*', metavar='domain:selector:key_path', help='Domain configuration')
args = parser.parse_args()
if args.config:
with open(args.config, 'r') as input:
for line in input:
line = line.strip()
if line:
if not
config = {}
for entry in
domain, selector, key = entry.split(':', 2)
config[domain] = {'selector': selector, 'key': key}
server = FilterServer()
server.register_message_filter(lambda context, lines: sign(config, lines))
def sign(config, lines):
parsed = email.message_from_string('\n'.join(lines))
sender = parsed.get('from', '')
match ='<([^<>]+)>', sender)
if match:
sender =
domain = re.sub(r'.*@', '', sender)
if domain in config:
with open(config[domain]['key'], 'rb') as input:
key =
signature = dkim_sign(
header, value = re.split(r':\s*', signature, 1)
parsed[header] = value
lines = parsed.as_string().splitlines(False)
return lines
if __name__ == '__main__':
#!/usr/bin/env python3
import argparse
import email
import re
import traceback
from dkim import dkim_verify
from opensmtpd import FilterServer
import spf
spf = None
def start():
parser = argparse.ArgumentParser(description='DKIM verifying filter for OpenSMTPD.')
parser.add_argument('hostname', nargs='?', default='localhost')
args = parser.parse_args()
server = FilterServer()
server.register_handler('report', 'link-identify', save_identity)
server.register_handler('report', 'tx-mail', save_sender)
server.register_message_filter(lambda context, lines: verify(server, args.hostname, context, lines))
def save_identity(context, method, identity):
context['identity'] = identity
def save_sender(context, message_id, result, sender):
context['sender'] = sender
def verify(server, hostname, context, lines):
message = '\n'.join(lines)
parsed = email.message_from_string(message)
if 'authentication-results' in parsed:
del parsed['authentication-results']
dkim_result = 'unknown'
if 'dkim-signature' in parsed:
if dkim_verify(message.encode('latin-1')):
dkim_result = 'pass'
dkim_result = 'fail'
if spf:
ip = re.sub(r':\d+$', '', context['src'])
ip = re.sub(r'^\[(.*)\]$', r'\1', ip)
spf_result, code, message = spf.check(i=ip, s=context['sender'], h=context['identity'])
clean = lambda value: re.sub(r'\s', '', value)
spf_result = '{} (sender is {}/{}) smtp.mailfrom={}'.format(
spf_result = 'unknown'
spf_result = '; spf={}'.format(spf_result)
spf_result = ''
parsed['Authentication-Results'] = '{}; dkim={}{}'.format(hostname, dkim_result, spf_result)
return parsed.as_string().splitlines(False)
if __name__ == '__main__':
#!/usr/bin/env python3
import os
import re
import sys
import traceback
class FilterServer:
"""Filter server implementation, communicates with OpenSMTPD via stdin and stdout."""
def __init__(self):
"""Handles the initial communication with OpenSMTPD."""
self._stdin = os.fdopen(sys.stdin.fileno(), 'r', encoding='latin-1', buffering=1)
self._stdout = os.fdopen(sys.stdout.fileno(), 'w', encoding='latin-1', buffering=1)
self._stderr = os.fdopen(sys.stderr.fileno(), 'w', encoding='latin-1', buffering=1)
self._handlers = {}
self._contexts = None
while self.recv() != 'config|ready':
def recv(self):
"""Low-level functionality. Receives one line from OpenSMTPD."""
return self._stdin.readline().rstrip('\r\n')
def send(self, line):
"""Low-level functionality. Sends one line to OpenSMTPD."""
print(line, file=self._stdout)
def log_exception(self):
"""Prints the current exception message to stderr."""
def register_handler(self, event, phase, handler):
"""Registers an event processor, has to be called before serve_forever().
Supported event types are 'report' and 'filter'. Handlers for report
events will receive session ID and phase-specific parameters as
arguments, no response expected. Handlers for most filter events will
receive the same parameters but have to return a result like 'proceed'
or 'reject|550 Spam'. Handler for 'data-line' filter will receive an
additional send_dataline handler as last parameter which can be called
any number of times to send lines to OpenSMTPD. Return value is ignored
for this filter.
key = '{}|{}'.format(event, phase)
if key in self._handlers:
raise Exception('Handler for {} is already registered'.format(key))
self._handlers[key] = handler
self.send('register|{}|smtp-in|{}'.format(event, phase))
def _call_handlers(self, result_handler, event, phase, session, *args):
key = '{}|{}'.format(event, phase)
if key in self._handlers:
if self._contexts is not None and key != 'report|link-connect':
session = self._contexts[session]
result_handler(self._handlers[key](session, *args))
def _filter_response(self, version, kind, session, token, payload):
if'^0\.[1-4]$', version):
self.send('|'.join([kind, token, session, payload]))
self.send('|'.join([kind, session, token, payload]))
def serve_forever(self):
"""Ends initialization phase and processes any requests coming from
OpenSMTPD. This function never returns.
def noop(result):
def send_filter_response(result):
self._filter_response(version, 'filter-result', session, token, result)
def send_dataline(line):
self._filter_response(version, 'filter-dataline', session, token, line)
while True:
line = self.recv()
count = line.count('|')
if count < 5:
elif count == 5:
event, version, timestamp, subsystem, phase, session = line.split('|', 5)
payload = None
event, version, timestamp, subsystem, phase, session, payload = line.split('|', 6)
if event == 'report':
args = []
if payload is not None:
args = payload.split('|')
if phase == 'tx-mail' and'^0\.[1-4]$', version) and len(args) == 3:
# Older protocol versions had result and sender reversed
args[1], args[2] = (args[2], args[1])
self._call_handlers(noop, event, phase, session, *args)
elif event == 'filter':
if payload is not None and '|' in payload:
token, payload = payload.split('|', 1)
token = payload
payload = None
if phase == 'data-line':
self._call_handlers(noop, event, phase, session, payload, send_dataline)
args = []
if payload is not None:
args = payload.split('|')
self._call_handlers(send_filter_response, event, phase, session, *args)
def track_context(self):
"""Calling this method before serve_forever() ensures that the first
parameter passed to all handlers will be the session context rather
than merely a session ID. A session context is a dict object containing
the following keys by default: 'session', 'rdns', 'fcrdns', 'src',
'dest'. These match the parameters of the link-connect report event.
Handlers can add and modify context object at will.
if self._contexts is not None:
self._contexts = {}
def handle_link_connect(session, rdns, fcrdns, src, dest):
self._contexts[session] = dict(session=session, rdns=rdns, fcrdns=fcrdns, src=src, dest=dest)
def handle_link_disconnect(context):
del self._contexts[context['session']]
self.register_handler('report', 'link-connect', handle_link_connect)
self.register_handler('report', 'link-disconnect', handle_link_disconnect)
def register_message_filter(self, handler):
"""Convenience method allowing to filter message bodies without
registering multiple event handlers. The handler will be called with
the session context and a list of message lines. It has to return a
filtered list of lines. Note: this will call track_context().
def handle_dataline(context, line, send_dataline):
if line != '.':
context.setdefault('message_lines', []).append(line)
lines = handler(context, context.get('message_lines', []))
for l in lines:
context.pop('message_lines', None)
context['message_error'] = True
context.pop('message_lines', None)
def handle_commit(context, *args):
if 'message_error' in context:
del context['message_error']
return 'reject|451 Internal server error'
return 'proceed'
self.register_handler('filter', 'data-line', handle_dataline)
self.register_handler('filter', 'commit', handle_commit)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment