Edit: /opt/dedrads/clean_exim.py (6390B)
#!/usr/lib/rads/venv/bin/python3
"""Python exim cleanup script to reduce the urge to run hacky oneliners"""
import argparse
import re
import subprocess
import sys
from collections import defaultdict
# compile regex once and recycle them throughout the script for efficiency
EXIM_BP_RE = re.compile(
r'\s*(?P
[0-9]+[mhd])\s+[0-9\.]+[M|K]? ' # 5m 15K
r'(?P[a-zA-Z0-9\-]+)\s+' # 1XEhho-0006AU-Tx
r'\<(?P.*)\>' #
)
def run(command: list[str] | str, shell=False) -> subprocess.CompletedProcess:
"""
Run a process with arguments.
Optionally, as a shell command
returns CompletedProcess.
"""
args = command
result = subprocess.run(
args,
capture_output=True,
shell=shell,
errors="surrogateescape",
encoding="utf-8",
check=False,
)
return result
def get_queue(exclude_bounces=False, bounces_only=False):
"""Get current exim queue as a list of dicts, each dict
containing a single message, with keys 'age', 'id', and 'sender'"""
queue = []
out = run(["exim", "-bp"])
queue_lines = out.stdout.splitlines()
for line in queue_lines:
match = EXIM_BP_RE.match(line)
if match is None:
continue
groupdict = match.groupdict()
if exclude_bounces and groupdict['sender'] == '':
continue
if bounces_only and groupdict['sender'] != '':
continue
groupdict['age'] = exim_age_to_secs(groupdict['age'])
if groupdict['age'] > 3600:
queue.append(groupdict)
return queue
def exim_age_to_secs(age):
"""convert exim age to seconds"""
conversions = {'m': 60, 'h': 3600, 'd': 86400}
# find the multiplier based on above conversions
multiplier = conversions[age[-1]]
# typecast to int and use multiplier above
return int(age.rstrip('mhd')) * multiplier
def is_boxtrapper(msg_id):
"""Determine if a message ID is a boxtrapper msg in queue"""
try:
out = run(['exim', '-Mvh', msg_id])
out.check_returncode()
head = out.stdout.strip()
except subprocess.CalledProcessError:
return False # likely no longer in queue
return 'Subject: Your email requires verification verify#' in head
def remove_msg(msg_id):
"""Here, msg_id may be a list or just one string"""
if isinstance(msg_id, list):
msg_id = ' '.join(msg_id)
try:
run(['exim', '-Mrm', msg_id]).check_returncode()
except subprocess.CalledProcessError:
pass # may have already been removed from queue or sent
def print_removed(removed):
"""Given a dict of user:count mappings, print removed"""
if len(list(removed.keys())) == 0:
print('None to remove')
else:
print('Removed:')
for user, count in removed.items():
print(' %s: %d' % (user, count))
def clean_boxtrapper():
"""Remove old boxtrapper messages from queue. The theory is that if
still stuck in queue after min_age_secs, they were likely sent from
an illigitimate email address and will never clear from queue normally"""
print('Removing old boxtrapper messages...')
removed = defaultdict(int)
queue = get_queue(exclude_bounces=True)
queue = [x for x in queue if is_boxtrapper(x['id'])]
for msg in queue:
remove_msg(msg['id'])
removed[msg['sender']] += 1
print_removed(removed)
def get_full_quota_user(msg_id):
"""Get the user for a message which is at full quota, or None"""
try:
out = run(['exim', '-Mvb', msg_id])
out.check_returncode()
body = out.stdout.strip()
except subprocess.CalledProcessError:
return None
if not ' Mailbox quota exceeded' in body:
return None
body = body.splitlines()
addr_line = -1
for line_num, line in enumerate(body):
if ' Mailbox quota exceeded' in line:
addr_line = line_num - 1
break
if addr_line >= 0:
return body[addr_line].strip()
return None
def clean_full_inbox_bounces():
"""Remove "mailbox quota exceeded" bounces from queue"""
print('Removing old full inbox bounces...')
removed = defaultdict(int)
queue = get_queue(bounces_only=True)
for msg in queue:
user = get_full_quota_user(msg['id'])
if user is None:
continue
remove_msg(msg['id'])
removed[user] += 1
print_removed(removed)
def clean_autoresponders():
"""Remove old auto-responder emails from queue which are
likely responses to spam if they have been stuck in queue"""
print('Removing old auto-responses...')
removed = defaultdict(int)
queue = get_queue(exclude_bounces=True)
for msg in queue:
try:
out = run(['exim', '-Mvh', msg['id']])
out.check_returncode()
headers = out.stdout.strip()
except subprocess.CalledProcessError:
continue
if 'X-Autorespond:' in headers and 'auto_reply' in headers:
remove_msg(msg['id'])
removed[msg['sender']] += 1
print_removed(removed)
def _parse_args():
"""Parse commandline arguments"""
parser = argparse.ArgumentParser(
description="Exim cleanup tool",
epilog="One and only one option is allowed",
)
parser.add_argument(
'-a', '--all', action='store_true', help='Do all cleanup procedures'
)
parser.add_argument(
'-b',
'--boxtrapper',
action='store_true',
help='Remove old boxtrapper messages',
)
parser.add_argument(
'-r',
'--autorespond',
action='store_true',
help='Remove old auto-responder messages',
)
parser.add_argument(
'-f',
'--full',
action='store_true',
help='Remove old bounces for full inbox',
)
args = parser.parse_args()
chosen = [key for key, value in vars(args).items() if value is True]
if len(chosen) != 1:
parser.print_help()
sys.exit(1)
return args
def main():
"""Main logic"""
args = _parse_args()
if args.boxtrapper or args.all:
clean_boxtrapper()
if args.full or args.all:
clean_full_inbox_bounces()
if args.autorespond or args.all:
clean_autoresponders()
if __name__ == '__main__':
main()