Hilde/doorstatus
1
0
Fork 0
forked from Krautspace/doorstatus
Code Pull requests Activity

Umwandlung in eine Klasse

Browse source
This commit is contained in:
+++ 2020-10-22 23:58:17 +02:00
parent cc77f32afa
commit a2a533ac65
1 changed files with 202 additions and 193 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

263
setstatus.py
View file

@ -1,4 +1,4 @@
#!/usr/bin/python3 #!/usr/bin/env python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# file: setstatus.py # file: setstatus.py
@ -21,69 +21,116 @@ from time import sleep
from sys import exit, argv from sys import exit, argv
# basic configuration
loglevel = logging.WARNING
formatstring = '%(asctime)s: %(levelname)s: %(message)s'
logging.basicConfig(format=formatstring, level=loglevel)
def check_certs(certs):
''' class SetStatus:
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'
}
}
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
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())
def check_certs(self, certs):
"""
Check if certs readable. Check if certs readable.
''' """
logging.debug('Check certificates') self.log.debug('Check certificates')
for certfile in certs: for certfile in certs:
if os.access(certfile, os.R_OK) is False: if os.access(certfile, os.R_OK) is False:
logging.error('Failed to read certificate: {}'.format(certfile)) self.log.error('Failed to read certificate: {}'.format( \
certfile))
return False return False
return True return True
def check_arguments(argv): def log_config(self):
''' """
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. Logs the config if loglevel is debug.
''' """
logging.debug('Using config:') logging.debug('Using config:')
for section in config.sections(): for section in self.config.sections():
logging.debug('Section {}'.format(section)) logging.debug('Section {}'.format(section))
for i in config[section]: for i in self.config[section]:
logging.debug(' {}: {}'.format(i, config[section][i])) logging.debug(' {}: {}'.format(i, self.config[section][i]))
def create_ssl_context(config): def create_ssl_context(self):
''' """
''' """
context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH,
cafile=config['server']['cert']) cafile=self.config['server']['cert'])
if not context: if not context:
logging.error('Failed to create SSL Context') self.log.error('Failed to create SSL Context')
return False return False
context.set_ciphers('EECDH+AESGCM') # only ciphers for tls 1.2 and 1.3 context.set_ciphers('EECDH+AESGCM') # only ciphers for tls 1.2 and 1.3
context.options |= getattr(ssl._ssl, 'OP_NO_COMPRESSION', 0) context.options |= getattr(ssl._ssl, 'OP_NO_COMPRESSION', 0)
try: try:
context.load_cert_chain(certfile=config['client']['cert'], context.load_cert_chain(certfile=self.config['client']['cert'],
keyfile=config['client']['key']) keyfile=self.config['client']['key'])
except Exception as e: except Exception as e:
logging.error('Failed to load cert chain') self.log.error('Failed to load cert chain')
return False; return False;
logging.debug('SSL context created') self.log.debug('SSL context created')
return context return context
def create_ssl_socket(config, context): def create_ssl_socket(self, config, context):
''' """
''' """
bare_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) bare_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
if not bare_socket: if not bare_socket:
logging.error('Socket creation failed') logging.error('Socket creation failed')
@ -102,125 +149,87 @@ def create_ssl_socket(config, context):
logging.debug('Failed to set timeout: {}'.format(e)) logging.debug('Failed to set timeout: {}'.format(e))
return secure_socket return secure_socket
def create_ssl_connection(config, context): def create_ssl_connection(self):
''' """
''' """
counter = 0 counter = 0
conn = False conn = False
while conn is False and counter < 5: while conn is False and counter < 5:
ssl_socket = create_ssl_socket(config, context) ssl_socket = self.create_ssl_socket(self.config, self.context)
if ssl_socket is False: if ssl_socket is False:
exit(4) exit(4)
try: try:
logging.debug('Connect {}: {}'.format(config['server']['host'], self.log.debug('Connect {}: {}'.format(self.config['server']['host'],
int(config['server']['port']))) int(self.config['server']['port'])))
conn = ssl_socket.connect((config['server']['host'], conn = ssl_socket.connect((self.config['server']['host'],
int(config['server']['port']))) int(self.config['server']['port'])))
except socket.timeout: except socket.timeout:
logging.error('Connection timeout') self.log.error('Connection timeout')
ssl_socket.close() ssl_socket.close()
sleep(5) sleep(5)
counter += 1 counter += 1
except Exception as e: except Exception as e:
logging.error('Connection failed: {}'.format(e)) self.log.error('Connection failed: {}'.format(e))
exit(5) exit(5)
logging.debug('Conection established') self.log.debug('Conection established')
logging.debug('Peer certificate commonName: {}'.format( self.log.debug('Peer certificate commonName: {}'.format(
ssl_socket.getpeercert()['subject'][5][0][1])) ssl_socket.getpeercert()['subject'][5][0][1]))
logging.debug('Peer certificate serialNumber: {}'.format( self.log.debug('Peer certificate serialNumber: {}'.format(
ssl_socket.getpeercert()['serialNumber'])) ssl_socket.getpeercert()['serialNumber']))
return ssl_socket return ssl_socket
def main(): def run(self, status):
"""
starts the engine.
param 1: integer
"""
self.status = status
STATUS = None # read config
RESPONSE = None self.set_config()
# basic configuration # check given status
loglevel = logging.WARNING if self.check_status() is False:
formatstring = '%(asctime)s: %(levelname)s: %(message)s' self.log.error('No valid status given')
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) exit(1)
# check cli arguments # log config if level is debug
STATUS = check_arguments(argv) if self.config['general']['loglevel'].lower() == 'debug':
if STATUS is None: self.log_config()
logging.error('No valid status given')
# 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) exit(2)
# create ssl context # create ssl context
context = create_ssl_context(config) self.context = self.create_ssl_context()
if context is False: if self.context is False:
exit(3) exit(3)
# get connection # get connection
conn = create_ssl_connection(config, context) self.connection = self.create_ssl_connection()
# send status # send status
try: try:
logging.debug('Send new status: {}'.format(STATUS)) self.log.debug('Send new status: {}'.format(self.status))
conn.send(STATUS) self.connection.send(self.status)
except Exception as e: except Exception as e:
logging.error('Error: {}'.format(e)) self.log.error('Error: {}'.format(e))
exit(6) exit(6)
try: try:
RESPONSE = conn.recv(1) response = self.connection.recv(1)
logging.debug('Server returns: {}'.format(RESPONSE)) self.log.debug('Server returns: {}'.format(response))
if RESPONSE == STATUS: if response == self.status:
logging.info('Status sucessfull updated') self.log.info('Status sucessfull updated')
else: else:
logging.error('Failed to update status') self.log.error('Failed to update status')
logging.debug('Disconnect from server') self.log.debug('Disconnect from server')
except Exception as e: except Exception as e:
logging.error('Error: {}'.format(e)) self.log.error('Error: {}'.format(e))
exit(7) exit(7)
if __name__ == '__main__':
main()