doorstatus/statusd.py

327 lines
11 KiB
Python
Executable file

#!/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
import json
from time import time, ctime, 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()
value = strip_argument(value)
CONFIG[key] = value
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) == 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
'''
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')
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 some security options like
OP_NO_SSLv2 (SSLv3): they are insecure
PROTOCOL_TLS: only use tls
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 != False:
logger = logging.getLogger()
logger.setLevel(loglevel)
print_config(CONFIG)
if certs_readable(CONFIG) == 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
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 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.info('SSL connection established')
logging.info('commonName: {}'.format(conn.getpeercert()['subject'][5][0][1]))
logging.info('serialNumber: {}'.format(conn.getpeercert()['serialNumber']))
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:
logging.info('Exit')
exit()
except Exception as e:
logging.error('{}'.format(e))
continue
return 0
if __name__ == '__main__':
main()