otcerts/OTcerts.py

import os
import sys
import errno
import argparse
import configparser
import logging
import mysql.connector
import subprocess

# Query for IMAP/POP3 certificate
mbox_list_stmt = "SELECT DISTINCT(name) FROM records WHERE content in ({}) and (name LIKE 'imap.%' or name LIKE 'pop3.%' or name LIKE 'mail.%')"

# Query for SMTP certificate
smtp_list_stmt = "SELECT DISTINCT(name) FROM records WHERE content in ({}) and (name LIKE 'smtp.%' or name LIKE 'mail.%')"


# Get list of defined domains in vhosts configuration database
domains_list_stmt = """SELECT DISTINCT(SUBSTRING_INDEX(urls.dns_name, '.', -2)) AS domain_names 
FROM urls INNER JOIN (hosts_urls, hosts, vhosts_features, vhosts) 
ON (urls.url_id = hosts_urls.url_id and urls.url_id = vhosts.url_id and vhosts.vhost_id = vhosts_features.vhost_id) 
WHERE (hosts_urls.http = 'Y' and hosts.hostname =  %(webserver)s)
"""

# Get domain_id if defined in nameserver database
domain_id_stmt="SELECT domains.id as domain_id FROM domains WHERE domains.name=%(domain)s"

subdomains_list_stmt = "SELECT DISTINCT(urls.dns_name) AS domain_names "\
                       "FROM urls INNER JOIN (hosts_urls, hosts, vhosts_features, vhosts ) "\
                       "ON (urls.url_id = hosts_urls.url_id and urls.url_id = vhosts.url_id and vhosts.vhost_id = vhosts_features.vhost_id) "\
                       "WHERE (hosts_urls.http = 'Y' and hosts.hostname =  %(webserver)s and "\
                       "urls.dns_name LIKE %(domain)s)"

default_conf_file="./etc/ot_certs.ini"
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()


def init_prog(argv):
    """
    Parse command line args and config file
    """

    parser = argparse.ArgumentParser(
        description="Manage LetsEncrypt certificates")
    parser.add_argument("-c", "--config", type=open,
                        required=False,
                        default=default_conf_file,
                        help="Specifity config file (default: {})".format(default_conf_file))
    parser.add_argument("--renew", default=False, action='store_true', required=False,
                        help="Invoca solamente il renew per i certificati gia' presenti")
    
    service_group = parser.add_mutually_exclusive_group(required=True)
    service_group.add_argument("--liste", default=False, action='store_true', required=False,
                        help="Richiedi i certificati per liste.indivia.net")
    service_group.add_argument("--hosting", default=False, action='store_true', required=False,
                        help="Richiedi i certificati per i siti in hosting")
    service_group.add_argument("--webmail", default=False, action='store_true', required=False,
                        help="Richiedi i certificati per le webmail")
    service_group.add_argument("--smtp", default=False, action='store_true', required=False,
                        help="Richiedi i certificati per il server SMTP")
    service_group.add_argument("--mbox", default=False, action='store_true', required=False,
                        help="Richiedi i certificati per il server POP/IMAP")

    args = parser.parse_args()
    try:
        config = configparser.ConfigParser()
        config.read_file(args.config)
    except Exception as e:
        logger.error("Error parsing configuration {}".format(e))
        exit(-1)
    return args, config



def connect_db(conf_dict):

    try:
        cnx = mysql.connector.connect(**conf_dict)
    except mysql.connector.Error as err:
        if err.errno == mysql.connector.errorcode.ER_ACCESS_DENIED_ERROR:
            logger.error("Something is wrong with your user name or password")
        elif err.errno == mysql.connector.errorcode.ER_BAD_DB_ERROR:
            logger.error("Database does not exist")
        else:
            logger.error(err)
        return None
    return cnx


