moved configuration handling to CryptoBoxSettings

added configuration validation
updated unittests.CryptoBox for new configuration validation
This commit is contained in:
lars 2006-09-08 11:02:27 +00:00
parent 0aa1f9f74b
commit 88fc900cc5
5 changed files with 281 additions and 134 deletions

View File

@ -15,18 +15,10 @@ if (ver_major < 2) or ((ver_major == 2) and (ver_minor < 4)):
sys.exit(1)
import CryptoBoxContainer
import types
from CryptoBoxExceptions import *
import re
import os
import unittest
import logging
import subprocess
from CryptoBoxExceptions import *
CONF_LOCATIONS = [
"./cryptobox.conf",
"~/.cryptobox.conf",
"/etc/cryptobox/cryptobox.conf"]
class CryptoBox:
@ -35,12 +27,13 @@ class CryptoBox:
put things like logging, conf and oter stuff in here,
that might be used by more classes, it will be passed on to them'''
def __init__(self, config_file=None):
self.__initLogging()
self.__initPreferences(config_file)
import CryptoBoxSettings
self.log = self.__getStartupLogger()
self.prefs = CryptoBoxSettings.CryptoBoxSettings(config_file)
self.__runTests()
def __initLogging(self):
def __getStartupLogger(self):
import logging
'''initialises the logging system
@ -53,119 +46,36 @@ class CryptoBox:
to the configured destination'''
## basicConfig(...) needs python >= 2.4
try:
self.log = logging.getLogger("CryptoBox")
log_handler = logging.getLogger("CryptoBox")
logging.basicConfig(
format='%(asctime)s %(module)s %(levelname)s %(message)s',
stderr=sys.stderr)
self.log.setLevel(logging.ERROR)
self.log.info("loggingsystem is up'n running")
log_handler.setLevel(logging.ERROR)
log_handler.info("loggingsystem is up'n running")
## from now on everything can be logged via self.log...
except:
raise CBEnvironmentError("couldn't initialise the loggingsystem. I give up.")
def __initPreferences(self, config_file):
try:
import configobj ## needed for reading and writing of the config file
except:
raise CBEnvironmentError("couldn't import 'configobj'! Try 'apt-get install python-configobj'.")
# search for the configuration file
if config_file is None:
# no config file was specified - we will look for it in the ususal locations
conf_file_list = [os.path.expanduser(f)
for f in CONF_LOCATIONS
if os.path.exists(os.path.expanduser(f))]
if not conf_file_list:
# no possible config file found in the usual locations
raise CBConfigUnavailableError()
config_file = conf_file_list[0]
else:
# a config file was specified (e.g. via command line)
if type(config_file) != types.StringType:
raise CBConfigUnavailableError("invalid config file specified: %s" % config_file)
if not os.path.exists(config_file):
raise CBConfigUnavailableError("could not find the specified configuration file (%s)" % config_file)
try:
self.cbxPrefs = configobj.ConfigObj(config_file)
if self.cbxPrefs:
self.log.info("found config: %s" % self.cbxPrefs.items())
else:
raise CBConfigUnavailableError("failed to load the config file: %s" % config_file)
except IOError:
raise CBConfigUnavailableError("unable to open the config file: %s" % config_file)
try:
try:
nameDB_file = self.cbxPrefs["Locations"]["NameDatabase"]
except KeyError:
raise CBConfigUndefinedError("Locations", "NameDatabase")
except SyntaxError:
raise CBConfigInvalidValueError("Locations", "NameDatabase", nameDB_file, "failed to interprete the filename of the name database correctly")
## create nameDB is necessary
if os.path.exists(nameDB_file):
self.nameDB = configobj.ConfigObj(nameDB_file)
else:
self.nameDB = configobj.ConfigObj(nameDB_file, create_empty=True)
## check if nameDB file was created successfully?
if not os.path.exists(nameDB_file):
raise CBEnvironmentError("failed to create name database (%s)" % nameDB_file)
# get the loglevel
try:
log_level = self.cbxPrefs["Log"]["Level"].upper()
log_level_avail = ["DEBUG", "INFO", "WARN", "ERROR"]
if not log_level in log_level_avail:
raise TypeError
except KeyError:
raise CBConfigUndefinedError("Log", "Level")
except TypeError:
raise CBConfigInvalidValueError("Log", "Level", log_level, "invalid log level: only %s are allowed" % log_level_avail)
try:
try:
new_handler = logging.FileHandler(self.cbxPrefs["Log"]["Details"])
except KeyError:
raise CBConfigUndefinedError("Log", "Details")
except IOError:
raise CBEnvironmentError("could not create the log file (%s)" % self.cbxPrefs["Log"]["Details"])
new_handler.setFormatter(logging.Formatter('%(asctime)s %(module)s %(levelname)s: %(message)s'))
self.log.addHandler(new_handler)
## do not call parent's handlers
self.log.propagate = False
## 'log_level' is a string -> use 'getattr'
self.log.setLevel(getattr(logging,log_level))
return log_handler
# do some initial checks
def __runTests(self):
self.__runTestRootPriv()
self.__runTestLocations()
def __runTestLocations(self):
"""check if all configured locations exist"""
try:
conf_locations = self.cbxPrefs["Locations"]
except KeyError:
raise CBConfigUndefinedError("Locations")
try:
for key in ["MountParentDir", "NameDatabase", "TemplateDir", "LangDir", "DocDir"]:
value = self.cbxPrefs["Locations"][key]
if not (os.path.exists(value) and os.access(value, os.R_OK)):
raise CBConfigInvalidValueError("Locations", key, value, "could not access")
except KeyError:
raise CBConfigUndefinedError("Locations", key)
def __runTestRootPriv(self):
"""try to run 'super' with 'CryptoBoxRootActions'"""
import subprocess
try:
devnull = open(os.devnull, "w")
except IOError:
raise CBEnvironmentError("could not open %s for writing!" % os.devnull)
try:
prog_super = self.cbxPrefs["Programs"]["super"]
prog_super = self.prefs["Programs"]["super"]
except KeyError:
raise CBConfigUndefinedError("Programs", "super")
try:
prog_rootactions = self.cbxPrefs["Programs"]["CryptoBoxRootActions"]
prog_rootactions = self.prefs["Programs"]["CryptoBoxRootActions"]
except KeyError:
raise CBConfigUndefinedError("Programs", "CryptoBoxRootActions")
try:
@ -175,7 +85,7 @@ class CryptoBox:
stderr = devnull,
args = [prog_super, prog_rootactions, "check"])
except OSError:
raise CBEnvironmentError("failed to execute 'super' (%s)" % self.cbxPrefs["Programs"]["super"])
raise CBEnvironmentError("failed to execute 'super' (%s)" % self.prefs["Programs"]["super"])
proc.wait()
if proc.returncode != 0:
raise CBEnvironmentError("failed to call CryptoBoxRootActions (%s) via 'super' - maybe you did not add the appropriate line to /etc/super.tab?" % prog_rootactions)
@ -199,11 +109,6 @@ class CryptoBoxProps(CryptoBox):
def __init__(self, config_file=None):
'''read config and fill class variables'''
CryptoBox.__init__(self, config_file)
#self.cbx_inheritance_test()
#print self.cbxPrefs.items()
####
self.containers = []
for device in self.__getAvailablePartitions():
if self.isDeviceAllowed(device):
@ -212,7 +117,8 @@ class CryptoBoxProps(CryptoBox):
def isDeviceAllowed(self, devicename):
"check if a device is white-listed for being used as cryptobox containers"
allowed = self.cbxPrefs["Main"]["AllowedDevices"]
import types
allowed = self.prefs["Main"]["AllowedDevices"]
if type(allowed) == types.StringType: allowed = [allowed]
for a_dev in allowed:
"remove double dots and so on ..."
@ -228,8 +134,8 @@ class CryptoBoxProps(CryptoBox):
"""
# return nothing if the currently selected log output is not a file
try:
if self.cbxPrefs["Log"]["Destination"].upper() != "FILE": return []
log_file = self.cbxPrefs["Log"]["Details"]
if self.prefs["Log"]["Destination"].upper() != "FILE": return []
log_file = self.prefs["Log"]["Details"]
except KeyError:
self.log.error("could not evaluate one of the following config settings: [Log]->Destination or [Log]->Details")
return []
@ -271,20 +177,19 @@ class CryptoBoxProps(CryptoBox):
return None
def setNameForUUID(self, uuid, name):
"assign a name to a uuid in the ContainerNameDatabase"
used_uuid = self.getUUIDForName(name)
"first remove potential conflicting uuid/name combination"
if used_uuid: del self.nameDB[used_uuid]
self.nameDB[uuid] = name
self.nameDB.write()
if used_uuid: del self.prefs.nameDB[used_uuid]
self.prefs.nameDB[uuid] = name
self.prefs.nameDB.write()
def getNameForUUID(self, uuid):
"get the name belonging to a specified key (usually the UUID of a fs)"
try:
return self.nameDB[uuid]
return self.prefs.nameDB[uuid]
except KeyError:
return None
@ -292,8 +197,8 @@ class CryptoBoxProps(CryptoBox):
def getUUIDForName(self, name):
""" get the key belonging to a value in the ContainerNameDatabase
this is the reverse action of 'getNameForUUID' """
for key in self.nameDB.keys():
if self.nameDB[key] == name: return key
for key in self.prefs.nameDB.keys():
if self.prefs.nameDB[key] == name: return key
"the uuid was not found"
return None
@ -303,7 +208,7 @@ class CryptoBoxProps(CryptoBox):
basenames from existing hdf files, that should are all available
languages'''
languages = []
for file in os.listdir(self.cbxPrefs["Locations"]["LangDir"]):
for file in os.listdir(self.prefs["Locations"]["LangDir"]):
if file.endswith(".hdf"): languages.append(file.rstrip(".hdf"))
if len(languages) < 1:
self.log.warn("No .hdf files found! The website won't render properly.")
@ -316,7 +221,7 @@ class CryptoBoxProps(CryptoBox):
doclangs = []
regpat = re.compile(r"^\w+$")
try:
doc_dir = self.cbxPrefs["Locations"]["DocDir"]
doc_dir = self.prefs["Locations"]["DocDir"]
except KeyError:
self.log.error("Could not find a configuration setting: [Locations]->DocDir - please check the config file")
return []

