forked from berhsi/doorstatus
cert-check added, fix typo
This commit is contained in:
+++
parent
b240de859c
commit
380661cc64
80
setstatus.py
80
setstatus.py
|
|
@ -12,6 +12,7 @@
|
|||
# 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
|
||||
|
|
@ -19,6 +20,18 @@ import configparser
|
|||
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
|
||||
|
|
@ -38,23 +51,7 @@ def check_arguments(argv):
|
|||
byte_value = None
|
||||
return byte_value
|
||||
|
||||
|
||||
def read_argument():
|
||||
'''
|
||||
Reads from stdin until the given value is valid. Convert the given
|
||||
string to a byte value and return this value.
|
||||
return: byte value
|
||||
'''
|
||||
status = None
|
||||
|
||||
while status is None:
|
||||
buf = input('Enter new status (0/1): ')
|
||||
if buf == '0' or buf == '1':
|
||||
status = bytes([int(buf)])
|
||||
logging.debug('Read status: {}'.format(status))
|
||||
return status
|
||||
|
||||
def print_config(config):
|
||||
def log_config(config):
|
||||
'''
|
||||
Logs the config with level debug.
|
||||
'''
|
||||
|
|
@ -70,7 +67,8 @@ def main():
|
|||
STATUS = None
|
||||
RESPONSE = None
|
||||
|
||||
loglevel = logging.DEBUG
|
||||
# basic configuration
|
||||
loglevel = logging.WARNING
|
||||
formatstring = '%(asctime)s: %(levelname)s: %(message)s'
|
||||
logging.basicConfig(format=formatstring, level=loglevel)
|
||||
|
||||
|
|
@ -82,14 +80,15 @@ def main():
|
|||
'server': {
|
||||
'host': 'localhost',
|
||||
'port': 10001,
|
||||
'cert': './certs/server.crt',
|
||||
'fqdn': 'server.status.kraut.space'
|
||||
'cert': './certs/server-pub.pem',
|
||||
'fqdn': 'kraut.space'
|
||||
},
|
||||
'client': {
|
||||
'cert': './certs/client.crt',
|
||||
'key': './certs/client.key'
|
||||
'cert': './certs/client-pub.pem',
|
||||
'key': './certs/client-key.pem'
|
||||
}
|
||||
}
|
||||
# read config file
|
||||
configfile = './setstatus.conf'
|
||||
config = configparser.ConfigParser()
|
||||
config.read_dict(default_config)
|
||||
|
|
@ -97,32 +96,43 @@ def main():
|
|||
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'] in ('critical', 'error', 'warning',
|
||||
'info', 'debug'):
|
||||
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)
|
||||
|
||||
print_config(config)
|
||||
# check certificates are readable
|
||||
certs = (config['server']['cert'],
|
||||
config['client']['cert'],
|
||||
config['client']['key'])
|
||||
if check_certs(certs) is False:
|
||||
sys.exit(1)
|
||||
|
||||
# check cli arguments
|
||||
STATUS = check_arguments(argv)
|
||||
while STATUS is None:
|
||||
STATUS = read_argument()
|
||||
if STATUS is None:
|
||||
logging.error('No valid status given')
|
||||
sys.exit(2)
|
||||
|
||||
# create ssl context
|
||||
context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH,
|
||||
cafile=config['server']['cert'])
|
||||
# use only cyphers for tls version 1.2 and 1.3
|
||||
context.set_ciphers('EECDH+AESGCM')
|
||||
context.set_ciphers('EECDH+AESGCM') # only ciphers for tls 1.2 and 1.3
|
||||
context.options |= getattr(ssl._ssl, 'OP_NO_COMPRESSION', 0)
|
||||
context.load_cert_chain(certfile=config['client']['cert'],
|
||||
keyfile=config['client']['key'])
|
||||
logging.debug('SSL context created')
|
||||
|
||||
# create socket and send status bit
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) as mySocket:
|
||||
logging.debug('Socket created')
|
||||
try:
|
||||
|
|
@ -138,10 +148,10 @@ def main():
|
|||
try:
|
||||
conn.connect((config['server']['host'], int(config['server']['port'])))
|
||||
except socket.timeout:
|
||||
logging.eror('Connection timeout')
|
||||
logging.error('Connection timeout')
|
||||
except Exception as e:
|
||||
logging.error('Connection failed: {}'.format(e))
|
||||
exit(1)
|
||||
exit(3)
|
||||
logging.debug('Peer certificate commonName: {}'.format(
|
||||
conn.getpeercert()['subject'][5][0][1]))
|
||||
logging.debug('Peer certificate serialNumber: {}'.format(
|
||||
|
|
@ -151,7 +161,7 @@ def main():
|
|||
conn.send(STATUS)
|
||||
except Exception as e:
|
||||
logging.error('Error: {}'.format(e))
|
||||
exit(2)
|
||||
exit(4)
|
||||
try:
|
||||
RESPONSE = conn.recv(1)
|
||||
logging.debug('Server returns: {}'.format(RESPONSE))
|
||||
|
|
@ -162,7 +172,7 @@ def main():
|
|||
logging.debug('Disconnect from server')
|
||||
except Exception as e:
|
||||
logging.error('Error: {}'.format(e))
|
||||
exit(3)
|
||||
exit(5)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
Loading…
Reference in a new issue