def get_subdomain_list(config,  domain, ot_conn, ex_subdomains=tuple()):
    """
    Return a Python list containing subdomain of domain paramer
    eg:  ['app.arkiwi.org']
    """
    result_dict=dict()


    ot_cursor=ot_conn.cursor()
    ot_cursor.execute(subdomains_list_stmt, {'webserver':config['main']['webserver'].strip(" '\""),
                                             'domain':"%{}%".format(domain)})

    subdomains_res = ot_cursor.fetchall()
    ot_cursor.close()

    subdomains_filtered = [s[0].decode('utf-8') for s in subdomains_res
                               if not(s[0].decode('utf-8').startswith(ex_subdomains))]

    return subdomains_filtered


def get_domain_list(config,  ot_conn, dns_conn):
    """
    Return a dictionary of domains and properties
    eg:{'indivia.net': {manage_ns=True, domain_id=92}, 
    'metalabs.org': {managed_ns=False}}
    """
    result_dict=dict()

    ot_cursor=ot_conn.cursor()
    ot_cursor.execute(domains_list_stmt, {'webserver':config['main']['webserver'].strip(" '\"")})

    dns_cursor=dns_conn.cursor()
    for domain_barr, in ot_cursor:
        domain_name = domain_barr.decode("utf-8")
        try:
            dns_cursor.execute(domain_id_stmt, {'domain':domain_name})
        except Exception as e:
            logger.error(e)
            exit(-1)
        dns_res = dns_cursor.fetchall()
        if dns_cursor.rowcount == 1 :
            result_dict.update({domain_name: {'managed_ns':True, 'domain_id':dns_res[0][0]}})
        elif dns_cursor.rowcount == 0:
            result_dict.update({domain_name: {'managed_ns':False, 'domain_id':None}})
        else:
            logger.error('Unexpected result for domain {}'.format(domain_name))

    dns_cursor.close()
    ot_cursor.close()
    return result_dict

def  get_alias_list(config, dns_conn, query, aliases):
    """
    Return a list of domains to get the certificate for

    """
    result_list = list()

    dns_cursor=dns_conn.cursor()
    try:
        dns_cursor.execute(query, aliases)
    except Exception as e:
        logger.error(e)
        exit(-1)
    dns_res = dns_cursor.fetchall()
    result_list = [name[0].decode('utf-8') for name in dns_res]
    dns_cursor.close()
    return result_list


def acme_renew(config, pre_hook_cmd, post_hook_cmd, dryrun=False):

    args = config['certbot']['base_args']
    # args += " -m {} ".format(config['certbot']['email'])
    # args += "--server {} ".format(config['certbot']['server'])


    if dryrun:
        args += "--dry-run "

    if not pre_hook_cmd is None:
        args +=' --pre-hook "{}"'.format(pre_hook_cmd)

    if not post_hook_cmd is None:
        args +=' --post-hook "{}"'.format(post_hook_cmd)

    args += " renew"
    
    if dryrun:
        logging.info("{} {}".format(config['certbot']['bin'], args))
    else:
        os.system("{} {}".format(config['certbot']['bin'], args))

    return True



def acme_request(config, domain_name, acme_test='DNS-01', webroot=None, dryrun=False, domains_list=None):

    args = config['certbot']['base_args']
    args += " -m {} ".format(config['certbot']['email'])
    args += "--server {} ".format(config['certbot']['server'])

    if dryrun:
        args += "--dry-run "
    if acme_test == 'DNS-01':
        args += " --manual certonly "
        args += "--preferred-challenges dns-01 "
        args += "--manual-auth-hook {} ".format(config['certbot']['auth_hook'])
        args += "--manual-cleanup-hook {} ".format(config['certbot']['cleanup_hook'])
        args += "-d {},*.{}".format(domain_name, domain_name)
    elif acme_test == 'HTTP-01':
        args += " --webroot certonly "
        args += "--preferred-challenges http-01 "
        if webroot is None:
            args += "-w {}/{}/htdocs ".format(config['apache']['webroot'], domain_name)
        else:
            args += "-w {} ".format(webroot)

        if domains_list is None:
            args += "-d {}".format(domain_name)
        else:
            args += "--expand --cert-name {} ".format(domain_name)
            args += "-d {}".format(",".join(domains_list))
    else:
        logger.error('acme test {} not supported'.format(acme_test))
        return False
    if dryrun:
        logging.info("{} {}".format(config['certbot']['bin'], args))
    else:
        os.system("{} {}".format(config['certbot']['bin'], args))

    return True

