Merge pull request 'Improve logging and configuration' (#65) from logging-and-config into master

- Replace custom logging code with "logging" package.
- Unify access to configuration and extract to "lacre.config" package.
- Introduce a new configuration file (with a sample included) to control how and where Lacre writes diagnostic output.
- Update sample configuration.

Reviewed-on: #65
This commit is contained in:
pfm 2022-05-13 20:01:10 +00:00
commit 5639d8e5b6
13 changed files with 421 additions and 258 deletions

View File

@ -14,7 +14,7 @@ These instructions are based on an installation on an Ubuntu 14.04 LTS virtual m
## Install GPG-Mailgate ## Install GPG-Mailgate
### Requirements ### Requirements
- Python 2.X is already installed (GPG-Mailgate is not Python 3 compatible) - Python 3.X is already installed
- Postfix is already installed and configured. It is recommended that you have already tested your configuration so we can exclude this as a main cause of problems - Postfix is already installed and configured. It is recommended that you have already tested your configuration so we can exclude this as a main cause of problems
- GnuPG is already installed and configured - GnuPG is already installed and configured
@ -39,11 +39,13 @@ These instructions are based on an installation on an Ubuntu 14.04 LTS virtual m
chown nobody:nogroup /usr/local/bin/gpg-mailgate.py chown nobody:nogroup /usr/local/bin/gpg-mailgate.py
chmod u+x /usr/local/bin/gpg-mailgate.py chmod u+x /usr/local/bin/gpg-mailgate.py
5. Place the `GnuPG` directory in `/usr/local/lib/python2.7/dist-packages` (replace 2.7 with your Python 2 version) 5. Place the `GnuPG` directory in `/usr/local/lib/python3.x/dist-packages` (replace 3.x with your Python version)
6. Configure `/etc/gpg-mailgate.conf` based on the provided `gpg-mailgate.conf.sample`. Change the settings according to your configuration. If you follow this guide and have a standard configuration for postfix, you don't need to change much. 6. Configure `/etc/gpg-mailgate.conf` based on the provided `gpg-mailgate.conf.sample`. Change the settings according to your configuration. If you follow this guide and have a standard configuration for postfix, you don't need to change much.
7. Add the following to the end of `/etc/postfix/master.cf` 7. Configure logging by copying `gpg-lacre-logging.conf.sample` to `/etc/gpg-lacre-logging.conf` and editing it according to your needs. The path to this file is included in `[logging]` section of `gpg-mailgate.conf` file, so if you place it somewhere else, make sure to update the path too. See also: [Configuration file format](https://docs.python.org/3/library/logging.config.html#configuration-file-format).
8. Add the following to the end of `/etc/postfix/master.cf`
gpg-mailgate unix - n n - - pipe gpg-mailgate unix - n n - - pipe
flags= user=nobody argv=/usr/local/bin/gpg-mailgate.py ${recipient} flags= user=nobody argv=/usr/local/bin/gpg-mailgate.py ${recipient}
@ -60,15 +62,15 @@ These instructions are based on an installation on an Ubuntu 14.04 LTS virtual m
If you use Postfix versions from 2.5 onwards, it is recommended to change `${recipient}` to `${original_recipient}` in line two of the lines above. If you use Postfix versions from 2.5 onwards, it is recommended to change `${recipient}` to `${original_recipient}` in line two of the lines above.
8. Add the following line to `/etc/postfix/main.cf` 9. Add the following line to `/etc/postfix/main.cf`
content_filter = gpg-mailgate content_filter = gpg-mailgate
9. Optional: GPG can automatically download new public keys for automatic signature verification. To enable automatic create the file `/var/gpgmailgate/.gnupg/gpg.conf`. Add the following line to the file: 10. Optional: GPG can automatically download new public keys for automatic signature verification. To enable automatic create the file `/var/gpgmailgate/.gnupg/gpg.conf`. Add the following line to the file:
keyserver-options auto-key-retrieve keyserver-options auto-key-retrieve
10. Restart Postfix 11. Restart Postfix
You are now ready to go. To add a public key for encryption just use the following command: You are now ready to go. To add a public key for encryption just use the following command:
@ -112,10 +114,10 @@ You also can remove a private key by using the following command. Replace `user@
- A webserver is installed and reachable - A webserver is installed and reachable
- The webserver is able to handle PHP scripts - The webserver is able to handle PHP scripts
- MySQL is installed - MySQL is installed
- Python 2.X is already installed - Python 3.X is already installed
### Installation ### Installation
All files you need can be found in the [gpg-mailgate-web] (gpg-mailgate-web/) directory. All files you need can be found in the [gpg-mailgate-web](gpg-mailgate-web/) directory.
1. Install the Python-mysqldb and Python-markdown modules: 1. Install the Python-mysqldb and Python-markdown modules:
@ -127,7 +129,7 @@ All files you need can be found in the [gpg-mailgate-web] (gpg-mailgate-web/) di
4. Edit the config file located at `/etc/gpg-mailgate.conf`. Set `enabled = yes` in `[database]` and fill in the necessary settings for the database connection. 4. Edit the config file located at `/etc/gpg-mailgate.conf`. Set `enabled = yes` in `[database]` and fill in the necessary settings for the database connection.
5. Copy the files located in the [public_html] (gpg-mailgate-web/public_html) directory onto your webserver. They can also be placed in a subdirectory on your webserver. 5. Copy the files located in the [public_html](gpg-mailgate-web/public_html) directory onto your webserver. They can also be placed in a subdirectory on your webserver.
6. On your webserver move the `config.sample.php` file to `config.php` and edit the configuration file. 6. On your webserver move the `config.sample.php` file to `config.php` and edit the configuration file.
@ -135,7 +137,7 @@ All files you need can be found in the [gpg-mailgate-web] (gpg-mailgate-web/) di
mkdir -p /var/gpgmailgate/cron_templates mkdir -p /var/gpgmailgate/cron_templates
8. Copy the templates found in the [cron_templates] (cron_templates/) directory into the newly created directory and transfer ownership: 8. Copy the templates found in the [cron_templates](cron_templates/) directory into the newly created directory and transfer ownership:
chown -R nobody:nogroup /var/gpgmailgate/cron_templates chown -R nobody:nogroup /var/gpgmailgate/cron_templates
@ -151,7 +153,7 @@ All files you need can be found in the [gpg-mailgate-web] (gpg-mailgate-web/) di
11. Test your installation. 11. Test your installation.
### GPG-Mailgate-Web as keyserver ### GPG-Mailgate-Web as keyserver
GPG-Mailgate-Web can also be used as a keyserver. For more information have a look at GPG-Mailgate-Web's [readme] (gpg-mailgate-web/README). GPG-Mailgate-Web can also be used as a keyserver. For more information have a look at GPG-Mailgate-Web's [readme](gpg-mailgate-web/README).
## Install Register-handler ## Install Register-handler
### Requirements ### Requirements
@ -168,7 +170,7 @@ GPG-Mailgate-Web can also be used as a keyserver. For more information have a lo
mkdir -p /var/gpgmailgate/register_templates mkdir -p /var/gpgmailgate/register_templates
3. Copy the templates found in the [register_templates] (register_templates/) directory into the newly created directory and transfer ownership: 3. Copy the templates found in the [register_templates](register_templates/) directory into the newly created directory and transfer ownership:
chown -R nobody:nogroup /var/gpgmailgate/register_templates chown -R nobody:nogroup /var/gpgmailgate/register_templates

View File

@ -0,0 +1,57 @@
# Example configuration for Lacre logging. If you don't intend to change the
# log format, you can just keep this file unchanged.
# HANDLERS:
#
# Two main targets for log entries are defined here: syslog and a plain text
# log file. They are available as "handlers" named "syslog" and "lacrelog"
# respectively.
[loggers]
keys=root
[logger_root]
level=NOTSET
# Comma-separated handler names, see HANDLERS note at the top.
handlers=syslog
[handlers]
# Comma-separated handler names, see HANDLERS note at the top.
keys=syslog
[formatters]
keys=postfixfmt
#
# By default, include messages from all log levels up to DEBUG.
# However, productive systems may use something less verbose, like
# WARN or even ERROR.
#
[handler_lacrelog]
class=FileHandler
level=DEBUG
formatter=postfixfmt
args=('test/logs/lacre.log', 'a+')
# You may want to change the second argument (handlers.SysLogHandler.LOG_MAIL)
# to change the syslog facility used to record messages from Lacre.
#
# Options you can consider are "localX" facilities, available under names from
# handlers.SysLogHandler.LOG_LOCAL0 to handlers.SysLogHandler.LOG_LOCAL7.
#
# Please refer to your syslog configuration for details on how to separate
# records from different facilities.
[handler_syslog]
class=handlers.SysLogHandler
level=INFO
formatter=postfixfmt
args=('/dev/log', handlers.SysLogHandler.LOG_MAIL)
#
# Default Postfix log format.
#
[formatter_postfixfmt]
format=%(asctime)s %(module)s[%(process)d]: %(message)s
datefmt=%b %e %H:%M:%S
style=%
validate=True

View File

@ -30,20 +30,9 @@ import os
from email.mime.text import MIMEText from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart from email.mime.multipart import MIMEMultipart
# Environment variable name we read to retrieve configuration path. This is to import logging
# enable non-root users to set up and run GPG Mailgate and to make the software import lacre
# testable. import lacre.config as conf
CONFIG_PATH_ENV = "GPG_MAILGATE_CONFIG"
def appendLog(msg):
print(msg)
if 'logging' in cfg and 'file' in cfg['logging']:
if cfg['logging'].get('file') == "syslog":
syslog.syslog(syslog.LOG_INFO | syslog.LOG_MAIL, msg)
else:
logfile = open(cfg['logging']['file'], 'a')
logfile.write(msg + "\n")
logfile.close()
def load_file(name): def load_file(name):
f = open(name) f = open(name)
@ -52,32 +41,34 @@ def load_file(name):
return data return data
def authenticate_maybe(smtp): def authenticate_maybe(smtp):
if 'smtp' in cfg and 'enabled' in cfg['smtp'] and cfg['smtp']['enabled'] == 'true': if conf.config_item_equals('smtp', 'enabled', 'true'):
smtp.connect(cfg['smtp']['host'],cfg['smtp']['port']) LOG.debug(f"Connecting to {conf.get_item('smtp', 'host')}:{conf.get_item('smtp', 'port')}")
smtp.connect(conf.get_item('smtp', 'host'), conf.get_item('smtp', 'port'))
smtp.ehlo() smtp.ehlo()
if 'starttls' in cfg['smtp'] and cfg['smtp']['starttls'] == 'true': if conf.config_item_equals('smtp', 'starttls', 'true'):
LOG.debug("StartTLS enabled")
smtp.starttls() smtp.starttls()
smtp.ehlo() smtp.ehlo()
smtp.login(cfg['smtp']['username'], cfg['smtp']['password']) smtp.login(conf.get_item('smtp', 'username'), conf.get_item('smtp', 'password'))
def send_msg( mailsubject, messagefile, recipients = None ): def send_msg( mailsubject, messagefile, recipients = None ):
mailbody = load_file( cfg['cron']['mail_templates'] + "/" + messagefile) mailbody = load_file( conf.get_item('cron', 'mail_templates') + "/" + messagefile).read()
msg = MIMEMultipart("alternative") msg = MIMEMultipart("alternative")
msg["From"] = cfg['cron']['notification_email'] msg["From"] = conf.get_item('cron', 'notification_email')
msg["To"] = recipients msg["To"] = recipients
msg["Subject"] = mailsubject msg["Subject"] = mailsubject
msg.attach(MIMEText(mailbody, 'plain')) msg.attach(MIMEText(mailbody, 'plain'))
msg.attach(MIMEText(markdown.markdown(mailbody), 'html')) msg.attach(MIMEText(markdown.markdown(mailbody), 'html'))
if 'relay' in cfg and 'host' in cfg['relay'] and 'enc_port' in cfg['relay']: if conf.config_item_set('relay', 'host') and conf.config_item_set('relay', 'enc_port'):
relay = (cfg['relay']['host'], int(cfg['relay']['enc_port'])) relay = (conf.get_item('relay', 'host'), int(conf.get_item('relay', 'enc_port')))
smtp = smtplib.SMTP(relay[0], relay[1]) smtp = smtplib.SMTP(relay[0], relay[1])
authenticate_maybe(smtp) authenticate_maybe(smtp)
smtp.sendmail( cfg['cron']['notification_email'], recipients, msg.as_string() ) smtp.sendmail( conf.get_item('cron', 'notification_email'), recipients, msg.as_string() )
else: else:
appendLog("Could not send mail due to wrong configuration") LOG.info("Could not send mail due to wrong configuration")
def setup_db_connection(url): def setup_db_connection(url):
engine = sqlalchemy.create_engine(url) engine = sqlalchemy.create_engine(url)
@ -98,49 +89,52 @@ def define_db_schema():
# Read configuration from /etc/gpg-mailgate.conf # Read configuration from /etc/gpg-mailgate.conf
_cfg = RawConfigParser() conf.load_config()
_cfg.read(os.getenv(CONFIG_PATH_ENV, '/etc/gpg-mailgate.conf'))
cfg = dict()
for sect in _cfg.sections():
cfg[sect] = dict()
for (name, value) in _cfg.items(sect):
cfg[sect][name] = value
if 'database' in cfg and 'enabled' in cfg['database'] and cfg['database']['enabled'] == 'yes' and 'url' in cfg['database']: lacre.init_logging(conf.get_item('logging', 'config'))
(engine, conn) = setup_db_connection(cfg["database"]["url"]) LOG = logging.getLogger(__name__)
if conf.config_item_equals('database', 'enabled', 'yes') and conf.config_item_set('database', 'url'):
(engine, conn) = setup_db_connection(conf.get_item("database", "url"))
(gpgmw_keys) = define_db_schema() (gpgmw_keys) = define_db_schema()
selq = select(gpgmw_keys.c.publickey, gpgmw_keys.c.id, gpgmw_keys.c.email)\ selq = select(gpgmw_keys.c.publickey, gpgmw_keys.c.id, gpgmw_keys.c.email)\
.where(and_(gpgmw_keys.c.status == 0, gpgmw_keys.c.confirm == ""))\ .where(and_(gpgmw_keys.c.status == 0, gpgmw_keys.c.confirm == ""))\
.limit(100) .limit(100)
LOG.debug(f"Retrieving keys to be processed: {selq}")
result_set = conn.execute(selq) result_set = conn.execute(selq)
for row in result_set: for row in result_set:
# delete any other public keys associated with this confirmed email address # delete any other public keys associated with this confirmed email address
delq = delete(gpgmw_keys).where(and_(gpgmw_keys.c.email == row[2], gpgmw_keys.c.id != row[1])) delq = delete(gpgmw_keys).where(and_(gpgmw_keys.c.email == row[2], gpgmw_keys.c.id != row[1]))
LOG.debug(f"Deleting public keys associated with confirmed email: {delq}")
conn.execute(delq) conn.execute(delq)
GnuPG.delete_key(cfg['gpg']['keyhome'], row[2]) GnuPG.delete_key(conf.get_item('gpg', 'keyhome'), row[2])
appendLog('Deleted key for <' + row[2] + '> via import request') LOG.info('Deleted key for <' + row[2] + '> via import request')
if row[0].strip(): # we have this so that user can submit blank key to remove any encryption if row[0].strip(): # we have this so that user can submit blank key to remove any encryption
if GnuPG.confirm_key(row[0], row[2]): if GnuPG.confirm_key(row[0], row[2]):
GnuPG.add_key(cfg['gpg']['keyhome'], row[0]) # import the key to gpg GnuPG.add_key(conf.get_item('gpg', 'keyhome'), row[0]) # import the key to gpg
modq = gpgmw_keys.update().where(gpgmw_keys.c.id == row[1]).values(status = 1) modq = gpgmw_keys.update().where(gpgmw_keys.c.id == row[1]).values(status = 1)
LOG.debug(f"Key imported, updating key: {modq}")
conn.execute(modq) # mark key as accepted conn.execute(modq) # mark key as accepted
appendLog('Imported key from <' + row[2] + '>') LOG.warning('Imported key from <' + row[2] + '>')
if 'send_email' in cfg['cron'] and cfg['cron']['send_email'] == 'yes': if conf.config_item_equals('cron', 'send_email', 'yes'):
send_msg( "PGP key registration successful", "registrationSuccess.md", row[2] ) send_msg( "PGP key registration successful", "registrationSuccess.md", row[2] )
else: else:
delq = delete(gpgmw_keys).where(gpgmw_keys.c.id == row[1]) delq = delete(gpgmw_keys).where(gpgmw_keys.c.id == row[1])
LOG.debug(f"Cannot confirm key, deleting it: {delq}")
conn.execute(delq) # delete key conn.execute(delq) # delete key
appendLog('Import confirmation failed for <' + row[2] + '>') LOG.warning('Import confirmation failed for <' + row[2] + '>')
if 'send_email' in cfg['cron'] and cfg['cron']['send_email'] == 'yes': if conf.config_item_equals('cron', 'send_email', 'yes'):
send_msg( "PGP key registration failed", "registrationError.md", row[2] ) send_msg( "PGP key registration failed", "registrationError.md", row[2] )
else: else:
# delete key so we don't continue processing it # delete key so we don't continue processing it
delq = delete(gpgmw_keys).where(gpgmw_keys.c.id == row[1]) delq = delete(gpgmw_keys).where(gpgmw_keys.c.id == row[1])
LOG.debug(f"Deleting key: {delq}")
conn.execute(delq) conn.execute(delq)
if 'send_email' in cfg['cron'] and cfg['cron']['send_email'] == 'yes': if conf.config_item_equals('cron', 'send_email', 'yes'):
send_msg( "PGP key deleted", "keyDeleted.md", row[2]) send_msg( "PGP key deleted", "keyDeleted.md", row[2])
# delete keys # delete keys
@ -148,9 +142,11 @@ if 'database' in cfg and 'enabled' in cfg['database'] and cfg['database']['enabl
stat2_result_set = conn.execute(stat2q) stat2_result_set = conn.execute(stat2q)
for row in stat2_result_set: for row in stat2_result_set:
GnuPG.delete_key(cfg['gpg']['keyhome'], row[0]) GnuPG.delete_key(conf.get_item('gpg', 'keyhome'), row[0])
delq = delete(gpgmw_keys).where(gpgmw_keys.c.id == row[1]) delq = delete(gpgmw_keys).where(gpgmw_keys.c.id == row[1])
LOG.debug(f"Deleting keys that have already been processed: {delq}")
conn.execute(delq) conn.execute(delq)
appendLog('Deleted key for <' + row[0] + '>') LOG.info('Deleted key for <' + row[0] + '>')
else: else:
print("Warning: doing nothing since database settings are not configured!") print("Warning: doing nothing since database settings are not configured!")
LOG.error("Warning: doing nothing since database settings are not configured!")

View File

@ -66,9 +66,9 @@ notification_email = gpg-mailgate@yourdomain.tld
mail_templates = /var/gpgmailgate/cron_templates mail_templates = /var/gpgmailgate/cron_templates
[logging] [logging]
# For logging to syslog. 'file = syslog', otherwise use path to the file. # path to the logging configuration; see documentation for details:
file = syslog # https://docs.python.org/3/library/logging.config.html#logging-config-fileformat
verbose = yes config = /etc/gpg-lacre-logging.conf
[relay] [relay]
# the relay settings to use for Postfix # the relay settings to use for Postfix

View File

@ -39,44 +39,18 @@ import traceback
from M2Crypto import BIO, Rand, SMIME, X509 from M2Crypto import BIO, Rand, SMIME, X509
from email.mime.message import MIMEMessage from email.mime.message import MIMEMessage
# Environment variable name we read to retrieve configuration path. This is to import logging
# enable non-root users to set up and run GPG Mailgate and to make the software import lacre
# testable. import lacre.config as conf
CONFIG_PATH_ENV = "GPG_MAILGATE_CONFIG"
# Read configuration from /etc/gpg-mailgate.conf
_cfg = RawConfigParser()
_cfg.read(os.getenv(CONFIG_PATH_ENV, '/etc/gpg-mailgate.conf'))
cfg = dict()
for sect in _cfg.sections():
cfg[sect] = dict()
for (name, value) in _cfg.items(sect):
cfg[sect][name] = value
def log( msg ):
if 'logging' in cfg and 'file' in cfg['logging']:
if cfg['logging'].get('file') == "syslog":
syslog.syslog(syslog.LOG_INFO | syslog.LOG_MAIL, msg)
else:
logfile = open(cfg['logging']['file'], 'a')
logfile.write(msg + "\n")
logfile.close()
verbose = 'logging' in cfg and 'verbose' in cfg['logging'] and cfg['logging'].get('verbose') == 'yes'
# Read e-mail from stdin
raw = sys.stdin.read()
raw_message = email.message_from_string( raw )
from_addr = raw_message['From']
to_addrs = sys.argv[1:]
def gpg_encrypt( raw_message, recipients ): def gpg_encrypt( raw_message, recipients ):
global LOG
if not get_bool_from_cfg('gpg', 'keyhome'): if not conf.config_item_set('gpg', 'keyhome'):
log("No valid entry for gpg keyhome. Encryption aborted.") LOG.error("No valid entry for gpg keyhome. Encryption aborted.")
return recipients return recipients
keys = GnuPG.public_keys( cfg['gpg']['keyhome'] ) keys = GnuPG.public_keys( conf.get_item('gpg', 'keyhome') )
for fingerprint in keys: for fingerprint in keys:
keys[fingerprint] = sanitize_case_sense(keys[fingerprint]) keys[fingerprint] = sanitize_case_sense(keys[fingerprint])
@ -86,17 +60,17 @@ def gpg_encrypt( raw_message, recipients ):
for to in recipients: for to in recipients:
# Check if recipient is in keymap # Check if recipient is in keymap
if get_bool_from_cfg('enc_keymap', to): if conf.config_item_set('enc_keymap', to):
log("Encrypt keymap has key '%s'" % cfg['enc_keymap'][to] ) LOG.info("Encrypt keymap has key '%s'" % conf.get_item('enc_keymap', to) )
# Check we've got a matching key! # Check we've got a matching key!
if cfg['enc_keymap'][to] in keys: if conf.get_item('enc_keymap', to) in keys:
gpg_to.append( (to, cfg['enc_keymap'][to]) ) gpg_to.append( (to, conf.get_item('enc_keymap', to)) )
continue continue
else: else:
log("Key '%s' in encrypt keymap not found in keyring for email address '%s'." % (cfg['enc_keymap'][to], to)) LOG.info("Key '%s' in encrypt keymap not found in keyring for email address '%s'." % (conf.get_item('enc_keymap', to), to))
# Check if key in keychain is present # Check if key in keychain is present
if to in keys.values() and not get_bool_from_cfg('default', 'enc_keymap_only', 'yes'): if to in keys.values() and not conf.config_item_equals('default', 'enc_keymap_only', 'yes'):
gpg_to.append( (to, to) ) gpg_to.append( (to, to) )
continue continue
@ -104,23 +78,22 @@ def gpg_encrypt( raw_message, recipients ):
splitted_to = to.split('@') splitted_to = to.split('@')
if len(splitted_to) > 1: if len(splitted_to) > 1:
domain = splitted_to[1] domain = splitted_to[1]
if get_bool_from_cfg('enc_domain_keymap', domain): if conf.config_item_set('enc_domain_keymap', domain):
log("Encrypt domain keymap has key '%s'" % cfg['enc_dec_keymap'][domain] ) LOG.info("Encrypt domain keymap has key '%s'" % conf.get_item('enc_dec_keymap', domain) )
# Check we've got a matching key! # Check we've got a matching key!
if cfg['enc_domain_keymap'][domain] in keys: if conf.get_item('enc_domain_keymap', domain) in keys:
log("Using default domain key for recipient '%s'" % to) LOG.info("Using default domain key for recipient '%s'" % to)
gpg_to.append( (to, cfg['enc_domain_keymap'][domain]) ) gpg_to.append( (to, conf.get_item('enc_domain_keymap', domain)) )
continue continue
else: else:
log("Key '%s' in encrypt domain keymap not found in keyring for email address '%s'." % (cfg['enc_domain_keymap'][domain], to)) LOG.info("Key '%s' in encrypt domain keymap not found in keyring for email address '%s'." % (conf.get_item('enc_domain_keymap', domain), to))
# At this point no key has been found # At this point no key has been found
if verbose: LOG.debug("Recipient (%s) not in PGP domain list for encrypting." % to)
log("Recipient (%s) not in PGP domain list for encrypting." % to)
ungpg_to.append(to) ungpg_to.append(to)
if gpg_to != list(): if gpg_to:
log("Encrypting email to: %s" % ' '.join( x[0] for x in gpg_to )) LOG.info("Encrypting email to: %s" % ' '.join( x[0] for x in gpg_to ))
# Getting PGP style for recipient # Getting PGP style for recipient
gpg_to_smtp_mime = list() gpg_to_smtp_mime = list()
@ -131,30 +104,30 @@ def gpg_encrypt( raw_message, recipients ):
for rcpt in gpg_to: for rcpt in gpg_to:
# Checking pre defined styles in settings first # Checking pre defined styles in settings first
if get_bool_from_cfg('pgp_style', rcpt[0], 'mime'): if conf.config_item_equals('pgp_style', rcpt[0], 'mime'):
gpg_to_smtp_mime.append(rcpt[0]) gpg_to_smtp_mime.append(rcpt[0])
gpg_to_cmdline_mime.extend(rcpt[1].split(',')) gpg_to_cmdline_mime.extend(rcpt[1].split(','))
elif get_bool_from_cfg('pgp_style', rcpt[0], 'inline'): elif conf.config_item_equals('pgp_style', rcpt[0], 'inline'):
gpg_to_smtp_inline.append(rcpt[0]) gpg_to_smtp_inline.append(rcpt[0])
gpg_to_cmdline_inline.extend(rcpt[1].split(',')) gpg_to_cmdline_inline.extend(rcpt[1].split(','))
else: else:
# Log message only if an unknown style is defined # Log message only if an unknown style is defined
if get_bool_from_cfg('pgp_style', rcpt[0]): if conf.config_item_set('pgp_style', rcpt[0]):
log("Style %s for recipient %s is not known. Use default as fallback." % (cfg['pgp_style'][rcpt[0]], rcpt[0])) LOG.info("Style %s for recipient %s is not known. Use default as fallback." % (conf.get_item("pgp_style", rcpt[0]), rcpt[0]))
# If no style is in settings defined for recipient, use default from settings # If no style is in settings defined for recipient, use default from settings
if get_bool_from_cfg('default', 'mime_conversion', 'yes'): if conf.config_item_equals('default', 'mime_conversion', 'yes'):
gpg_to_smtp_mime.append(rcpt[0]) gpg_to_smtp_mime.append(rcpt[0])
gpg_to_cmdline_mime.extend(rcpt[1].split(',')) gpg_to_cmdline_mime.extend(rcpt[1].split(','))
else: else:
gpg_to_smtp_inline.append(rcpt[0]) gpg_to_smtp_inline.append(rcpt[0])
gpg_to_cmdline_inline.extend(rcpt[1].split(',')) gpg_to_cmdline_inline.extend(rcpt[1].split(','))
if gpg_to_smtp_mime != list(): if gpg_to_smtp_mime:
# Encrypt mail with PGP/MIME # Encrypt mail with PGP/MIME
raw_message_mime = copy.deepcopy(raw_message) raw_message_mime = copy.deepcopy(raw_message)
if get_bool_from_cfg('default', 'add_header', 'yes'): if conf.config_item_equals('default', 'add_header', 'yes'):
raw_message_mime['X-GPG-Mailgate'] = 'Encrypted by GPG Mailgate' raw_message_mime['X-GPG-Mailgate'] = 'Encrypted by GPG Mailgate'
if 'Content-Transfer-Encoding' in raw_message_mime: if 'Content-Transfer-Encoding' in raw_message_mime:
@ -167,11 +140,11 @@ def gpg_encrypt( raw_message, recipients ):
send_msg( raw_message_mime.as_string(), gpg_to_smtp_mime ) send_msg( raw_message_mime.as_string(), gpg_to_smtp_mime )
if gpg_to_smtp_inline != list(): if gpg_to_smtp_inline:
# Encrypt mail with PGP/INLINE # Encrypt mail with PGP/INLINE
raw_message_inline = copy.deepcopy(raw_message) raw_message_inline = copy.deepcopy(raw_message)
if get_bool_from_cfg('default', 'add_header', 'yes'): if conf.config_item_equals('default', 'add_header', 'yes'):
raw_message_inline['X-GPG-Mailgate'] = 'Encrypted by GPG Mailgate' raw_message_inline['X-GPG-Mailgate'] = 'Encrypted by GPG Mailgate'
if 'Content-Transfer-Encoding' in raw_message_inline: if 'Content-Transfer-Encoding' in raw_message_inline:
@ -247,21 +220,20 @@ def encrypt_all_payloads_mime( message, gpg_to_cmdline ):
return [ submsg1, encrypt_payload(submsg2, gpg_to_cmdline, check_nested) ] return [ submsg1, encrypt_payload(submsg2, gpg_to_cmdline, check_nested) ]
def encrypt_payload( payload, gpg_to_cmdline, check_nested = True ): def encrypt_payload( payload, gpg_to_cmdline, check_nested = True ):
global LOG
raw_payload = payload.get_payload(decode=True) raw_payload = payload.get_payload(decode=True)
if check_nested and b"-----BEGIN PGP MESSAGE-----" in raw_payload and b"-----END PGP MESSAGE-----" in raw_payload: if check_nested and b"-----BEGIN PGP MESSAGE-----" in raw_payload and b"-----END PGP MESSAGE-----" in raw_payload:
if verbose: LOG.debug("Message is already pgp encrypted. No nested encryption needed.")
log("Message is already pgp encrypted. No nested encryption needed.")
return payload return payload
# No check is needed for cfg['gpg']['keyhome'] as this is already done in method gpg_encrypt # No check is needed for conf.get_item('gpg', 'keyhome') as this is already done in method gpg_encrypt
gpg = GnuPG.GPGEncryptor( cfg['gpg']['keyhome'], gpg_to_cmdline, payload.get_content_charset() ) gpg = GnuPG.GPGEncryptor( conf.get_item('gpg', 'keyhome'), gpg_to_cmdline, payload.get_content_charset() )
gpg.update( raw_payload ) gpg.update( raw_payload )
encrypted_data, returncode = gpg.encrypt() encrypted_data, returncode = gpg.encrypt()
if verbose: LOG.debug("Return code from encryption=%d (0 indicates success)." % returncode)
log("Return code from encryption=%d (0 indicates success)." % returncode)
if returncode != 0: if returncode != 0:
log("Encrytion failed with return code %d. Encryption aborted." % returncode) LOG.info("Encrytion failed with return code %d. Encryption aborted." % returncode)
return payload return payload
payload.set_payload( encrypted_data ) payload.set_payload( encrypted_data )
@ -281,12 +253,13 @@ def encrypt_payload( payload, gpg_to_cmdline, check_nested = True ):
return payload return payload
def smime_encrypt( raw_message, recipients ): def smime_encrypt( raw_message, recipients ):
global LOG
if not get_bool_from_cfg('smime', 'cert_path'): if not conf.config_item_set('smime', 'cert_path'):
log("No valid path for S/MIME certs found in config file. S/MIME encryption aborted.") LOG.info("No valid path for S/MIME certs found in config file. S/MIME encryption aborted.")
return recipients return recipients
cert_path = cfg['smime']['cert_path']+"/" cert_path = conf.get_item('smime', 'cert_path')+"/"
s = SMIME.SMIME() s = SMIME.SMIME()
sk = X509.X509_Stack() sk = X509.X509_Stack()
smime_to = list() smime_to = list()
@ -297,15 +270,14 @@ def smime_encrypt( raw_message, recipients ):
if not (cert_and_email is None): if not (cert_and_email is None):
(to_cert, normal_email) = cert_and_email (to_cert, normal_email) = cert_and_email
if verbose: LOG.debug("Found cert " + to_cert + " for " + addr + ": " + normal_email)
log("Found cert " + to_cert + " for " + addr + ": " + normal_email)
smime_to.append(addr) smime_to.append(addr)
x509 = X509.load_cert(to_cert, format=X509.FORMAT_PEM) x509 = X509.load_cert(to_cert, format=X509.FORMAT_PEM)
sk.push(x509) sk.push(x509)
else: else:
unsmime_to.append(addr) unsmime_to.append(addr)
if smime_to != list(): if smime_to:
s.set_x509_stack(sk) s.set_x509_stack(sk)
s.set_cipher(SMIME.Cipher('aes_192_cbc')) s.set_cipher(SMIME.Cipher('aes_192_cbc'))
p7 = s.encrypt( BIO.MemoryBuffer( raw_message.as_string() ) ) p7 = s.encrypt( BIO.MemoryBuffer( raw_message.as_string() ) )
@ -320,22 +292,21 @@ def smime_encrypt( raw_message, recipients ):
if raw_message['Subject']: if raw_message['Subject']:
out.write('Subject: '+ raw_message['Subject'] + '\n') out.write('Subject: '+ raw_message['Subject'] + '\n')
if get_bool_from_cfg('default', 'add_header', 'yes'): if conf.config_item_equals('default', 'add_header', 'yes'):
out.write('X-GPG-Mailgate: Encrypted by GPG Mailgate\n') out.write('X-GPG-Mailgate: Encrypted by GPG Mailgate\n')
s.write(out, p7) s.write(out, p7)
if verbose: LOG.debug("Sending message from " + from_addr + " to " + str(smime_to))
log("Sending message from " + from_addr + " to " + str(smime_to))
send_msg(out.read(), smime_to) send_msg(out.read(), smime_to)
if unsmime_to != list(): if unsmime_to:
if verbose: LOG.debug("Unable to find valid S/MIME certificates for " + str(unsmime_to))
log("Unable to find valid S/MIME certificates for " + str(unsmime_to))
return unsmime_to return unsmime_to
def get_cert_for_email( to_addr, cert_path ): def get_cert_for_email( to_addr, cert_path ):
global LOG
files_in_directory = os.listdir(cert_path) files_in_directory = os.listdir(cert_path)
for filename in files_in_directory: for filename in files_in_directory:
@ -343,7 +314,7 @@ def get_cert_for_email( to_addr, cert_path ):
if not os.path.isfile(file_path): if not os.path.isfile(file_path):
continue continue
if get_bool_from_cfg('default', 'mail_case_insensitive', 'yes'): if conf.config_item_equals('default', 'mail_case_insensitive', 'yes'):
if filename.lower() == to_addr: if filename.lower() == to_addr:
return (file_path, to_addr) return (file_path, to_addr)
else: else:
@ -353,26 +324,14 @@ def get_cert_for_email( to_addr, cert_path ):
multi_email = re.match('^([^\+]+)\+([^@]+)@(.*)$', to_addr) multi_email = re.match('^([^\+]+)\+([^@]+)@(.*)$', to_addr)
if multi_email: if multi_email:
fixed_up_email = "%s@%s" % (multi_email.group(1), multi_email.group(3)) fixed_up_email = "%s@%s" % (multi_email.group(1), multi_email.group(3))
if verbose: LOG.debug("Multi-email %s converted to %s" % (to_addr, fixed_up_email))
log("Multi-email %s converted to %s" % (to_addr, fixed_up_email))
return get_cert_for_email(fixed_up_email) return get_cert_for_email(fixed_up_email)
return None return None
def get_bool_from_cfg( section, key = None, evaluation = None ):
if not (key is None) and not (evaluation is None):
return section in cfg and cfg[section].get(key) == evaluation
elif not (key is None) and (evaluation is None):
return section in cfg and not (cfg[section].get(key) is None)
else:
return section in cfg
def sanitize_case_sense( address ): def sanitize_case_sense( address ):
if get_bool_from_cfg('default', 'mail_case_insensitive', 'yes'): if conf.config_item_equals('default', 'mail_case_insensitive', 'yes'):
address = address.lower() address = address.lower()
else: else:
if isinstance(address, str): if isinstance(address, str):
@ -406,22 +365,24 @@ def get_first_payload( payloads ):
return payloads return payloads
def send_msg( message, recipients ): def send_msg( message, recipients ):
global LOG
recipients = [_f for _f in recipients if _f] recipients = [_f for _f in recipients if _f]
if recipients: if recipients:
if not (get_bool_from_cfg('relay', 'host') and get_bool_from_cfg('relay', 'port')): if not (conf.config_item_set('relay', 'host') and conf.config_item_set('relay', 'port')):
log("Missing settings for relay. Sending email aborted.") LOG.warning("Missing settings for relay. Sending email aborted.")
return None return None
log("Sending email to: <%s>" % '> <'.join( recipients )) LOG.info("Sending email to: <%s>" % '> <'.join( recipients ))
relay = (cfg['relay']['host'], int(cfg['relay']['port'])) relay = (conf.get_item('relay', 'host'), int(conf.get_item('relay', 'port')))
smtp = smtplib.SMTP(relay[0], relay[1]) smtp = smtplib.SMTP(relay[0], relay[1])
if 'relay' in cfg and 'starttls' in cfg['relay'] and cfg['relay']['starttls'] == 'yes': if conf.config_item_equals('relay', 'starttls', 'yes'):
smtp.starttls() smtp.starttls()
smtp.sendmail( from_addr, recipients, message ) smtp.sendmail( from_addr, recipients, message )
else: else:
log("No recipient found") LOG.info("No recipient found")
def sort_recipients( raw_message, from_addr, to_addrs ): def sort_recipients( raw_message, from_addr, to_addrs ):
global LOG
recipients_left = list() recipients_left = list()
for recipient in to_addrs: for recipient in to_addrs:
@ -430,37 +391,46 @@ def sort_recipients( raw_message, from_addr, to_addrs ):
# There is no need for nested encryption # There is no need for nested encryption
first_payload = get_first_payload(raw_message) first_payload = get_first_payload(raw_message)
if first_payload.get_content_type() == 'application/pkcs7-mime': if first_payload.get_content_type() == 'application/pkcs7-mime':
if verbose: LOG.debug("Message is already encrypted with S/MIME. Encryption aborted.")
log("Message is already encrypted with S/MIME. Encryption aborted.")
send_msg(raw_message.as_string(), recipients_left) send_msg(raw_message.as_string(), recipients_left)
return return
first_payload = first_payload.get_payload(decode=True) first_payload = first_payload.get_payload(decode=True)
if b"-----BEGIN PGP MESSAGE-----" in first_payload and b"-----END PGP MESSAGE-----" in first_payload: if b"-----BEGIN PGP MESSAGE-----" in first_payload and b"-----END PGP MESSAGE-----" in first_payload:
if verbose: LOG.debug("Message is already encrypted as PGP/INLINE. Encryption aborted.")
log("Message is already encrypted as PGP/INLINE. Encryption aborted.")
send_msg(raw_message.as_string(), recipients_left) send_msg(raw_message.as_string(), recipients_left)
return return
if raw_message.get_content_type() == 'multipart/encrypted': if raw_message.get_content_type() == 'multipart/encrypted':
if verbose: LOG.debug("Message is already encrypted. Encryption aborted.")
log("Message is already encrypted. Encryption aborted.")
send_msg(raw_message.as_string(), recipients_left) send_msg(raw_message.as_string(), recipients_left)
return return
# Encrypt mails for recipients with known public PGP keys # Encrypt mails for recipients with known public PGP keys
recipients_left = gpg_encrypt(raw_message, recipients_left) recipients_left = gpg_encrypt(raw_message, recipients_left)
if recipients_left == list(): if not recipients_left:
return return
# Encrypt mails for recipients with known S/MIME certificate # Encrypt mails for recipients with known S/MIME certificate
recipients_left = smime_encrypt(raw_message, recipients_left) recipients_left = smime_encrypt(raw_message, recipients_left)
if recipients_left == list(): if not recipients_left:
return return
# Send out mail to recipients which are left # Send out mail to recipients which are left
send_msg(raw_message.as_string(), recipients_left) send_msg(raw_message.as_string(), recipients_left)
conf.load_config()
lacre.init_logging(conf.get_item('logging', 'config'))
LOG = logging.getLogger(__name__)
# Read e-mail from stdin
raw = sys.stdin.read()
raw_message = email.message_from_string( raw )
from_addr = raw_message['From']
to_addrs = sys.argv[1:]
# Let's start # Let's start
sort_recipients(raw_message, from_addr, to_addrs) sort_recipients(raw_message, from_addr, to_addrs)

36
lacre/__init__.py Normal file
View File

@ -0,0 +1,36 @@
"""Lacre --- the Postfix mail filter encrypting incoming email
"""
import logging
import logging.config
# Following structure configures logging iff a file-based configuration cannot
# be performed. It only sets up a syslog handler, so that the admin has at
# least some basic information.
FAIL_OVER_LOGGING_CONFIG = {
'version': 1,
'formatters': {
'sysfmt': {
'format': '%(asctime)s %(module)s %(message)s',
'datefmt': '%Y-%m-%d %H:%M:%S'
},
},
'handlers': {
'syslog': {
'class': 'logging.handlers.SysLogHandler',
'level': 'INFO',
'formatter': 'sysfmt'
}
},
'root': {
'level': 'INFO',
'handlers': ['syslog']
}
}
def init_logging(config_filename):
if config_filename is not None:
logging.config.fileConfig(config_filename)
else:
logging.config.dictConfig(FAIL_OVER_LOGGING_CONFIG)
logging.warning('Lacre logging configuration missing, using syslog as default')

70
lacre/config.py Normal file
View File

@ -0,0 +1,70 @@
"""Lacre configuration
Routines defined here are responsible for processing configuration.
"""
from configparser import RawConfigParser
import os
# Environment variable name we read to retrieve configuration path. This is to
# enable non-root users to set up and run GPG Mailgate and to make the software
# testable.
CONFIG_PATH_ENV = "GPG_MAILGATE_CONFIG"
# Global dict to keep configuration parameters. It's hidden behind several
# utility functions to make it easy to replace it with ConfigParser object in
# the future.
cfg = dict()
def load_config() -> dict:
"""Parses configuration file.
If environment variable identified by CONFIG_PATH_ENV
variable is set, its value is taken as a configuration file
path. Otherwise, the default is taken
('/etc/gpg-mailgate.conf').
"""
configFile = os.getenv(CONFIG_PATH_ENV, '/etc/gpg-mailgate.conf')
parser = read_config(configFile)
global cfg
cfg = copy_to_dict(parser)
return cfg
def read_config(fileName) -> RawConfigParser:
cp = RawConfigParser()
cp.read(fileName)
return cp
def copy_to_dict(confParser) -> dict:
config = dict()
for sect in confParser.sections():
config[sect] = dict()
for (name, value) in confParser.items(sect):
config[sect][name] = value
return config
def get_item(section, key, empty_value = None):
global cfg
if config_item_set(section, key):
return cfg[section][key]
else:
return empty_value
def has_section(section) -> bool:
global cfg
return section in cfg
def config_item_set(section, key) -> bool:
global cfg
return section in cfg and (key in cfg[section]) and not (cfg[section][key] is None)
def config_item_equals(section, key, value) -> bool:
global cfg
return section in cfg and key in cfg[section] and cfg[section][key] == value

View File

@ -7,37 +7,28 @@ from M2Crypto import BIO, Rand, SMIME, X509
from email.mime.text import MIMEText from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart from email.mime.multipart import MIMEMultipart
# Read configuration from /etc/gpg-mailgate.conf import logging
_cfg = RawConfigParser()
_cfg.read('/etc/gpg-mailgate.conf')
cfg = dict()
for sect in _cfg.sections():
cfg[sect] = dict()
for (name, value) in _cfg.items(sect):
cfg[sect][name] = value
def log(msg): import lacre
if 'logging' in cfg and 'file' in cfg['logging']: import lacre.config as conf
if cfg['logging']['file'] == "syslog":
syslog.syslog(syslog.LOG_INFO | syslog.LOG_MAIL, msg)
else:
logfile = open(cfg['logging']['file'], 'a')
logfile.write(msg + "\n")
logfile.close()
CERT_PATH = cfg['smime']['cert_path']+"/"
def send_msg( message, from_addr, recipients = None ): def send_msg( message, from_addr, recipients = None ):
if conf.config_item_set('relay', 'host') and conf.config_item_set('relay', 'enc_port'):
if 'relay' in cfg and 'host' in cfg['relay'] and 'enc_port' in cfg['relay']: relay = (conf.get_item('relay', 'host'), int(conf.get_item('relay', 'enc_port')))
relay = (cfg['relay']['host'], int(cfg['relay']['enc_port']))
smtp = smtplib.SMTP(relay[0], relay[1]) smtp = smtplib.SMTP(relay[0], relay[1])
smtp.sendmail( from_addr, recipients, message.as_string() ) smtp.sendmail( from_addr, recipients, message.as_string() )
else: else:
log("Could not send mail due to wrong configuration") LOG.info("Could not send mail due to wrong configuration")
if __name__ == "__main__": if __name__ == "__main__":
# try: # try:
conf.load_config()
lacre.init_logging(conf.get_item('logging', 'config'))
LOG = logging.getLogger(__name__)
CERT_PATH = conf.get_item('smime', 'cert_path') + '/'
# Read e-mail from stdin # Read e-mail from stdin
raw = sys.stdin.read() raw = sys.stdin.read()
register_msg = email.message_from_string( raw ) register_msg = email.message_from_string( raw )
@ -63,18 +54,18 @@ if __name__ == "__main__":
break break
if sign_part == None: if sign_part == None:
log("Unable to find PKCS7 signature or public PGP key in registration email") LOG.info("Unable to find PKCS7 signature or public PGP key in registration email")
failure_msg = file( cfg['mailregister']['mail_templates'] + "/registrationError.md").read() failure_msg = file( conf.get_item('mailregister', 'mail_templates') + "/registrationError.md").read()
msg = MIMEMultipart("alternative") msg = MIMEMultipart("alternative")
msg["From"] = cfg['mailregister']['register_email'] msg["From"] = conf.get_item('mailregister', 'register_email')
msg["To"] = from_addr msg["To"] = from_addr
msg["Subject"] = "S/MIME / OpenPGP registration failed" msg["Subject"] = "S/MIME / OpenPGP registration failed"
msg.attach(MIMEText(failure_msg, 'plain')) msg.attach(MIMEText(failure_msg, 'plain'))
msg.attach(MIMEText(markdown.markdown(failure_msg), 'html')) msg.attach(MIMEText(markdown.markdown(failure_msg), 'html'))
send_msg(msg, cfg['mailregister']['register_email'], [from_addr]) send_msg(msg, conf.get_item('mailregister', 'register_email'), [from_addr])
sys.exit(0) sys.exit(0)
if sign_type == 'smime': if sign_type == 'smime':
@ -105,42 +96,42 @@ if __name__ == "__main__":
# format in user-specific data # format in user-specific data
# sending success mail only for S/MIME as GPGMW handles this on its own # sending success mail only for S/MIME as GPGMW handles this on its own
success_msg = file(cfg['mailregister']['mail_templates']+"/registrationSuccess.md").read() success_msg = file(conf.get_item('mailregister', 'mail_templates')+"/registrationSuccess.md").read()
success_msg = success_msg.replace("[:FROMADDRESS:]", from_addr) success_msg = success_msg.replace("[:FROMADDRESS:]", from_addr)
msg = MIMEMultipart("alternative") msg = MIMEMultipart("alternative")
msg["From"] = cfg['mailregister']['register_email'] msg["From"] = conf.get_item('mailregister', 'register_email')
msg["To"] = from_addr msg["To"] = from_addr
msg["Subject"] = "S/MIME certificate registration succeeded" msg["Subject"] = "S/MIME certificate registration succeeded"
msg.attach(MIMEText(success_msg, 'plain')) msg.attach(MIMEText(success_msg, 'plain'))
msg.attach(MIMEText(markdown.markdown(success_msg), 'html')) msg.attach(MIMEText(markdown.markdown(success_msg), 'html'))
send_msg(msg, cfg['mailregister']['register_email'], [from_addr]) send_msg(msg, conf.get_item('mailregister', 'register_email'), [from_addr])
log("S/MIME Registration succeeded") LOG.info("S/MIME Registration succeeded")
elif sign_type == 'pgp': elif sign_type == 'pgp':
# send POST to gpg-mailgate webpanel # send POST to gpg-mailgate webpanel
sig = sign_part sig = sign_part
payload = {'email': from_addr, 'key': sig} payload = {'email': from_addr, 'key': sig}
r = requests.post(cfg['mailregister']['webpanel_url'], data=payload) r = requests.post(conf.get_item('mailregister', 'webpanel_url'), data=payload)
if r.status_code != 200: if r.status_code != 200:
log("Could not hand registration over to GPGMW. Error: %s" % r.status_code) LOG.info("Could not hand registration over to GPGMW. Error: %s" % r.status_code)
error_msg = file(cfg['mailregister']['mail_templates']+"/gpgmwFailed.md").read() error_msg = open(conf.get_item('mailregister', 'mail_templates')+"/gpgmwFailed.md").read()
error_msg = error_msg.replace("[:FROMADDRESS:]", from_addr) error_msg = error_msg.replace("[:FROMADDRESS:]", from_addr)
msg = MIMEMultipart("alternative") msg = MIMEMultipart("alternative")
msg["From"] = cfg['mailregister']['register_email'] msg["From"] = conf.get_item('mailregister', 'register_email')
msg["To"] = from_addr msg["To"] = from_addr
msg["Subject"] = "PGP key registration failed" msg["Subject"] = "PGP key registration failed"
msg.attach(MIMEText(error_msg, 'plain')) msg.attach(MIMEText(error_msg, 'plain'))
msg.attach(MIMEText(markdown.markdown(error_msg), 'html')) msg.attach(MIMEText(markdown.markdown(error_msg), 'html'))
send_msg(msg, cfg['mailregister']['register_email'], [from_addr]) send_msg(msg, conf.get_item('mailregister', 'register_email'), [from_addr])
else: else:
log("PGP registration is handed over to GPGMW") LOG.info("PGP registration is handed over to GPGMW")
# except: # except:
# log("Registration exception") # LOG.info("Registration exception")
# sys.exit(0) # sys.exit(0)

View File

@ -35,6 +35,7 @@ e2e_log: test/logs/e2e.log
e2e_log_format: %(asctime)s %(pathname)s:%(lineno)d %(levelname)s [%(funcName)s] %(message)s e2e_log_format: %(asctime)s %(pathname)s:%(lineno)d %(levelname)s [%(funcName)s] %(message)s
e2e_log_datefmt: %Y-%m-%d %H:%M:%S e2e_log_datefmt: %Y-%m-%d %H:%M:%S
lacre_log: test/logs/gpg-mailgate.log lacre_log: test/logs/gpg-mailgate.log
log_config: test/gpg-lacre-log.ini
[case-1] [case-1]
descr: Clear text message to a user without a key descr: Clear text message to a user without a key

View File

@ -33,11 +33,10 @@ RELAY_SCRIPT = "test/relay.py"
CONFIG_FILE = "test/gpg-mailgate.conf" CONFIG_FILE = "test/gpg-mailgate.conf"
def build_config(config): def build_config(config):
cp = configparser.ConfigParser() cp = configparser.RawConfigParser()
cp.add_section("logging") cp.add_section("logging")
cp.set("logging", "file", config["log_file"]) cp.set("logging", "config", config["log_config"])
cp.set("logging", "verbose", "yes")
cp.add_section("gpg") cp.add_section("gpg")
cp.set("gpg", "keyhome", config["gpg_keyhome"]) cp.set("gpg", "keyhome", config["gpg_keyhome"])
@ -147,7 +146,7 @@ write_test_config(config_path,
port = config.get("relay", "port"), port = config.get("relay", "port"),
gpg_keyhome = config.get("dirs", "keys"), gpg_keyhome = config.get("dirs", "keys"),
smime_certpath = config.get("dirs", "certs"), smime_certpath = config.get("dirs", "certs"),
log_file = config.get("tests", "lacre_log")) log_config = config.get("tests", "log_config"))
for case_no in range(1, config.getint("tests", "cases")+1): for case_no in range(1, config.getint("tests", "cases")+1):
case_name = f"case-{case_no}" case_name = f"case-{case_no}"

24
test/gpg-lacre-log.ini Normal file
View File

@ -0,0 +1,24 @@
[loggers]
keys=root
[logger_root]
level=NOTSET
handlers=lacrelog
[handlers]
keys=lacrelog
[formatters]
keys=postfixfmt
[handler_lacrelog]
class=FileHandler
level=DEBUG
formatter=postfixfmt
args=('test/logs/lacre.log', 'a+')
[formatter_postfixfmt]
format=%(asctime)s %(module)s[%(process)d]: %(message)s
datefmt=%b %e %H:%M:%S
style=%
validate=True

View File

@ -11,6 +11,8 @@
import sys import sys
import socket import socket
import logging
EXIT_UNAVAILABLE = 1 EXIT_UNAVAILABLE = 1
@ -44,12 +46,17 @@ def serve(port):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try: try:
s.bind(localhost_at(port)) s.bind(localhost_at(port))
logging.info(f"Listening on localhost, port {port}")
s.listen(1) s.listen(1)
logging.info("Listening...")
except socket.error as e: except socket.error as e:
print("Cannot connect", e) print("Cannot connect", e)
logging.error(f"Cannot connect {e}")
sys.exit(EXIT_UNAVAILABLE) sys.exit(EXIT_UNAVAILABLE)
logging.debug("About to accept a connection...")
(conn, addr) = s.accept() (conn, addr) = s.accept()
logging.debug(f"Accepting connection from {conn}")
conn.sendall(welcome(b"TEST SERVER")) conn.sendall(welcome(b"TEST SERVER"))
receive_and_confirm(conn) # Ignore HELO/EHLO receive_and_confirm(conn) # Ignore HELO/EHLO
@ -70,14 +77,24 @@ def serve(port):
conn.close() conn.close()
logging.debug(f"Received {len(message)} characters of data")
# Trim EOM marker as we're only interested in the message body. # Trim EOM marker as we're only interested in the message body.
return message[:-len(EOM)] return message[:-len(EOM)]
def error(msg, exit_code): def error(msg, exit_code):
logging.error(msg)
print("ERROR: %s" % (msg)) print("ERROR: %s" % (msg))
sys.exit(exit_code) sys.exit(exit_code)
# filename is relative to where we run the tests from, i.e. the project root
# directory
logging.basicConfig(filename = 'test/logs/relay.log',
format = '%(asctime)s %(pathname)s:%(lineno)d %(levelname)s [%(funcName)s] %(message)s',
datefmt = '%Y-%m-%d %H:%M:%S',
level = logging.DEBUG)
if len(sys.argv) < 2: if len(sys.argv) < 2:
error("Usage: relay.py PORT_NUMBER", EXIT_UNAVAILABLE) error("Usage: relay.py PORT_NUMBER", EXIT_UNAVAILABLE)

View File

@ -31,60 +31,60 @@ conn = test_db.connect()
# Populate the database with dummy data # Populate the database with dummy data
conn.execute(gpgmw_keys.insert(), [ conn.execute(gpgmw_keys.insert(), [
{"id": 1, "email": "alice@disposlab", "publickey": "-----BEGIN PGP PUBLIC KEY BLOCK-----\ {"id": 1, "email": "alice@disposlab", "publickey": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n\
\ \n\
mQGNBGDYY5oBDAC+HAVjA05jsIpHfQ2KQ9m2olo1Qnlk+dkjD+Gagxj1ACezyiGL\ mQGNBGDYY5oBDAC+HAVjA05jsIpHfQ2KQ9m2olo1Qnlk+dkjD+Gagxj1ACezyiGL\n\
cfZfoE/MJYLCH9yPcX1fUIAPwdAyfJKlvkVcz+MhEpgl3aP3NM2L2unSx3v9ZFwT\ cfZfoE/MJYLCH9yPcX1fUIAPwdAyfJKlvkVcz+MhEpgl3aP3NM2L2unSx3v9ZFwT\n\
/qyMo9Zst5VSD04TVx2ySQB1vucd2ppgp66X7hlCxs+P8d0FV7VcdrNYol2oOtYP\ /qyMo9Zst5VSD04TVx2ySQB1vucd2ppgp66X7hlCxs+P8d0FV7VcdrNYol2oOtYP\n\
yEFXkdyXLI/INI6jrqNkBF87ej+dlTQZAm3zoj61Xwq4gW0YesAZoJyXs8X+a4Am\ yEFXkdyXLI/INI6jrqNkBF87ej+dlTQZAm3zoj61Xwq4gW0YesAZoJyXs8X+a4Am\n\
8KF7YYcTcIy89yXflotmExpE+i77datSBLM/FpIPiUfkfK6q/TNyno8Z3PBC0QD5\ 8KF7YYcTcIy89yXflotmExpE+i77datSBLM/FpIPiUfkfK6q/TNyno8Z3PBC0QD5\n\
21leqfp/QHRkwmqFbIVuoeonCvrAccjM0ITLjW+P0xXJa3q0lQQCgcGOgqTuNWPT\ 21leqfp/QHRkwmqFbIVuoeonCvrAccjM0ITLjW+P0xXJa3q0lQQCgcGOgqTuNWPT\n\
6FhlmvkXt6fBZ11C2I1b033HTePvjIwxOrEY8pSqYwerVX9EU7FXT+S98HNW/1nF\ 6FhlmvkXt6fBZ11C2I1b033HTePvjIwxOrEY8pSqYwerVX9EU7FXT+S98HNW/1nF\n\
cNk3SoofzUOcKZOwc5n0NEESrW7sWpmD6Qmf52+GURuO+15DSUt13xqmnte19Xqd\ cNk3SoofzUOcKZOwc5n0NEESrW7sWpmD6Qmf52+GURuO+15DSUt13xqmnte19Xqd\n\
n98y0wrYAUgyUY8AEQEAAbQPYWxpY2VAZGlzcG9zbGFiiQHUBBMBCAA+FiEEHNJF\ n98y0wrYAUgyUY8AEQEAAbQPYWxpY2VAZGlzcG9zbGFiiQHUBBMBCAA+FiEEHNJF\n\
MI8JY9A46INXlzz02Th8RNcFAmDYY5oCGwMFCQPCZwAFCwkIBwIGFQoJCAsCBBYC\ MI8JY9A46INXlzz02Th8RNcFAmDYY5oCGwMFCQPCZwAFCwkIBwIGFQoJCAsCBBYC\n\
AwECHgECF4AACgkQlzz02Th8RNdZeAv+IVVK49f0tY5QOSERu5RqdyFNpsVlUws9\ AwECHgECF4AACgkQlzz02Th8RNdZeAv+IVVK49f0tY5QOSERu5RqdyFNpsVlUws9\n\
swvSvXXK/ZQxZ3YD3o0WEJG5G8jRO+Zjrljx6zzH39ofEKn8QMQUuw+SVPrzbqQb\ swvSvXXK/ZQxZ3YD3o0WEJG5G8jRO+Zjrljx6zzH39ofEKn8QMQUuw+SVPrzbqQb\n\
Yp/idn1E9RZCyyhtwcYnIwUObq2NNsCk8UmnjYvpwoh/QcHic13/RSUj7vejujtB\ Yp/idn1E9RZCyyhtwcYnIwUObq2NNsCk8UmnjYvpwoh/QcHic13/RSUj7vejujtB\n\
SRTjNUE/RK5ROY8r+xZW9ZV/Q0NEzKl2wQtmbt8vTRX9yNEB171XZHG7dg4bTzm+\ SRTjNUE/RK5ROY8r+xZW9ZV/Q0NEzKl2wQtmbt8vTRX9yNEB171XZHG7dg4bTzm+\n\
zs0jPGNT0ygcx+uE7DZ3RkyPLRk3fB+GPiYrL2lfPF1KkrHGY4PGhClKdR1kjfBA\ zs0jPGNT0ygcx+uE7DZ3RkyPLRk3fB+GPiYrL2lfPF1KkrHGY4PGhClKdR1kjfBA\n\
Kweb6ExZg0fBYlB8ia8z3RZQF29pztoVfk8KIimg9RoYNOKw3Jp5SnHsbz9JygmZ\ Kweb6ExZg0fBYlB8ia8z3RZQF29pztoVfk8KIimg9RoYNOKw3Jp5SnHsbz9JygmZ\n\
mp3M3Lrs7357oSn9x25/nrFGeUBWbbKoXSdoXZr0Ix4xxkOJPAK966w0pQq+sP+o\ mp3M3Lrs7357oSn9x25/nrFGeUBWbbKoXSdoXZr0Ix4xxkOJPAK966w0pQq+sP+o\n\
Ozg3F2rFRc6SoQw1pNLQ57hhWTblQlz8ETY7GnVJ+0xiqkAq2hrLt0jhQ5taWjV6\ Ozg3F2rFRc6SoQw1pNLQ57hhWTblQlz8ETY7GnVJ+0xiqkAq2hrLt0jhQ5taWjV6\n\
Fgy8fKUPd5OAMvB9bfmAErclWcqKarMcuQGNBGDYY5oBDAC6yOtgUwtKUsI3jTu2\ Fgy8fKUPd5OAMvB9bfmAErclWcqKarMcuQGNBGDYY5oBDAC6yOtgUwtKUsI3jTu2\n\
VdjNDEnt/VLdRseT4JosSMglZ963nlA4mltCjxj59DeM0Ft8eyF7Bu4EFw5Kid+O\ VdjNDEnt/VLdRseT4JosSMglZ963nlA4mltCjxj59DeM0Ft8eyF7Bu4EFw5Kid+O\n\
vKGA5rGZBE0IVROOvSJQNbcELkY9XYtZjOJ7elfG37rDQKfDk82xqod9iTd48nm7\ vKGA5rGZBE0IVROOvSJQNbcELkY9XYtZjOJ7elfG37rDQKfDk82xqod9iTd48nm7\n\
vrllvylQhKfXa+m99KxWabtKqCyXVjaZP9vfD3nVauu16oHW6rQavlLXo5MetFan\ vrllvylQhKfXa+m99KxWabtKqCyXVjaZP9vfD3nVauu16oHW6rQavlLXo5MetFan\n\
Iwv1sTqnpzCt+cuG/7vUt89rOiJRalRP3/e1K5MSM6aWC/SHZs6HcrT+WT5nuPA+\ Iwv1sTqnpzCt+cuG/7vUt89rOiJRalRP3/e1K5MSM6aWC/SHZs6HcrT+WT5nuPA+\n\
5VQ4gFCSb8UlscF4sI++hhB/k821vyl9hIjnR3aRiFWdrkykQOfZNhovvsnmJmk9\ 5VQ4gFCSb8UlscF4sI++hhB/k821vyl9hIjnR3aRiFWdrkykQOfZNhovvsnmJmk9\n\
+Zcq0M3pZBnBuLgxVwJNVa4gi63cYwtExpcAZcG28wSVmcXcPN2wxEpYg5n/nvvG\ +Zcq0M3pZBnBuLgxVwJNVa4gi63cYwtExpcAZcG28wSVmcXcPN2wxEpYg5n/nvvG\n\
8Dsk0AA5WU5WW8aLLLQNBmVg2y4Oa1Fy0M7yfSylLWBAdj7y8+UzspN6JCbYhOpP\ 8Dsk0AA5WU5WW8aLLLQNBmVg2y4Oa1Fy0M7yfSylLWBAdj7y8+UzspN6JCbYhOpP\n\
lLRCJv5+JOgR0MrA+lxfFZwfcSO12x+gkfQ9oyUBdXNuydMAEQEAAYkBtgQYAQgA\ lLRCJv5+JOgR0MrA+lxfFZwfcSO12x+gkfQ9oyUBdXNuydMAEQEAAYkBtgQYAQgA\n\
IBYhBBzSRTCPCWPQOOiDV5c89Nk4fETXBQJg2GOaAhsMAAoJEJc89Nk4fETXpooL\ IBYhBBzSRTCPCWPQOOiDV5c89Nk4fETXBQJg2GOaAhsMAAoJEJc89Nk4fETXpooL\n\
/iJKgNF80neUamewma1aZJjwKWoHysSWWSlPeU6pGctuJv15fbAfI/NM1iXnSEGt\ /iJKgNF80neUamewma1aZJjwKWoHysSWWSlPeU6pGctuJv15fbAfI/NM1iXnSEGt\n\
odsn0oHtuAASlVB0ckSFdE0a2DwLgO6s6oEJof/yrE5hIAAlwzjHsi1G/dtHcfIo\ odsn0oHtuAASlVB0ckSFdE0a2DwLgO6s6oEJof/yrE5hIAAlwzjHsi1G/dtHcfIo\n\
SjHzE22qUZwwm5ketuvKvEDKKp3b1ccu37AZC1caRFh3q8xB5ByLh1gPiDJ+ehwU\ SjHzE22qUZwwm5ketuvKvEDKKp3b1ccu37AZC1caRFh3q8xB5ByLh1gPiDJ+ehwU\n\
puXkXPdFQhQTZib4LYuMxzh6A+S9U0AM7WMKjX7PhJ68maOeQ+yOIBSWtBKyWwZu\ puXkXPdFQhQTZib4LYuMxzh6A+S9U0AM7WMKjX7PhJ68maOeQ+yOIBSWtBKyWwZu\n\
Sx01w+Y/USPz02AxUn102se52FCISc/NijlX1JvFQdzf/WaZu28nTmW9OXSW3WeK\ Sx01w+Y/USPz02AxUn102se52FCISc/NijlX1JvFQdzf/WaZu28nTmW9OXSW3WeK\n\
ql7zNQqj494JD8gJuRGCU9AaiCmOaBokRdLiGbin/wxiG1CkXGRDN5/r0m/1IoNz\ ql7zNQqj494JD8gJuRGCU9AaiCmOaBokRdLiGbin/wxiG1CkXGRDN5/r0m/1IoNz\n\
I4m2SLsB/a89WACQ//CKJyNn4xPOEQoix35tXjdjTLAVyTrX502vHGieZ3HJU2tb\ I4m2SLsB/a89WACQ//CKJyNn4xPOEQoix35tXjdjTLAVyTrX502vHGieZ3HJU2tb\n\
nmmMf/H0kReMtNYFwHxoTpBJ8vk+xcZ+6ETzH8nk6av+zZ/5T5Y0aD5zO89PcQk6\ nmmMf/H0kReMtNYFwHxoTpBJ8vk+xcZ+6ETzH8nk6av+zZ/5T5Y0aD5zO89PcQk6\n\
pw==\ pw==\n\
=Tbwz\ =Tbwz\n\
-----END PGP PUBLIC KEY BLOCK-----\ -----END PGP PUBLIC KEY BLOCK-----\
", "status": 0, "confirm": "", "time": None}, ", "status": 0, "confirm": "", "time": None},
{"id": 2, "email": "bob@disposlab", "publickey": "-----BEGIN PGP PUBLIC KEY BLOCK-----\ {"id": 2, "email": "bob@disposlab", "publickey": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n\
\ \n\
mDMEYdTFkRYJKwYBBAHaRw8BAQdA2tgdP1pMt3cv3XAW7ov5AFn74mMZvyTksp9Q\ mDMEYdTFkRYJKwYBBAHaRw8BAQdA2tgdP1pMt3cv3XAW7ov5AFn74mMZvyTksp9Q\n\
eO1PkpK0GkJvYiBGb29iYXIgPGJvYkBkaXNwb3NsYWI+iJYEExYIAD4WIQQZz0tH\ eO1PkpK0GkJvYiBGb29iYXIgPGJvYkBkaXNwb3NsYWI+iJYEExYIAD4WIQQZz0tH\n\
7MnEevqE1L2W85/aDjG7ZwUCYdTFkQIbAwUJA8JnAAULCQgHAgYVCgkICwIEFgID\ 7MnEevqE1L2W85/aDjG7ZwUCYdTFkQIbAwUJA8JnAAULCQgHAgYVCgkICwIEFgID\n\
AQIeAQIXgAAKCRCW85/aDjG7ZxVnAP49t7BU2H+/WCpa3fCAlMEcik82sU4p+U9D\ AQIeAQIXgAAKCRCW85/aDjG7ZxVnAP49t7BU2H+/WCpa3fCAlMEcik82sU4p+U9D\n\
pMsbjawwYgEA1SbA5CF835cMjoEufy1h+2M4T9gI/0X2lk8OAtwwggm4OARh1MXg\ pMsbjawwYgEA1SbA5CF835cMjoEufy1h+2M4T9gI/0X2lk8OAtwwggm4OARh1MXg\n\
EgorBgEEAZdVAQUBAQdAUVNKx2OsGtNdRsnl3J/uv6obkUC0KcO4ikdRs+iejlMD\ EgorBgEEAZdVAQUBAQdAUVNKx2OsGtNdRsnl3J/uv6obkUC0KcO4ikdRs+iejlMD\n\
AQgHiHgEGBYIACAWIQQZz0tH7MnEevqE1L2W85/aDjG7ZwUCYdTF4AIbDAAKCRCW\ AQgHiHgEGBYIACAWIQQZz0tH7MnEevqE1L2W85/aDjG7ZwUCYdTF4AIbDAAKCRCW\n\
85/aDjG7Z039APwLGP5ibqCC9yIr4YVbdWff1Ch+2C91MR2ObF93Up9+ogD8D2zd\ 85/aDjG7Z039APwLGP5ibqCC9yIr4YVbdWff1Ch+2C91MR2ObF93Up9+ogD8D2zd\n\
OjjB6xRD0Q2FN+alsNGCtdutAs18AZ5l33RMzws=\ OjjB6xRD0Q2FN+alsNGCtdutAs18AZ5l33RMzws=\n\
=wWoq\ =wWoq\n\
-----END PGP PUBLIC KEY BLOCK-----\ -----END PGP PUBLIC KEY BLOCK-----\
", "status": 0, "confirm": "", "time": None}, ", "status": 0, "confirm": "", "time": None},
{"id": 3, "email": "cecil@lacre.io", "publickey": "RUBBISH", "status": 0, "confirm": "", "time": None} {"id": 3, "email": "cecil@lacre.io", "publickey": "RUBBISH", "status": 0, "confirm": "", "time": None}