doorstatus/setstatus.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-

# file: setstatus.py
# date: 26.07.2019
# email: berhsi@web.de

# Setstatus.py is part of doorstatus - a programm to deal with the
# krautspaces doorstatus.

# client, who connects to the statusserver at port 10001 to update the
# krautspace door status. If no status is given as argument, he reads from
# stdin until input is 0 or 1.

import os
import ssl
import socket
import logging
import configparser
from sys import exit, argv



def check_certs(certs):
    '''
    Check if certs readable.
    '''
    logging.debug('Check certificates')
    for certfile in certs:
        if os.access(certfile, os.R_OK) is False:
            logging.error('Failed to read certificate: {}'.format(certfile))
            return False
    return True

def check_arguments(argv):
    '''
    Checks length and validity of command line argument vectors. If there is
    no argument or argument is not valid, it returns None. Otherwise it
    converts the string value into a byte value.
    param 1: array of strings
    return: None or byte value
    '''
    if len(argv) == 1:
        byte_value = None
    else:
        if argv[1].strip() == '0' or argv[1].strip() == '1':
            i = int(argv[1].strip())
            logging.debug('Set value to {}'.format(i))
            byte_value = bytes([i])
        else:
            byte_value = None
    return byte_value

def log_config(config):
    '''
    Logs the config with level debug.
    '''
    logging.debug('Using config:')
    for section in config.sections():
        logging.debug('Section {}'.format(section))
        for i in config[section]:
            logging.debug('  {}: {}'.format(i, config[section][i]))


def main():

    STATUS = None
    RESPONSE = None

    # basic configuration
    loglevel = logging.WARNING
    formatstring = '%(asctime)s: %(levelname)s: %(message)s'
    logging.basicConfig(format=formatstring, level=loglevel)

    default_config = {
        'general': {
            'timeout': 5.0,
            'loglevel': 'warning'
        },
        'server': {
            'host': 'localhost',
            'port': 10001,
            'cert': './certs/server-pub.pem',
            'fqdn': 'kraut.space'
        },
        'client': {
            'cert': './certs/client-pub.pem',
            'key': './certs/client-key.pem'
        }
    }
    # read config file
    configfile = './setstatus.conf'
    config = configparser.ConfigParser()
    config.read_dict(default_config)
    if not config.read(configfile):
        logging.warning('Configuration file {} not found or not readable.'.format(
                        configfile))
        logging.warning('Using default values.')
    # set loglevel
    logger = logging.getLogger()
    if not config['general']['loglevel'].lower() in ('critical', 'error',
                                                     'warning', 'info',
                                                     'debug'):
        logging.warning('Invalid loglevel %s given. Using default level %s.',
                        config['general']['loglevel'],
                        default_config['general']['loglevel'])
        config.set('general', 'loglevel', default_config['general']['loglevel'])
    logger.setLevel(config['general']['loglevel'].upper())
    # log config if level is debug
    if config['general']['loglevel'].lower() == 'debug':
        log_config(config)

    # check certificates are readable
    certs = (config['server']['cert'],
             config['client']['cert'],
             config['client']['key'])
    if check_certs(certs) is False:
        sys.exit(1)

    # check cli arguments
    STATUS = check_arguments(argv)
    if STATUS is None:
        logging.error('No valid status given')
        sys.exit(2)

    # create ssl context
    context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH,
                                         cafile=config['server']['cert'])
    context.set_ciphers('EECDH+AESGCM') # only ciphers for tls 1.2 and 1.3
    context.options |= getattr(ssl._ssl, 'OP_NO_COMPRESSION', 0)
    context.load_cert_chain(certfile=config['client']['cert'],
                            keyfile=config['client']['key'])
    logging.debug('SSL context created')

    # create socket and send status bit
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) as mySocket:
        logging.debug('Socket created')
        try:
            conn = context.wrap_socket(mySocket, server_side=False,
                                       server_hostname=config['server']['fqdn'])
            logging.debug('Connection wrapped with ssl.context')
        except Exception as e:
            logging.error('Context wrapper failed: {}'.format(e))
        try:
            conn.settimeout(float(config['general']['timeout']))
        except Exception as e:
            logging.debug('Failed to set timeout: {}'.format(e))
        try:
            conn.connect((config['server']['host'], int(config['server']['port'])))
        except socket.timeout:
            logging.error('Connection timeout')
        except Exception as e:
            logging.error('Connection failed: {}'.format(e))
            exit(3)
        logging.debug('Peer certificate commonName: {}'.format(
                      conn.getpeercert()['subject'][5][0][1]))
        logging.debug('Peer certificate serialNumber: {}'.format(
                      conn.getpeercert()['serialNumber']))
        try:
            logging.debug('Send new status: {}'.format(STATUS))
            conn.send(STATUS)
        except Exception as e:
            logging.error('Error: {}'.format(e))
            exit(4)
        try:
            RESPONSE = conn.recv(1)
            logging.debug('Server returns: {}'.format(RESPONSE))
            if RESPONSE == STATUS:
                logging.info('Status sucessfull updated')
            else:
                logging.error('Failed to update status')
            logging.debug('Disconnect from server')
        except Exception as e:
            logging.error('Error: {}'.format(e))
            exit(5)


if __name__ == '__main__':
    main()