View File

@ -44,7 +44,7 @@ class CryptoBoxContainer:
self.device = device
self.cbox = cbox
self.log = logging.getLogger("CryptoBox")
self.Progs = self.cbox.cbxPrefs["Programs"]
self.Progs = self.cbox.prefs["Programs"]
self.__resetObject()
@ -190,7 +190,7 @@ class CryptoBoxContainer:
def_name = self.cbox.getNameForUUID(self.uuid)
if def_name: return def_name
"there is no name defined for this uuid - we will propose a good one"
prefix = self.cbox.cbxPrefs["Main"]["DefaultVolumePrefix"]
prefix = self.cbox.prefs["Main"]["DefaultVolumePrefix"]
unused_found = False
counter = 1
while not unused_found:
@ -293,7 +293,7 @@ class CryptoBoxContainer:
def __getMountPoint(self):
"return the name of the mountpoint of this volume"
return os.path.join(self.cbox.cbxPrefs["Locations"]["MountParentDir"], self.name)
return os.path.join(self.cbox.prefs["Locations"]["MountParentDir"], self.name)
def __mountLuks(self, password):
@ -508,7 +508,7 @@ class CryptoBoxContainer:
"luksFormat",
self.device,
"--batch-mode",
"--cipher", self.cbox.cbxPrefs["Main"]["DefaultCipher"],
"--cipher", self.cbox.prefs["Main"]["DefaultCipher"],
"--iter-time", "2000"])
proc.stdin.write(password)
(output, errout) = proc.communicate()
@ -559,9 +559,9 @@ class CryptoBoxContainer:
def __cleanMountDirs(self):
""" remove all unnecessary subdirs of the mount parent directory
this should be called for every (u)mount """
subdirs = os.listdir(self.cbox.cbxPrefs["Locations"]["MountParentDir"])
subdirs = os.listdir(self.cbox.prefs["Locations"]["MountParentDir"])
for dir in subdirs:
abs_dir = os.path.join(self.cbox.cbxPrefs["Locations"]["MountParentDir"], dir)
abs_dir = os.path.join(self.cbox.prefs["Locations"]["MountParentDir"], dir)
if (not os.path.islink(abs_dir)) and os.path.isdir(abs_dir) and (not os.path.ismount(abs_dir)):
os.rmdir(abs_dir)

