#!/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()