Imaplar

Imaplar [1] monitors one or more mailboxes on one or more IMAP servers. Unseen messages are passed to a user defined policy for processing.

Imaplar is intended for the automated processing of incoming mail. Use cases include anti-spam measures and automated redirection. The tool operates in two phases:

  1. Startup. Each mailbox is examined for unseen messages. Each unseen message is processed by the mailbox’s policy.
  2. Ongoing. Each mailbox is monitored for new unseen messages. When they arrive, they are processed by the policy.

Synopsis

imaplar [–config path] [server…]

–config path
Read the specified configuration file.
server
IMAP server to monitor. If no servers are specified, then servers marked as default in the configuration will be monitored.

Installation

Imaplar is released on PyPI, so all you need to do is:

$ pip install imaplar

Configuration

Imaplar is configured by a YAML file, by default the ~/.imaplar file. This can be overridden on the command line.

Caution

The configuration file contains authentication secrets and code which will be executed. It should be readable and writable only by its owner.

The configuration file defines a dictionary with the following members:

servers [dictionary, required]
A dictionary mapping IMAP server hostnames to server configurations.
policies [dictionary, required]
A dictionary mapping policy names to python scripts.
logging [dictionary, optional]
A dictionary specifying logging configuration. If present, it is passed to the python logging configuration mechanism.

Server Configuration

Each server has an associated configuration dictionary with the following members:

default [boolean, default = False]
A server flagged as default will be monitored if no servers a specified on the command line.
port [integer, default = 993 if TLS is enabled, otherwise 143]
The IMAP server port.
tls [dictionary, optional]
TLS configuration.
authentication [dictionary, optional]
Authentication configuration.
poll [integer, default = 60]
Server polling interval in seconds. Only used when IDLE is not supported by the server.
idle [integer, default = 900]
Server idling interval in seconds. Only used when IDLE is supported by the server.
mailboxes [dictionary, required]
A mapping of mailbox names to policy names. Each mailbox will be monitored, with messages passed to the specified policy.
parameters [dictionary, optional]
Per-server parameters that will be passed to the policy.

TLS Configuration

A TLS configuration dictionary has the following members:

mode [string, default = “enabled”]
The TLS mode. This may be “enabled”, “disabled” or “starttls”.
verify_mode [string, default = “required”]
Certificate verification mode. This may be “none”, “optional” or “required”. Set this to “optional” or “none” if you are using a self signed certificate.
check_hostname [boolean, default = true]
Validate that the server name matches the certificate. Set this to false if there is a mismatch (such as connecting to localhost).
cafile [string, optional]
The path to a file containing concatenated PEM certificates. If defined, the system default certificate store is ignored.
capath [string, optional]
The path to a directory containing PEM certifcates as per the OpenSSL layout. If defined, the system default certificate store is ignored.
cadata [string, optional]
A string containing concatenated PEM certificates. If defined, the system default certificate store is ignored.

Authentication Configuration

An authetication configuration dictionary has the following members:

method [string, required]
The authentication method to use. Must be one of “login”, “plain” or “oauth2”.
login_username [string, required when method is “login”]
The login username.
login_password [string, required when method is “login”]
The login password.
plain_identity [string, required when method is “plain”]
The plain identity.
plain_password [string, required when method is “plain”]
The plain password.
plain_authorization_identity [string, optional when method is “plain”]
The plain authorization identity.
oauth2_user [string, required when method is “oauth2”]
The oauth2 user.
oauth2_access_token [string, required when method is “oauth2”]
The oauth2 access token.
oauth2_mech [string, default = “XOAUTH2” when method is “oauth2”]
The oauth2 mechanism.
oauth2_vendor [string, optional when method is “oauth2”]
The oauth2 vendor.

Example Configuration

A simple example configuration file looks like this:

---
servers:
  mail.example.com:
    authentication:
      method: login
      username: myname
      password: mypassword
    mailboxes:
      inbox: mypolicy

policies:
  mypolicy: |
    # this policy just logs a message
    import logging
    logging.info("Handled {}/{}".format(mailbox, message))

Systemd User Service (Optional)

If you are running Systemd, you may configure a user service in order to run imaplar automatically.

  1. Create the file ~/.config/systemd/user/imaplar.service:

    [Unit]
    Description = Imaplar IMAP monitoring service
    
    [Service]
    ExecStart = <path-to-imap-command>
    Restart = always
    
    [Install]
    WantedBy = default.target
    
  2. Enable and start the service with these shell commands:

    $ systemctl --user enable imaplar
    $ systemctl --user start imaplar
    
  3. If you want the service to keep running when you are logged out, run the following command as root:

    # loginctl enable-linger <your-username>
    

Writing Policies

Each policy is a python script. The following global variables are provided:

  • client: an instance of imapclient.IMAPClient, connected to the server
  • mailbox: the name of the monitored mailbox
  • message: the message id
  • parameters: parameters specified in the server configuration

Note

A policy script should not assume that the currently selected mailbox (if any) is the monitored mailbox.

The imaplar.policy Module

