doorstatus/statusd.py

#!/usr/bin/python3

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

# server, which listens for ipv4 connections at port 10001. now with ssl
# encrypted connection and client side authentication.

import socket
import ssl
import os
import logging
import json
from time import time, sleep
from sys import exit


def read_config(CONFIGFILE, CONFIG):
    '''
    reads the given config file and sets the values are founded.
    param 1: string
    param 2: dictionary
    return: boolean
    '''
    logging.debug('Read configfile {}'.format(CONFIGFILE))
    if os.access(CONFIGFILE, os.R_OK):
        logging.debug('Configfile is readable')
        with open(CONFIGFILE, 'r') as config:
            logging.debug('Configfile successfull read')
            for line in config.readlines():
                if not line[0] in ('#', ';', '\n', '\r'):
                    key, value = (line.strip().split('='))
                    key = strip_argument(key).upper()
                    if key in CONFIG.keys():
                        value = strip_argument(value)
                        CONFIG[key] = value
                    else:
                        pass
    else:
        logging.error('Failed to read {}'.format(CONFIGFILE))
        logging.error('Using default values')
        return False
    return True


def certs_readable(config):
    '''
    checks at start, if the needed certificates defined (no nullstring) and
    readable.
    param 1: dictionary
    return: boolean
    '''
    for i in (config['SERVER_KEY'], config['SERVER_CERT'],
              config['CLIENT_CERT']):
        if i == '' or os.access(i, os.R_OK) is False:
            logging.error('Cant read {}'.format(i))
            return False
    return True


def strip_argument(argument):
    '''
    Becomes a string and strips at first whitespaces, second apostrops and
    returns the clear string.
    param 1: string
    return: string
    '''
    argument = argument.strip()
    argument = argument.strip('"')
    argument = argument.strip("'")
    return argument


def print_config(CONFIG):
    '''
    Prints the used configuration, if loglevel ist debug.
    param 1: dictionary
    return: boolean (allways true)
    '''
    logging.debug('Using config:')
    for i in CONFIG.keys():
        logging.debug('{}: {}'.format(i, CONFIG[i]))
    return True


def print_ciphers(cipherlist):
    '''
    This function prints the list of the allowed ciphers.
    param1: dictionary
    return: boolean
    '''
    print('Available ciphers')
    for i in cipherlist:
        print('\n')
        for j in i.keys():
            print('{}: {}'.format(j, i[j]))
    print('\n')
    return True


def display_peercert(cert):
    '''
    This function displays the values of a given certificate.
    param1: dictionary
    return: boolean
    '''
    for i in cert.keys():
        print('{}:'.format(i))
        if i in ('subject', 'issuer'):
            for j in cert[i]:
                print('\t{}'.format(j))
        else:
            print('\t{}'.format(cert[i]))
    return True


def receive_buffer_is_valid(raw_data):
    '''
    checks, if the received buffer from the connection is valid or not.
    param 1: byte
    return: boolean
    '''
    if raw_data == b'\x00' or raw_data == b'\x01':
        logging.debug('Argument is valid: {}'.format(raw_data))
        return True
    else:
        logging.debug('Argument is not valid: {}'.format(raw_data))
        return False


def change_status(raw_data, api):
    '''
    Becomes the received byte and the path to API file. Grabs the content of
    the API with read_api() and replaces "open" and "lastchange". Write all
    lines back to API file.
    param 1: byte
    param 2: string
    return: boolean
    '''

    logging.debug('Change status API')
    data = read_api(api)
    if data is not False:
        status, timestamp = set_values(raw_data)
        if os.access(api, os.W_OK):
            logging.debug('API file is writable')
            with open(api, 'w') as api_file:
                logging.debug('API file open successfull')
                data["state"]["open"] = status
                data["state"]["lastchange"] = timestamp
                try:
                    json.dump(data, api_file, indent=4)
                except Exception as e:
                    logging.error('Failed to change API file')
                    logging.error('{}'.format(e))
                logging.debug('API file changed')
        else:
            logging.error('API file is not writable. Wrong permissions?')
            return False
        logging.info('Status successfull changed to {}'.format(status))
        return True
    return False


def read_api(api):
    '''
    Reads the API file in an buffer und returns the buffer. If anything goes
    wrong, it returns False - otherwise it returns the buffer.
    param 1: string
    return: string or boolean
    '''
    logging.debug('Open API file: {}'.format(api))
    if os.access(api, os.R_OK):
        logging.debug('API is readable')
        with open(api, 'r') as api_file:
            logging.debug('API opened successfull')
            try:
                api_json_data = json.load(api_file)
                logging.debug('API file read successfull')
            except Exception as e:
                logging.error('Failed to read API file(): {}'.format(e))
                return False
        return (api_json_data)
    logging.error('Failed to read API file')
    return False


def set_values(raw_data):
    '''
    Create a timestamp, changes the value of the given byte into a string
    and returns both.
    param 1: byte
    return: tuple
    '''
    timestamp = str(time()).split('.')[0]
    if raw_data == b'\x01':
        status = "true"
    else:
        status = "false"
    logging.debug('Set values for timestamp: {} and status: {}'.format(
                    timestamp, status))
    return (status, timestamp)


