#!/usr/bin/python3 # file: statusd.py # date: 26.07.2019 # email: berhsi@web.de # server, who listen for ipv4 connections at port 10001. now with ssl # encrypted connection and client side authentication. import socket import ssl import os import logging from time import time, ctime, sleep from sys import exit, byteorder 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() value = strip_argument(value) CONFIG[key] = value logging.debug('Set {} to {}'.format(key, value)) else: logging.error('Failed to read {}'.format(CONFIGFILE)) logging.error('Using default values') 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 display_peercert(cert): for i in cert.keys(): print(i) for j in cert[i]: print('\t{}'.format(j)) return 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 replace_entry(line, new): ''' The function becomes two strings and replaces the second part of the first string from ":" until end with the second string. Than appends a "," and returns the result. !!! Todo: needs exception handling !!! param 1: string param 2: string return: string ''' array = line.split(':') logging.debug('Replace {} with {}'.format(array[1], new)) array[1] = ''.join((new, ',')) line = ':'.join((array[0], array[1])) return line 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 ''' edit = False logging.debug('Change status API') data = read_api(api) if data != 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') for line in data.splitlines(): if line.strip().startswith('"state":'): edit = True elif edit == True and line.strip().startswith('"open":'): line = replace_entry(line, status) elif edit == True and line.strip().startswith('"lastchange":'): line = replace_entry(line, timestamp) edit = False try: api_file.write(line) api_file.write('\n') except Exception as e: logging.error('Failed to write line to API file') logging.error('Line: {}'.format(line)) 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_data = api_file.read() logging.debug('API read successfull') except Exception as e: logging.error('Failed to read API file(): {}'.format(e)) return False return (api_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" return (status, timestamp) def main(): ''' The main function - opens a socket, create a ssl context, load certs and listen for connections. ''' 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': 'info' } CONFIG_FILE = './statusd.conf' FINGERPRINT = \ '35:8E:35:FA:58:0A:DD:2B:C8:6A:F9:EA:A3:7B:10:F5:62:89:AB:D0:AB:53:3E:B5:8B:AB:E1:23:CF:93:F5:F9' loglevel = logging.DEBUG logging.basicConfig(format='%(levelname)s: %(message)s', level=loglevel) read_config(CONFIG_FILE, CONFIG) print_config(CONFIG) context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) context.options &= ~ssl.PROTOCOL_TLS 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.options &= ~ssl.OP_NO_SSLv3 logging.debug('SSL context created') 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 as e: 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) # display_peercert(conn.getpeercert()) logging.debug('SSL established. Peer: {}'.format(conn.getpeercert())) except Exception as e: logging.error('SSL handshake failed: {}'.format(e)) raw_data = conn.recv(1) if receive_buffer_is_valid(raw_data) == True: if change_status(raw_data, CONFIG['API']) == 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') # recive_handle returns false: else: logging.info('Inalid 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: print('\rExit') logging.info('Exit') exit() except Exception as e: logging.error('{}'.format(e)) continue if __name__ == '__main__': main()