View File

@ -0,0 +1,237 @@
import logging
import validate
import os
import CryptoBoxExceptions
try:
import configobj ## needed for reading and writing of the config file
except:
raise CryptoBoxExceptions.CBEnvironmentError("couldn't import 'configobj'! Try 'apt-get install python-configobj'.")
class CryptoBoxSettings:
CONF_LOCATIONS = [
"./cryptobox.conf",
"~/.cryptobox.conf",
"/etc/cryptobox/cryptobox.conf"]
def __init__(self, config_file=None):
self.log = logging.getLogger("CryptoBox")
config_file = self.__getConfigFileName(config_file)
self.log.info("loading config file: %s" % config_file)
self.prefs = self.__getPreferences(config_file)
self.__validateConfig()
self.__configureLogHandler()
self.__checkUnknownPreferences()
self.nameDB = self.__getNameDatabase()
def __getitem__(self, key):
"""redirect all requests to the 'prefs' attribute"""
return self.prefs[key]
def __getPreferences(self, config_file):
import StringIO
config_rules = StringIO.StringIO(self.validation_spec)
try:
prefs = configobj.ConfigObj(config_file, configspec=config_rules)
if prefs:
self.log.info("found config: %s" % prefs.items())
else:
raise CryptoBoxExceptions.CBConfigUnavailableError("failed to load the config file: %s" % config_file)
except IOError:
raise CryptoBoxExceptions.CBConfigUnavailableError("unable to open the config file: %s" % config_file)
return prefs
def __validateConfig(self):
result = self.prefs.validate(CryptoBoxSettingsValidator(), preserve_errors=True)
error_list = configobj.flatten_errors(self.prefs, result)
if not error_list: return
errorMsgs = []
for sections, key, text in error_list:
section_name = "->".join(sections)
if not text:
errorMsg = "undefined configuration value (%s) in section '%s'" % (key, section_name)
else:
errorMsg = "invalid configuration value (%s) in section '%s': %s" % (key, section_name, text)
errorMsgs.append(errorMsg)
raise CryptoBoxExceptions.CBConfigError, "\n".join(errorMsgs)
def __checkUnknownPreferences(self):
import StringIO
config_rules = configobj.ConfigObj(StringIO.StringIO(self.validation_spec), list_values=False)
self.__recursiveConfigSectionCheck("", self.prefs, config_rules)
def __recursiveConfigSectionCheck(self, section_path, section_config, section_rules):
"""should be called by '__checkUnknownPreferences' for every section
sends a warning message to the logger for every undefined (see validation_spec)
configuration setting
"""
for e in section_config.keys():
element_path = section_path + e
if e in section_rules.keys():
if isinstance(section_config[e], configobj.Section):
if isinstance(section_rules[e], configobj.Section):
self.__recursiveConfigSectionCheck(element_path + "->", section_config[e], section_rules[e])
else:
self.log.warn("configuration setting should be a value instead of a section name: %s" % element_path)
else:
if not isinstance(section_rules[e], configobj.Section):
pass # good - the setting is valid
else:
self.log.warn("configuration setting should be a section name instead of a value: %s" % element_path)
else:
self.log.warn("unknown configuration setting: %s" % element_path)
def __getNameDatabase(self):
try:
try:
nameDB_file = self.prefs["Locations"]["NameDatabase"]
except KeyError:
raise CryptoBoxExceptions.CBConfigUndefinedError("Locations", "NameDatabase")
except SyntaxError:
raise CryptoBoxExceptions.CBConfigInvalidValueError("Locations", "NameDatabase", nameDB_file, "failed to interprete the filename of the name database correctly")
## create nameDB is necessary
if os.path.exists(nameDB_file):
nameDB = configobj.ConfigObj(nameDB_file)
else:
nameDB = configobj.ConfigObj(nameDB_file, create_empty=True)
## check if nameDB file was created successfully?
if not os.path.exists(nameDB_file):
raise CryptoBoxExceptions.CBEnvironmentError("failed to create name database (%s)" % nameDB_file)
return nameDB
def __getConfigFileName(self, config_file):
# search for the configuration file
import types
if config_file is None:
# no config file was specified - we will look for it in the ususal locations
conf_file_list = [os.path.expanduser(f)
for f in self.CONF_LOCATIONS
if os.path.exists(os.path.expanduser(f))]
if not conf_file_list:
# no possible config file found in the usual locations
raise CryptoBoxExceptions.CBConfigUnavailableError()
config_file = conf_file_list[0]
else:
# a config file was specified (e.g. via command line)
if type(config_file) != types.StringType:
raise CryptoBoxExceptions.CBConfigUnavailableError("invalid config file specified: %s" % config_file)
if not os.path.exists(config_file):
raise CryptoBoxExceptions.CBConfigUnavailableError("could not find the specified configuration file (%s)" % config_file)
return config_file
def __configureLogHandler(self):
try:
log_level = self.prefs["Log"]["Level"].upper()
log_level_avail = ["DEBUG", "INFO", "WARN", "ERROR"]
if not log_level in log_level_avail:
raise TypeError
except KeyError:
raise CryptoBoxExceptions.CBConfigUndefinedError("Log", "Level")
except TypeError:
raise CryptoBoxExceptions.CBConfigInvalidValueError("Log", "Level", log_level, "invalid log level: only %s are allowed" % log_level_avail)
try:
try:
log_handler = logging.FileHandler(self.prefs["Log"]["Details"])
except KeyError:
raise CryptoBoxExceptions.CBConfigUndefinedError("Log", "Details")
except IOError:
raise CryptoBoxExceptions.CBEnvironmentError("could not create the log file (%s)" % self.prefs["Log"]["Details"])
log_handler.setFormatter(logging.Formatter('%(asctime)s %(module)s %(levelname)s: %(message)s'))
cbox_log = logging.getLogger("CryptoBox")
## remove previous handlers
cbox_log.handlers = []
## add new one
cbox_log.addHandler(log_handler)
## do not call parent's handlers
cbox_log.propagate = False
## 'log_level' is a string -> use 'getattr'
cbox_log.setLevel(getattr(logging,log_level))
## the logger named "CryptoBox" is configured now
validation_spec = """
[Main]
AllowedDevices = list(min=1)
DefaultVolumePrefix = string(min=1)
DefaultCipher = string(default="aes-cbc-essiv:sha256")
[Locations]
MountParentDir = directoryExists(default="/var/cache/cryptobox/mnt")
NameDatabase = fileWriteable(default="/var/cache/cryptobox/volumen_names.db")
TemplateDir = directoryExists(default="/usr/share/cryptobox/template")
LangDir = directoryExists(default="/usr/share/cryptobox/lang")
DocDir = directoryExists(default="/usr/share/doc/cryptobox/html")
[Log]
Level = option("debug", "info", "warn", "error", default="warn")
Destination = option("file", default="file")
Details = string(min=1)
[WebSettings]
Stylesheet = string(min=1)
Language = string(min=1, default="en")
DocLanguage = string(min=1, default="en")
[Programs]
cryptsetup = fileExecutable(default="/sbin/cryptsetup")
mkfs-data = fileExecutable(default="/sbin/mkfs.ext3")
mkfs-config = fileExecutable(default="/sbin/mkfs.ext2")
blkid = fileExecutable(default="/sbin/blkid")
mount = fileExecutable(default="/bin/mount")
umount = fileExecutable(default="/bin/umount")
super = fileExecutable(default="/usr/bin/super")
# this is the "program" name as defined in /etc/super.tab
CryptoBoxRootActions = string(min=1)
"""
class CryptoBoxSettingsValidator(validate.Validator):
def __init__(self):
validate.Validator.__init__(self)
self.functions["directoryExists"] = self.check_directoryExists
self.functions["fileExecutable"] = self.check_fileExecutable
self.functions["fileWriteable"] = self.check_fileWriteable
def check_directoryExists(self, value):
dir_path = os.path.abspath(value)
if not os.path.isdir(dir_path):
raise validate.VdtValueError("%s (not found)" % value)
if not os.access(dir_path, os.X_OK):
raise validate.VdtValueError("%s (access denied)" % value)
return dir_path
def check_fileExecutable(self, value):
file_path = os.path.abspath(value)
if not os.path.isfile(file_path):
raise validate.VdtValueError("%s (not found)" % value)
if not os.access(file_path, os.X_OK):
raise validate.VdtValueError("%s (access denied)" % value)
return file_path
def check_fileWriteable(self, value):
file_path = os.path.abspath(value)
if os.path.isfile(file_path):
if not os.access(file_path, os.W_OK):
raise validate.VdtValueError("%s (not found)" % value)
else:
parent_dir = os.path.dirname(file_path)
if os.path.isdir(parent_dir) and os.access(parent_dir, os.W_OK):
return file_path
raise validate.VdtValueError("%s (directory does not exist)" % value)
return file_path