def symlink_force(target, link_name):
    try:
        os.symlink(target, link_name)
    except Exception as e:
        if e.errno == errno.EEXIST:
            os.remove(link_name)
            os.symlink(target, link_name)
        else:
            raise e

def link_cert(config, source, dest, dryrun=False):
    src_name = os.path.join(config['certbot']['live_certificates_dir'], source)
    link_name = os.path.join(config['apache']['certificates_root'], dest)
    if dryrun:
        logger.info('{} -> {}'.format(link_name,src_name))
        return
    else:
        symlink_force(src_name, link_name)


if __name__ == '__main__':
    args, config = init_prog(sys.argv)

    dryrun=config['main'].getboolean('dryrun')
    service_reload = dict()

    if  dryrun:
        print("DRYRUN, nessun certificato verra' richiesto, nessun link/file creato o modificato")

    if args.renew:
        pre_hook_cmd = None
        post_hook_cmd = None
        logging.info('Renewing certificates ')
        if args.webmail or args.hosting or args.liste:
            post_hook_cmd = "systemctl reload apache2"
        elif args.smtp: 
            post_hook_cmd = "systemctl reload postfix"
        elif args.mbox:
            post_hook_cmd = "systemctl restart dovecot"

        logger.debug("post_hook_cmd: {}".format(post_hook_cmd))

        if acme_renew(config, pre_hook_cmd, post_hook_cmd, dryrun=dryrun):
            logger.info("Done renew")
            
    else:  
        # Fai le nuove richieste per i certificati
        
        # Caso speciale per le webmail
        if args.webmail:
            logging.info('Asking certificates for webmail')
            vhost_name = config['webmail']['vhost'].strip()
            webmails_list = ["webmail.{}".format(d.strip()) for d in config['webmail']['domains'].split(',') if len(d.strip())>0]
            logging.info('vhost {}, domains_list {}'.format(vhost_name, webmails_list))
            if acme_request(config, vhost_name, acme_test='HTTP-01', dryrun=dryrun, domains_list=webmails_list):
                link_cert(config, vhost_name, vhost_name, dryrun=dryrun)
            else:
                logger.error('Error asking certificate for {}'.format(vhost_name))
                
            # reload apache
            logger.info("Reloading apache")
            # ret = subprocess.run("systemctl reload apache2")
            ret = os.system("systemctl reload apache2")
            logger.info(ret)

        # Caso speciale per l'hosting
        if args.hosting:
            logging.info('Asking certificates for hosted web domains')
            ot_conn=connect_db(dict(config['ot_db']))
            dns_conn=connect_db(dict(config['dns_db']))
            # Subdomains da escludere
            ex_subdomains = tuple([s.strip() for s in config['main']['special_subdomains'].split(',') if len(s.strip())>0])
            domains_dict = get_domain_list(config, ot_conn, dns_conn)

            for domain_name, domain_feat in domains_dict.items():
                domain_feat['subdomains']=get_subdomain_list(config, domain_name, ot_conn, ex_subdomains=ex_subdomains)
                # Controlla se i nameserver sono gestiti da noi
                if domain_feat['managed_ns']:
                    # Nel caso il nameserver sia gestito, chiedi certificati per il dominio e la wildcard
                    logger.info('Get certificates for {}, *.{}'.format(domain_name, domain_name))
                    if acme_request(config, domain_name, acme_test='DNS-01', dryrun=dryrun):
                        link_cert(config, domain_name, domain_name, dryrun=dryrun)
                        # Crea il link per ogni subdomain
                        for subdomain in domain_feat['subdomains']:
                            link_cert(config, domain_name, subdomain, dryrun=dryrun)

                else:
                    # Nel caso i nameserver NON siano gestiti, allora chiedi un certificato per ogni sottodominio
                    # Crea il link per ogni subdomain
                    for subdomain in domain_feat['subdomains']:
                        logger.info('Get certificates for {}'.format(subdomain))
                        if acme_request(config, subdomain, acme_test='HTTP-01', dryrun=dryrun):
                            link_cert(config, subdomain, subdomain, dryrun=dryrun)

            ot_conn.close()
            dns_conn.close()

            # reload apache
            logger.info("Reloading apache")
            # ret = subprocess.run("systemctl reload apache2")
            ret = os.system("systemctl reload apache2")
            logger.info(ret)

            
        # Caso speciale per l'interfaccia di mailman
        if args.liste:
            logging.info('Asking certificates for liste.indivia.net')
            vhost_name = config['mailman']['vhost'].strip()
            liste_list = ["liste.{}".format(d.strip()) for d in config['mailman']['domains'].split(',') if len(d.strip())>0]
            if acme_request(config, vhost_name, acme_test='HTTP-01', dryrun=dryrun, domains_list=liste_list):
                link_cert(config, vhost_name, vhost_name, dryrun=dryrun)
            else:
                logger.error('Error asking certificate for {}'.format(vhost_name))
                
            # reload apache
            logger.info("Reloading apache")
            # ret = subprocess.run("systemctl reload apache2")
            ret = os.system("systemctl reload apache2")
            logger.info(ret)    

        
        # Caso speciale per il server POP/IMAP
        if args.mbox:
            dns_conn=connect_db(dict(config['dns_db']))
            logging.info('Asking certificates for POP/IMAP server')
            vhost_name = config['mail']['mbox_vhost'].strip()
            server_addresses = [s.strip() for s in config['mail']['mbox_server_addresses'].split(',') if len(s.strip())>0]
            mbox_fmt = ','.join(['%s'] * len(server_addresses))
            mbox_query = mbox_list_stmt.format(mbox_fmt)
            alias_list = get_alias_list(config, dns_conn, mbox_query, server_addresses)
            # Per usi futuri, aggiungo l'alias 'mail.indivia.net'
            alias_list.append('mail.indivia.net')
            logging.info('vhost {}, domains_list {}'.format(vhost_name, alias_list))
            if acme_request(config, vhost_name, acme_test='HTTP-01', webroot=config['mail']['mbox_webroot'].strip(),
                            dryrun=dryrun, domains_list=alias_list):
                # non e' richiesto il link, punto direttamente le configurazioni alle dir di letsencrypt
                # link_cert(config, vhost_name, vhost_name, dryrun=dryrun)
                service_reload['mbox'] = True
                pass
            else:
                logger.error('Error asking certificate for {}'.format(vhost_name))

            dns_conn.close()

            # restart dovecot
            logger.info("Restarting dovecot")
            # ret = subprocess.run("systemctl restart dovecot")
            ret = os.system("systemctl restart dovecot")
            logger.info(ret)


        # Caso speciale per il server SMTP
        if args.smtp:
            logging.info('Asking certificates for SMTP server')
            dns_conn=connect_db(dict(config['dns_db']))
            vhost_name = config['mail']['smtp_vhost'].strip()
            server_addresses = [s.strip() for s in config['mail']['smtp_server_addresses'].split(',') if len(s.strip())>0]
            smtp_fmt = ','.join(['%s'] * len(server_addresses))
            smtp_query = smtp_list_stmt.format(smtp_fmt)
            alias_list = get_alias_list(config, dns_conn, smtp_query, server_addresses)
            logging.info('vhost {}, domains_list {}'.format(vhost_name, alias_list))
            if acme_request(config, vhost_name, acme_test='HTTP-01', webroot=config['mail']['smtp_webroot'].strip(),
                            dryrun=dryrun, domains_list=alias_list):
                # non e' richiesto il link, punto direttamente le configurazioni alle dir di letsencrypt
                # link_cert(config, vhost_name, vhost_name, dryrun=dryrun)
                service_reload['smtp'] = True
                pass
            else:
                logger.error('Error asking certificate for {}'.format(vhost_name))

            dns_conn.close()
                
            # reload postfix
            logger.info("Restarting postfix")
            # ret = subprocess.run("systemctl reload postfix")
            ret = os.system("systemctl reload postfix")
            logger.info(ret)