diff --git a/setstatus.py b/setstatus.py index 2404ec7..f62a2e3 100755 --- a/setstatus.py +++ b/setstatus.py @@ -1,4 +1,4 @@ -#!/usr/bin/python3 +#!/usr/bin/env python3 # -*- coding: utf-8 -*- # file: setstatus.py @@ -21,206 +21,215 @@ 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 +# basic configuration +loglevel = logging.WARNING +formatstring = '%(asctime)s: %(levelname)s: %(message)s' +logging.basicConfig(format=formatstring, level=loglevel) -def main(): +class SetStatus: - 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' + def __init__(self): + """ + """ + self.status = None + self.config = None + self.log = None + self.context = None + self.connection = None + self.configfile = './setstatus.conf' + self.config = configparser.ConfigParser() + self.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) + def check_status(self): + """ + return: boolean + """ + if self.status in (0, 1): + self.log.debug('Set value to {}'.format(self.status)) + self.status = bytes([self.status]) + return True + return False - # check cli arguments - STATUS = check_arguments(argv) - if STATUS is None: - logging.error('No valid status given') - exit(2) + def set_config(self): + """ + """ + self.log = logging.getLogger() + # read config file + self.config.read_dict(self.default_config) + if not self.config.read(self.configfile): + self.log.warning('Configuration file {} not found or not ' + ' readable.'.format(self.configfile)) + self.log.warning('Using default values.') + # set loglevel + config_loglevel = self.config['general']['loglevel'] + default_loglevel = self.default_config['general']['loglevel'] + if not config_loglevel.lower() in ('critical', 'error','warning', + 'info', 'debug'): + self.log.warning('Invalid loglevel {} given. Using default ' + 'level {}.'.format(config_loglevel, + default_loglevel)) + self.config.set('general', 'loglevel', default_loglevel) + self.log.setLevel(config_loglevel.upper()) - # create ssl context - context = create_ssl_context(config) - if context is False: - exit(3) + def check_certs(self, certs): + """ + Check if certs readable. + """ + self.log.debug('Check certificates') + for certfile in certs: + if os.access(certfile, os.R_OK) is False: + self.log.error('Failed to read certificate: {}'.format( \ + certfile)) + return False + return True - # get connection - conn = create_ssl_connection(config, context) + def log_config(self): + """ + Logs the config if loglevel is debug. + """ + logging.debug('Using config:') + for section in self.config.sections(): + logging.debug('Section {}'.format(section)) + for i in self.config[section]: + logging.debug(' {}: {}'.format(i, self.config[section][i])) - # 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) + def create_ssl_context(self): + """ + """ + context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, + cafile=self.config['server']['cert']) + if not context: + self.log.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=self.config['client']['cert'], + keyfile=self.config['client']['key']) + except Exception as e: + self.log.error('Failed to load cert chain') + return False; + self.log.debug('SSL context created') + return context + + def create_ssl_socket(self, 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(self): + """ + """ + counter = 0 + conn = False + + while conn is False and counter < 5: + ssl_socket = self.create_ssl_socket(self.config, self.context) + if ssl_socket is False: + exit(4) + try: + self.log.debug('Connect {}: {}'.format(self.config['server']['host'], + int(self.config['server']['port']))) + conn = ssl_socket.connect((self.config['server']['host'], + int(self.config['server']['port']))) + except socket.timeout: + self.log.error('Connection timeout') + ssl_socket.close() + sleep(5) + counter += 1 + except Exception as e: + self.log.error('Connection failed: {}'.format(e)) + exit(5) + self.log.debug('Conection established') + self.log.debug('Peer certificate commonName: {}'.format( + ssl_socket.getpeercert()['subject'][5][0][1])) + self.log.debug('Peer certificate serialNumber: {}'.format( + ssl_socket.getpeercert()['serialNumber'])) + return ssl_socket -if __name__ == '__main__': - main() + def run(self, status): + """ + starts the engine. + param 1: integer + """ + self.status = status + + # read config + self.set_config() + + # check given status + if self.check_status() is False: + self.log.error('No valid status given') + exit(1) + + # log config if level is debug + if self.config['general']['loglevel'].lower() == 'debug': + self.log_config() + + # check certificates are readable + certs = (self.config['server']['cert'], + self.config['client']['cert'], + self.config['client']['key']) + if self.check_certs(certs) is False: + exit(2) + + # create ssl context + self.context = self.create_ssl_context() + if self.context is False: + exit(3) + + # get connection + self.connection = self.create_ssl_connection() + + # send status + try: + self.log.debug('Send new status: {}'.format(self.status)) + self.connection.send(self.status) + except Exception as e: + self.log.error('Error: {}'.format(e)) + exit(6) + try: + response = self.connection.recv(1) + self.log.debug('Server returns: {}'.format(response)) + if response == self.status: + self.log.info('Status sucessfull updated') + else: + self.log.error('Failed to update status') + self.log.debug('Disconnect from server') + except Exception as e: + self.log.error('Error: {}'.format(e)) + exit(7) +