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 ssl
import socket
import logging
import configparser
from sys import exit, argv


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 read_argument():
    '''
    Reads from stdin until the given value is valid. Convert the given
    string to a byte value and return this value.
    return: byte value
    '''
    status = None

    while status is None:
        buf = input('Enter new status (0/1): ')
        if buf == '0' or buf == '1':
            status = bytes([int(buf)])
            logging.debug('Read status: {}'.format(status))
    return status

def print_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

    loglevel = logging.DEBUG
    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.crt',
            'fqdn': 'server.status.kraut.space'
        },
        'client': {
            'cert': './certs/client.crt',
            'key': './certs/client.key'
        }
    }
    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.')

    logger = logging.getLogger()
    if not config['general']['loglevel'] 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())

    print_config(config)

    STATUS = check_arguments(argv)
    while STATUS is None:
        STATUS = read_argument()

    context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH,
                                         cafile=config['server']['cert'])
    # use only cyphers for tls version 1.2 and 1.3
    context.set_ciphers('EECDH+AESGCM')
    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')

    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.eror('Connection timeout')
        except Exception as e:
            logging.error('Connection failed: {}'.format(e))
            exit(1)
        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(2)
        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(3)


if __name__ == '__main__':
    main()