View File

@ -12,7 +12,7 @@ class WebInterfaceSites:
import logging
self.cbox = CryptoBox.CryptoBoxProps()
self.log = logging.getLogger("CryptoBox")
self.prefs = self.cbox.cbxPrefs
self.prefs = self.cbox.prefs
self.__resetDataset()

View File

@ -3,6 +3,8 @@
import unittest
import sys
from CryptoBox import *
from CryptoBoxExceptions import *
import CryptoBoxSettings
class CryptoBoxPropsDeviceTests(unittest.TestCase):
import CryptoBox
@ -50,6 +52,8 @@ DocDir = ../doc/html
Level = debug
Destination = file
Details = %s/cryptobox.log
[WebSettings]
Stylesheet = /cryptobox-misc/cryptobox.css
[Programs]
blkid = /sbin/blkid
cryptsetup = /sbin/cryptsetup
@ -90,25 +94,26 @@ CryptoBoxRootActions = CryptoBoxRootActions
2) do we break, if no config file is there?
depending on the existence of a config file, only one of these conditions
can be checked - hints for more comprehensive tests are appreciated :) """
for a in self.CryptoBox.CONF_LOCATIONS:
for a in CryptoBoxSettings.CryptoBoxSettings.CONF_LOCATIONS:
if os.path.exists(a):
self.CryptoBox.CryptoBoxProps()
break # this skips the 'else' clause
else: self.assertRaises(CBConfigUnavailableError, self.CryptoBox.CryptoBoxProps)
self.assertRaises(CBConfigUnavailableError, self.CryptoBox.CryptoBoxProps,[])
def testBrokenConfigs(self):
"""Check various broken configurations"""
self.writeConfig("NameDatabase", "#out", filename=self.filenames["configFileBroken"])
self.assertRaises(CBConfigUndefinedError, self.CryptoBox.CryptoBoxProps,self.filenames["configFileBroken"])
self.writeConfig("Level", "#out", filename=self.filenames["configFileBroken"])
self.assertRaises(CBConfigUndefinedError, self.CryptoBox.CryptoBoxProps,self.filenames["configFileBroken"])
self.assertRaises(CBConfigError, self.CryptoBox.CryptoBoxProps,self.filenames["configFileBroken"])
self.writeConfig("Level", "Level = ho", filename=self.filenames["configFileBroken"])
self.assertRaises(CBConfigError, self.CryptoBox.CryptoBoxProps,self.filenames["configFileBroken"])
self.writeConfig("Details", "#out", filename=self.filenames["configFileBroken"])
self.assertRaises(CBConfigUndefinedError, self.CryptoBox.CryptoBoxProps,self.filenames["configFileBroken"])
self.assertRaises(CBConfigError, self.CryptoBox.CryptoBoxProps,self.filenames["configFileBroken"])
self.writeConfig("super", "super=/bin/invalid/no", filename=self.filenames["configFileBroken"])
self.assertRaises(CBEnvironmentError, self.CryptoBox.CryptoBoxProps,self.filenames["configFileBroken"])
self.assertRaises(CBConfigError, self.CryptoBox.CryptoBoxProps,self.filenames["configFileBroken"])
self.writeConfig("CryptoBoxRootActions", "#not here", filename=self.filenames["configFileBroken"])
self.assertRaises(CBConfigUndefinedError, self.CryptoBox.CryptoBoxProps,self.filenames["configFileBroken"])
self.assertRaises(CBConfigError, self.CryptoBox.CryptoBoxProps,self.filenames["configFileBroken"])
self.writeConfig("CryptoBoxRootActions", "CryptoBoxRootActions = /bin/false", filename=self.filenames["configFileBroken"])
self.assertRaises(CBEnvironmentError, self.CryptoBox.CryptoBoxProps,self.filenames["configFileBroken"])