This module provides convenience classes and methods to help make writing policies easier.

class imaplar.policy.Originators(envelope)[source]

Bases: set

Envelope originator addresses.

Parameters:envelope (imapclient.response_types.Envelope) – message envelope
class imaplar.policy.Recipients(envelope)[source]

Bases: set

Envelope recipient addresses.

Parameters:envelope (imapclient.response_types.Envelope) – message envelope
class imaplar.policy.Query[source]

Bases: list

A list of IMAP search criteria.

Parameters:criteria (iterable of strings) – search criteria
__call__(client, *mailboxes)[source]

Generate message ids by executing query.

Parameters:
  • client (imapclient.IMAPClient) – imap client
  • mailboxes (strings) – mailbox names
Returns:

message ids

Return type:

generator of ints

__and__(query)[source]

AND queries together.

Parameters:query (Query) – other query
Returns:a new query
Return type:Query
__or__(query)[source]

OR queries together.

Parameters:query (Query) – other query
Returns:a new query
Return type:Query
__not__()[source]

NOT this query.

Returns:a new query
Return type:Query
class imaplar.policy.ToQuery(addresses)[source]

Bases: imaplar.policy.Query

A query for “To” addresses.

Parameters:addresses (iterable of imapclient.response_types.Address objects) – match any of these addresses
class imaplar.policy.CcQuery(addresses)[source]

Bases: imaplar.policy.Query

A query for “Cc” addresses.

Parameters:addresses (iterable of imapclient.response_types.Address objects) – match any of these addresses
class imaplar.policy.BccQuery(addresses)[source]

Bases: imaplar.policy.Query

A query for “Bcc” addresses.

Parameters:addresses (iterable of imapclient.response_types.Address objects) – match any of these addresses
class imaplar.policy.FromQuery(addresses)[source]

Bases: imaplar.policy.Query

A query for “From” addresses.

Parameters:addresses (iterable of imapclient.response_types.Address objects) – match any of these addresses
class imaplar.policy.SenderQuery(addresses)[source]

Bases: imaplar.policy.Query

A query for “Sender” addresses.

Parameters:addresses (iterable of imapclient.response_types.Address objects) – match any of these addresses
class imaplar.policy.ReplyToQuery(addresses)[source]

Bases: imaplar.policy.Query

A query for “Reply-To” addresses.

Parameters:addresses (iterable of imapclient.response_types.Address objects) – match any of these addresses
class imaplar.policy.OriginatorQuery(addresses)[source]

Bases: imaplar.policy.Query

A query for “From”, “Sender” and “Reply-To” addresses.

Parameters:addresses (iterable of imapclient.response_types.Address objects) – match any of these addresses
class imaplar.policy.RecipientQuery(addresses)[source]

Bases: imaplar.policy.Query

A query for “To”, “Cc” and “Bcc” addresses.

Parameters:addresses (iterable of imapclient.response_types.Address objects) – match any of these addresses
imaplar.policy.fetch_envelope(client, mailbox, message)[source]

Fetch the envelope of a message.

Parameters:
  • client (imapclient.IMAPClient) – imap client
  • mailbox (string) – mailbox name
  • message (int) – message id
Returns:

an envelope

Return type:

imapclient.response_types.Envelope

imaplar.policy.move_message(client, mailbox, message, to_mailbox)[source]

Move a message to a different mailbox.

Uses the IMAP MOVE capability if available, otherwise it copies the message to the destination and then deletes the original.

Parameters:
  • client (imapclient.IMAPClient) – imap client
  • mailbox (string) – source mailbox name
  • message (int) – message id
  • to_mailbox (string) – destination mailbox name

Example Policy: Spamalot

This policy assumes nearly all email is spam. Only messages from known originators are retained.

from imaplar.policy import *
import logging
import re

# per server parameters
outbox = parameters.get("outbox", "Sent")
spambox = parameters.get("spambox", "Spam")

# get envelope and extract originator addresses
envelope = fetch_envelope(client, mailbox, message)
originators = Originators(envelope)

# default is spam
spam = True

# have we sent mail to at least one of these originators?
query = RecipientQuery(originators)
if any(query(client, outbox)):
    spam = False

# have we read mail from at least one of these originators?
query = Query(["SEEN"]) & OriginatorQuery(originators)
if any(query(client, mailbox)):
    spam = False

# any noreply originator implies spam
noreply = re.compile(r"no.{0,2}reply", re.IGNORECASE)
if any(noreply.search(str(a)) for a in originators):
    spam = True

# marked as spam?
if spam:
    logging.info("{}({})/{}/{}: moving to {}".format(
        client.host, client.port, mailbox, message, spambox))
    move_message(client, mailbox, message, spambox)

Licenses

Copyright (C) 2017-2022 Michael Paddon

This program, including this manual, is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with this program. If not, see <http://www.gnu.org/licenses/>.

The Imaplar logo is derived from an image licensed under the Creative Commons Attribution-Share Alike 3.0 Unported license. Attribution: Luis García.

Footnotes

[1]The Lares (singular Lar) were ancient Roman guardian deities.