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 time import sleep
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 if loglevel is 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 create_ssl_context(config):
    '''
    '''
    context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH,
                                         cafile=config['server']['cert'])
    if not context:
        logging.error('Failed to create SSL Context')
        return False
    context.set_ciphers('EECDH+AESGCM') # only ciphers for tls 1.2 and 1.3
    context.options |= getattr(ssl._ssl, 'OP_NO_COMPRESSION', 0)
    try:
        context.load_cert_chain(certfile=config['client']['cert'],
                                keyfile=config['client']['key'])
    except Exception as e:
        logging.error('Failed to load cert chain')
        return False;
    logging.debug('SSL context created')
    return context

def create_ssl_socket(config, context):
    '''
    '''
    bare_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
    if not bare_socket:
        logging.error('Socket creation failed')
        return False
    logging.debug('Socket created')
    try:
        secure_socket = context.wrap_socket(bare_socket, server_side=False,
                                   server_hostname=config['server']['fqdn'])
        logging.debug('Socket wrapped with SSL')
    except Exception as e:
        logging.error('Context wrapper failed: {}'.format(e))
        return False
    try:
        secure_socket.settimeout(float(config['general']['timeout']))
    except Exception as e:
        logging.debug('Failed to set timeout: {}'.format(e))
    return secure_socket

def create_ssl_connection(config, context):
    '''
    '''
    counter = 0
    conn = False

    while conn is False and counter < 5:
        ssl_socket = create_ssl_socket(config, context)
        if ssl_socket is False:
            exit(4)
        try:
            logging.debug('Connect {}: {}'.format(config['server']['host'],
                                                  int(config['server']['port'])))
            conn = ssl_socket.connect((config['server']['host'],
                                      int(config['server']['port'])))
        except socket.timeout:
            logging.error('Connection timeout')
            ssl_socket.close()
            sleep(5)
            counter += 1
        except Exception as e:
            logging.error('Connection failed: {}'.format(e))
            exit(5)
    logging.debug('Conection established')
    logging.debug('Peer certificate commonName: {}'.format(
                  ssl_socket.getpeercert()['subject'][5][0][1]))
    logging.debug('Peer certificate serialNumber: {}'.format(
                  ssl_socket.getpeercert()['serialNumber']))
    return ssl_socket


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:
        exit(1)

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

    # create ssl context
    context = create_ssl_context(config)
    if context is False:
        exit(3)

    # get connection
    conn = create_ssl_connection(config, context)

    # send status
    try:
        logging.debug('Send new status: {}'.format(STATUS))
        conn.send(STATUS)
    except Exception as e:
        logging.error('Error: {}'.format(e))
        exit(6)
    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(7)


if __name__ == '__main__':
    main()