def read_loglevel(CONFIG):
    '''
    The function translates the value string from config verbosity option to
    a valid logging option.
    param1: dictionary
    return: boolean or integer
    '''
    if CONFIG['VERBOSITY'] == 'critical':
        loglevel = logging.CRITICAL
    elif CONFIG['VERBOSITY'] == 'error':
        loglevel = logging.ERROR
    elif CONFIG['VERBOSITY'] == 'warning':
        loglevel = logging.WARNING
    elif CONFIG['VERBOSITY'] == 'info':
        loglevel = logging.INFO
    elif CONFIG['VERBOSITY'] == 'debug':
        loglevel = logging.DEBUG
    else:
        loglevel = False
    return(loglevel)


def main():
    '''
    The main function - opens a socket, create a ssl context, load certs and
    listen for connections. at ssl context we set only one available cipher
    suite and disable compression.
    OP_NO_COMPRESSION: prevention against crime attack
    OP_DONT_ISERT_EMPTY_FRAGMENTS: prevention agains cbc 4 attack
    (cve-2011-3389)
    '''

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

    CONFIG = {
        'HOST': 'localhost',
        'PORT': 10001,
        'SERVER_CERT': './server.crt',
        'SERVER_KEY':  './server.key',
        'CLIENT_CERT': './client.crt',
        'TIMEOUT': 3.0,
        'API': './api',
        'API_TEMPLATE': './api_template',
        'VERBOSITY': 'warning'
        }
    CONFIG_FILE = './statusd.conf'
    read_config(CONFIG_FILE, CONFIG)
    loglevel = read_loglevel(CONFIG)
    if loglevel is not False:
        logger = logging.getLogger()
        logger.setLevel(loglevel)
    else:
        loglevel = logging.WARNING
        logger = logging.getLogger()
        logger.setLevel(loglevel)
        logging.warning('Invalid value for loglevel. Set default value')

    print_config(CONFIG)

    # todo: zertifikate sollten nur lesbar sein!

    if certs_readable(CONFIG) is False:
        logging.error('Cert check failed\nExit')
        exit()

    context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    context.verify_mode = ssl.CERT_REQUIRED
    context.load_cert_chain(certfile=CONFIG['SERVER_CERT'],
                            keyfile=CONFIG['SERVER_KEY'])
    context.load_verify_locations(cafile=CONFIG['CLIENT_CERT'])
    context.set_ciphers('EECDH+AESGCM')   # only ciphers for tls 1.2 and 1.3
    context.options = ssl.OP_CIPHER_SERVER_PREFERENCE
    # ssl + kompression = schlecht
    context.options |= getattr(ssl._ssl, 'OP_NO_COMPRESSION', 0)
    logging.debug('SSL context created')
    # print_ciphers(context.get_ciphers())

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) as mySocket:
        logging.debug('Socket created')
        try:
            mySocket.bind((CONFIG['HOST'], int(CONFIG['PORT'])))
            mySocket.listen(5)
            logging.info('Listen on {} at Port {}'.format(CONFIG['HOST'],
                                                          CONFIG['PORT']))
        except Exception as e:
            logging.error('unable to bind and listen')
            logging.error('{}'.format(e))
            exit()
        while True:
            try:
                fromSocket, fromAddr = mySocket.accept()
                logging.info('Client connected: {}:{}'.format(fromAddr[0],
                                                              fromAddr[1]))
                try:
                    fromSocket.settimeout(float(CONFIG['TIMEOUT']))
                    logging.debug('Connection timeout set to {}'.format(
                                    CONFIG['TIMEOUT']))
                except Exception:
                    logging.error('Canot set timeout to {}'.format(
                                    CONFIG['TIMEOUT']))
                    logging.error('Use default value: 3.0')
                    fromSocket.settimeout(3.0)
                try:
                    conn = context.wrap_socket(fromSocket, server_side=True)
                    conn.settimeout(3.0)
                    # display_peercert(conn.getpeercert())
                    logging.debug('Connection established')
                    logging.debug('Peer certificate commonName: {}'.format
                                  (conn.getpeercert()['subject'][5][0][1]))
                    logging.debug('Peer certificate serialNumber: {}'.format
                                  (conn.getpeercert()['serialNumber']))
                except socket.timeout:
                    logging.error('Socket timeout')
                except Exception as e:
                    logging.error('Connection failed: {}'.format(e))
                raw_data = conn.recv(1)
                if receive_buffer_is_valid(raw_data) is True:
                    if change_status(raw_data, CONFIG['API']) is True:
                        logging.debug('Send {} back'.format(raw_data))
                        conn.send(raw_data)
                    # change_status returns false:
                    else:
                        logging.info('Failed to change status')
                        if conn:
                            conn.send(b'\x03')
                # receive_handle returns false:
                else:
                    logging.info('Invalid argument recived: {}'.format(
                                 raw_data))
                    logging.debug('Send {} back'.format(b'\x03'))
                    if conn:
                        conn.send(b'\x03')
                sleep(0.1)  # protection against dos
            except KeyboardInterrupt:
                logging.info('Exit')
                exit()
            except Exception as e:
                logging.error('{}'.format(e))
                continue
    return 0


if __name__ == '__main__':
    main()