transition to "logging" instead of "CryptoBoxLogger" finished

names of cryptobox.conf settings changed
cryptsetup interface for CryptoBoxRootActions implemented
This commit is contained in:
lars 2006-08-22 06:55:07 +00:00
parent 12868beb21
commit f95f33ba71
5 changed files with 119 additions and 303 deletions

View file

@ -13,6 +13,7 @@ import re
import os import os
import sys import sys
import unittest import unittest
import logging
CONF_LOCATIONS = [ CONF_LOCATIONS = [
"./cryptobox.conf", "./cryptobox.conf",
@ -35,20 +36,19 @@ class CryptoBox:
'''initialises the logging system '''initialises the logging system
use it with: 'self.log.[debug|info|warning|error|critical](logmessage)' use it with: 'self.log.[debug|info|warning|error|critical](logmessage)'
from all classes the inherited from CryptoBox all classes should get the logging instance during __init__:
self.log = logging.getLogger("CryptoBox")
TODO/RFC: read the logfile from the config - this is a hen-egg problem first we output all warnings/errors to stderr
i would prefer start logging to stdout, read the config and redirect as soon as we opened the config file successfully, we redirect debug output
logging to the logfile found in the config [a] to the configured destination'''
[l]: ok'''
## basicConfig(...) needs python >= 2.4 ## basicConfig(...) needs python >= 2.4
try: try:
logging.basicConfig(level=logging.DEBUG, logging.basicConfig(
format='%(asctime)s %(module)s %(levelname)s %(message)s', format='%(asctime)s %(module)s %(levelname)s %(message)s',
filename='./cryptobox.log', stream = sys.stderr)
filemode='a') self.log = logging.getLogger("CryptoBox")
self.log = logging.getLogger("CryptoBoxProps") self.log.setLevel(logging.WARN)
self.log.info("loggingsystem is up'n running") self.log.info("loggingsystem is up'n running")
## from now on everything can be logged via self.log... ## from now on everything can be logged via self.log...
except: except:
@ -98,6 +98,21 @@ class CryptoBox:
self.log.error("Error during parsing of name database file (%s).\n" % (nameDB_file, )) self.log.error("Error during parsing of name database file (%s).\n" % (nameDB_file, ))
sys.exit(1) sys.exit(1)
# TODO: check if nameDB file was created successfully? # TODO: check if nameDB file was created successfully?
# get the loglevel
try:
log_level = self.cbxPrefs["Log"]["Level"].upper()
if not log_level in ["DEBUG", "INFO", "WARN", "ERROR"]:
self.log.error("invalid log level: %s" % self.cbxPrefs["Log"]["Level"])
sys.exit(1)
self.log.setLevel(getattr(logging, log_level))
self.log.addHandler(logging.FileHandler(self.cbxPrefs["Log"]["Details"]))
except TypeError:
self.log.error("invalid log level: %s" % self.cbxPrefs["Log"]["Level"])
sys.exit(1)
except IOError:
self.log.error("could not open logfile: %s" % self.cbxPrefs["Log"]["Details"])
sys.exit(1)
# RFC: what is this method useful for? # RFC: what is this method useful for?

View file

@ -5,6 +5,7 @@ except:
print "Couldn't import 'subprocess'. You need a python version >= 2.4" print "Couldn't import 'subprocess'. You need a python version >= 2.4"
import os import os
import re import re
import logging
"""exceptions: """exceptions:
VolumeIsActive VolumeIsActive
@ -37,7 +38,7 @@ class CryptoBoxContainer:
def __init__(self, device, cbox): def __init__(self, device, cbox):
self.device = device self.device = device
self.cbox = cbox self.cbox = cbox
self.debugMessage = self.cbox.debugMessage self.log = logging.getLogger("CryptoBox")
self.Progs = self.cbox.cbxPrefs["Programs"] self.Progs = self.cbox.cbxPrefs["Programs"]
self.__resetObject() self.__resetObject()
@ -103,9 +104,7 @@ class CryptoBoxContainer:
try: try:
devnull = open(os.devnull, "w") devnull = open(os.devnull, "w")
except IOError: except IOError:
self.debugMessage( self.log.warn("Could not open %s" % (os.devnull, ))
CryptoBoxLogger.DebugLevels["warn"],
"Could not open %s" % (os.devnull, ))
"remove any potential open luks mapping" "remove any potential open luks mapping"
self.__umountLuks() self.__umountLuks()
"create the luks header" "create the luks header"
@ -123,7 +122,7 @@ class CryptoBoxContainer:
(output, errout) = proc.communicate() (output, errout) = proc.communicate()
if proc.returncode != 0: if proc.returncode != 0:
errorMsg = "Could not add a new luks key: %s - %s" % (output.strip(), errout.strip(), ) errorMsg = "Could not add a new luks key: %s - %s" % (output.strip(), errout.strip(), )
self.debugMessage(CryptoBoxLogger.DebugLevels["error"], errorMsg) self.log.error(errorMsg)
raise "ChangePasswordError", errorMsg raise "ChangePasswordError", errorMsg
keys_found = re.search(r'key slot (\d{1,3}) unlocked', output).groups() keys_found = re.search(r'key slot (\d{1,3}) unlocked', output).groups()
if keys_found: if keys_found:
@ -145,7 +144,7 @@ class CryptoBoxContainer:
proc.wait() proc.wait()
if proc.returncode != 0: if proc.returncode != 0:
errorMsg = "Could not remove the old luks key: %s" % (proc.stderr.read().strip(), ) errorMsg = "Could not remove the old luks key: %s" % (proc.stderr.read().strip(), )
self.debugMessage(CryptoBoxLogger.DebugLevels["error"], errorMsg) self.log.error(errorMsg)
raise "ChangePasswordError", errorMsg raise "ChangePasswordError", errorMsg
@ -190,9 +189,7 @@ class CryptoBoxContainer:
try: try:
devnull = open(os.devnull, "w") devnull = open(os.devnull, "w")
except IOError: except IOError:
self.debugMessage( self.warn("Could not open %s" % (os.devnull, ))
CryptoBoxLogger.DebugLevels["warn"],
"Could not open %s" % (os.devnull, ))
proc = subprocess.Popen( proc = subprocess.Popen(
shell=False, shell=False,
stdin=None, stdin=None,
@ -207,9 +204,7 @@ class CryptoBoxContainer:
proc.wait() proc.wait()
result = proc.stdout.read().strip() result = proc.stdout.read().strip()
if proc.returncode != 0: if proc.returncode != 0:
self.debugMessage( self.log.warn("retrieving of partition type via 'blkid' failed: %s" % (proc.stderr.read().strip(), ))
CryptoBoxLogger.DebugLevels["warn"],
"retrieving of partition type via 'blkid' failed: %s" % (proc.stderr.read().strip(), ))
return None return None
devnull.close() devnull.close()
if result: return result if result: return result
@ -233,9 +228,7 @@ class CryptoBoxContainer:
try: try:
devnull = open(os.devnull, "w") devnull = open(os.devnull, "w")
except IOError: except IOError:
self.debugMessage( self.log.warn("Could not open %s" % (os.devnull, ))
CryptoBoxLogger.DebugLevels["warn"],
"Could not open %s" % (os.devnull, ))
proc = subprocess.Popen( proc = subprocess.Popen(
shell=False, shell=False,
stdin=None, stdin=None,
@ -250,9 +243,7 @@ class CryptoBoxContainer:
proc.wait() proc.wait()
output = proc.stdout.read().strip() output = proc.stdout.read().strip()
if proc.returncode != 0: if proc.returncode != 0:
self.debugMessage( self.log.warn("retrieving of partition type via 'blkid' failed: %s" % (proc.stderr.read().strip(), ))
CryptoBoxLogger.DebugLevels["warn"],
"retrieving of partition type via 'blkid' failed: %s" % (proc.stderr.read().strip(), ))
return None return None
devnull.close() devnull.close()
return output return output
@ -264,9 +255,7 @@ class CryptoBoxContainer:
try: try:
devnull = open(os.devnull, "w") devnull = open(os.devnull, "w")
except IOError: except IOError:
self.debugMessage( self.log.warn("Could not open %s" % (os.devnull, ))
CryptoBoxLogger.DebugLevels["warn"],
"Could not open %s" % (os.devnull, ))
proc = subprocess.Popen( proc = subprocess.Popen(
shell = False, shell = False,
stdin = None, stdin = None,
@ -296,15 +285,13 @@ class CryptoBoxContainer:
try: try:
devnull = open(os.devnull, "w") devnull = open(os.devnull, "w")
except IOError: except IOError:
self.debugMessage( self.log.warn("Could not open %s" % (os.devnull, ))
CryptoBoxLogger.DebugLevels["warn"],
"Could not open %s" % (os.devnull, ))
self.__cleanMountDirs() self.__cleanMountDirs()
if not os.path.exists(self.__getMountPoint()): if not os.path.exists(self.__getMountPoint()):
os.mkdir(self.__getMountPoint()) os.mkdir(self.__getMountPoint())
if not os.path.exists(self.__getMountPoint()): if not os.path.exists(self.__getMountPoint()):
errorMsg = "Could not create mountpoint (%s)" % (self.__getMountPoint(), ) errorMsg = "Could not create mountpoint (%s)" % (self.__getMountPoint(), )
self.debugMessage("error", errorMsg) self.log.error(errorMsg)
raise "MountError", errorMsg raise "MountError", errorMsg
proc = subprocess.Popen( proc = subprocess.Popen(
shell = False, shell = False,
@ -321,7 +308,7 @@ class CryptoBoxContainer:
(output, errout) = proc.communicate() (output, errout) = proc.communicate()
if proc.returncode != 0: if proc.returncode != 0:
errorMsg = "Could not open the luks mapping: %s" % (errout.strip(), ) errorMsg = "Could not open the luks mapping: %s" % (errout.strip(), )
self.debugMessage(CryptoBoxLogger.DebugLevels["warn"], errorMsg) self.log.warn(errorMsg)
raise "MountError", errorMsg raise "MountError", errorMsg
proc = subprocess.Popen( proc = subprocess.Popen(
shell = False, shell = False,
@ -335,7 +322,7 @@ class CryptoBoxContainer:
proc.wait() proc.wait()
if proc.returncode != 0: if proc.returncode != 0:
errorMsg = "Could not mount the filesystem: %s" % (proc.stderr.read().strip(), ) errorMsg = "Could not mount the filesystem: %s" % (proc.stderr.read().strip(), )
self.debugMessage(CryptoBoxLogger.DebugLevels["warn"], errorMsg) self.log.warn(errorMsg)
raise "MountError", errorMsg raise "MountError", errorMsg
devnull.close() devnull.close()
@ -346,9 +333,7 @@ class CryptoBoxContainer:
try: try:
devnull = open(os.devnull, "w") devnull = open(os.devnull, "w")
except IOError: except IOError:
self.debugMessage( self.log.warn("Could not open %s" % (os.devnull, ))
CryptoBoxLogger.DebugLevels["warn"],
"Could not open %s" % (os.devnull, ))
if self.isMounted(): if self.isMounted():
proc = subprocess.Popen( proc = subprocess.Popen(
shell = False, shell = False,
@ -359,7 +344,7 @@ class CryptoBoxContainer:
proc.wait() proc.wait()
if proc.returncode != 0: if proc.returncode != 0:
errorMsg = "Could not umount the filesystem: %s" % (proc.stderr.read().strip(), ) errorMsg = "Could not umount the filesystem: %s" % (proc.stderr.read().strip(), )
self.debugMessage(CryptoBoxLogger.DebugLevels["warn"], errorMsg) self.log.warn(errorMsg)
raise "MountError", errorMsg raise "MountError", errorMsg
if os.path.exists(os.path.join(self.__dmDir, self.name)): if os.path.exists(os.path.join(self.__dmDir, self.name)):
proc = subprocess.Popen( proc = subprocess.Popen(
@ -375,7 +360,7 @@ class CryptoBoxContainer:
proc.wait() proc.wait()
if proc.returncode != 0: if proc.returncode != 0:
errorMsg = "Could not remove the luks mapping: %s" % (proc.stderr.read().strip(), ) errorMsg = "Could not remove the luks mapping: %s" % (proc.stderr.read().strip(), )
self.debugMessage(CryptoBoxLogger.DebugLevels["warn"], errorMsg) self.log.warn(errorMsg)
raise "MountError", errorMsg raise "MountError", errorMsg
devnull.close() devnull.close()
@ -387,15 +372,13 @@ class CryptoBoxContainer:
try: try:
devnull = open(os.devnull, "w") devnull = open(os.devnull, "w")
except IOError: except IOError:
self.debugMessage( self.log.warn("Could not open %s" % (os.devnull, ))
CryptoBoxLogger.DebugLevels["warn"],
"Could not open %s" % (os.devnull, ))
self.__cleanMountDirs() self.__cleanMountDirs()
if not os.path.exists(self.__getMountPoint()): if not os.path.exists(self.__getMountPoint()):
os.mkdir(self.__getMountPoint()) os.mkdir(self.__getMountPoint())
if not os.path.exists(self.__getMountPoint()): if not os.path.exists(self.__getMountPoint()):
errorMsg = "Could not create mountpoint (%s)" % (self.__getMountPoint(), ) errorMsg = "Could not create mountpoint (%s)" % (self.__getMountPoint(), )
self.debugMessage("error", errorMsg) self.log.error(errorMsg)
raise "MountError", errorMsg raise "MountError", errorMsg
proc = subprocess.Popen( proc = subprocess.Popen(
shell = False, shell = False,
@ -409,7 +392,7 @@ class CryptoBoxContainer:
proc.wait() proc.wait()
if proc.returncode != 0: if proc.returncode != 0:
errorMsg = "Could not mount the filesystem: %s" % (proc.stderr.read().strip(), ) errorMsg = "Could not mount the filesystem: %s" % (proc.stderr.read().strip(), )
self.debugMessage(CryptoBoxLogger.DebugLevels["warn"], errorMsg) self.log.warn(errorMsg)
raise "MountError", errorMsg raise "MountError", errorMsg
devnull.close() devnull.close()
@ -420,9 +403,7 @@ class CryptoBoxContainer:
try: try:
devnull = open(os.devnull, "w") devnull = open(os.devnull, "w")
except IOError: except IOError:
self.debugMessage( self.log.warn("Could not open %s" % (os.devnull, ))
CryptoBoxLogger.DebugLevels["warn"],
"Could not open %s" % (os.devnull, ))
if self.isMounted(): if self.isMounted():
proc = subprocess.Popen( proc = subprocess.Popen(
shell = False, shell = False,
@ -436,7 +417,7 @@ class CryptoBoxContainer:
proc.wait() proc.wait()
if proc.returncode != 0: if proc.returncode != 0:
errorMsg = "Could not umount the filesystem: %s" % (proc.stderr.read().strip(), ) errorMsg = "Could not umount the filesystem: %s" % (proc.stderr.read().strip(), )
self.debugMessage(CryptoBoxLogger.DebugLevels["warn"], errorMsg) self.log.warn(errorMsg)
raise "MountError", errorMsg raise "MountError", errorMsg
devnull.close() devnull.close()
@ -450,9 +431,7 @@ class CryptoBoxContainer:
try: try:
devnull = open(os.devnull, "w") devnull = open(os.devnull, "w")
except IOError: except IOError:
self.debugMessage( self.log.warn("Could not open %s" % (os.devnull, ))
CryptoBoxLogger.DebugLevels["warn"],
"Could not open %s" % (os.devnull, ))
proc = subprocess.Popen( proc = subprocess.Popen(
shell = False, shell = False,
stdin = None, stdin = None,
@ -464,7 +443,7 @@ class CryptoBoxContainer:
proc.wait() proc.wait()
if proc.returncode != 0: if proc.returncode != 0:
errorMsg = "Could not create the filesystem: %s" % (proc.stderr.read().strip(), ) errorMsg = "Could not create the filesystem: %s" % (proc.stderr.read().strip(), )
self.debugMessage(CryptoBoxLogger.DebugLevels["error"], errorMsg) self.log.error(errorMsg)
raise "CreateError", errorMsg raise "CreateError", errorMsg
devnull.close() devnull.close()
@ -480,9 +459,7 @@ class CryptoBoxContainer:
try: try:
devnull = open(os.devnull, "w") devnull = open(os.devnull, "w")
except IOError: except IOError:
self.debugMessage( self.log.warn("Could not open %s" % (os.devnull, ))
CryptoBoxLogger.DebugLevels["warn"],
"Could not open %s" % (os.devnull, ))
"remove any potential open luks mapping" "remove any potential open luks mapping"
self.__umountLuks() self.__umountLuks()
"create the luks header" "create the luks header"
@ -502,7 +479,7 @@ class CryptoBoxContainer:
(output, errout) = proc.communicate() (output, errout) = proc.communicate()
if proc.returncode != 0: if proc.returncode != 0:
errorMsg = "Could not create the luks header: %s" % (errout.strip(), ) errorMsg = "Could not create the luks header: %s" % (errout.strip(), )
self.debugMessage("error", errorMsg) self.log.error(errorMsg)
raise "CreateError", errorMsg raise "CreateError", errorMsg
"open the luks container for mkfs" "open the luks container for mkfs"
proc = subprocess.Popen( proc = subprocess.Popen(
@ -520,7 +497,7 @@ class CryptoBoxContainer:
(output, errout) = proc.communicate() (output, errout) = proc.communicate()
if proc.returncode != 0: if proc.returncode != 0:
errorMsg = "Could not open the new luks mapping: %s" % (errout.strip(), ) errorMsg = "Could not open the new luks mapping: %s" % (errout.strip(), )
self.debugMessage(CryptoBoxLogger.DebugLevels["error"], errorMsg) self.log.error(errorMsg)
raise "CreateError", errorMsg raise "CreateError", errorMsg
"make the filesystem" "make the filesystem"
proc = subprocess.Popen( proc = subprocess.Popen(
@ -536,7 +513,7 @@ class CryptoBoxContainer:
self.__umountLuks() self.__umountLuks()
if proc.returncode != 0: if proc.returncode != 0:
errorMsg = "Could not create the filesystem: %s" % (proc.stderr.read().strip(), ) errorMsg = "Could not create the filesystem: %s" % (proc.stderr.read().strip(), )
self.debugMessage(CryptoBoxLogger.DebugLevels["error"], errorMsg) self.log.error(errorMsg)
"remove the luks mapping" "remove the luks mapping"
raise "CreateError", errorMsg raise "CreateError", errorMsg
devnull.close() devnull.close()
@ -548,6 +525,7 @@ class CryptoBoxContainer:
subdirs = os.listdir(self.cbox.cbxPrefs["System"]["MountParentDir"]) subdirs = os.listdir(self.cbox.cbxPrefs["System"]["MountParentDir"])
for dir in subdirs: for dir in subdirs:
abs_dir = os.path.join(self.cbox.cbxPrefs["System"]["MountParentDir"], dir) abs_dir = os.path.join(self.cbox.cbxPrefs["System"]["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) 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

@ -1,222 +0,0 @@
'''
this class will not be used anymore and can be removed from svn
manages logging events of the CryptoBox
'''
import sys
import os
import syslog
import unittest
class CryptoBoxLogger:
'''
handles logging events and prints them e.g. to a logfile
'''
DebugLevels = {
"debug": syslog.LOG_DEBUG,
"info": syslog.LOG_INFO,
"notice": syslog.LOG_NOTICE,
"warn": syslog.LOG_WARNING,
"error": syslog.LOG_ERR,
"crit": syslog.LOG_CRIT,
"alert": syslog.LOG_ALERT,
"emerg": syslog.LOG_EMERG}
DebugDestinations = {"file":0, "syslog":1}
def __init__(self, level, destination, args=None):
"""create a CryptoBoxLogger object and connect it to an output destination
level: string (debug/info/notice/warn/error/crit/alert/emerg) or syslog level
destination: the string "file" or "syslog"
args: e.g. the name of the logfile or syslog facility
"""
try:
try:
destination = int(destination)
except Exception:
try:
destination = self.DebugDestinations[destination]
except KeyError:
raise "LoggerError"
if not destination in self.DebugDestinations.values(): raise "LoggerError"
except "LoggerError":
errorMsg = "Invalid debug destination: %s" % destination
sys.stderr.write(errorMsg + "\n")
raise "LoggerError", errorMsg
try:
try:
level = int(level)
except Exception:
try:
level = self.DebugLevels[level]
except KeyError:
raise "LoggerError"
if not level in self.DebugLevels.values(): raise "LoggerError"
except "LoggerError":
errorMsg = "Invalid debug level: %s" % level
sys.stderr.write(errorMsg + "\n")
raise "LoggerError", errorMsg
self.debug_level = level
if destination == self.DebugDestinations["file"]:
self.logFunc = self.message2file
if args is not None:
self.logFile = args
else:
self.logFile = '/var/log/cryptobox.log'
try:
fsock = open(self.logFile, "a")
fsock.close()
except IOError:
errorMsg ="Unable to open logfile (%s) for writing." % (self.logFile,)
sys.stderr.write(errorMsg + "\n")
raise "LoggerError", errorMsg
elif destination == self.DebugDestinations["syslog"]:
self.logFunc = self.message2syslog
if args is None:
syslog.openlog("CryptoBox", 0, syslog.LOG_USER)
else:
syslog.openlog("CryptoBox", 0, args)
else:
errorMsg = "Invalid logging facility: %d." % (facility, )
sys.stderr.write(errorMsg + "\n")
raise "LoggerError", errorMsg
def printMessage(self, msg_level, text):
if msg_level is None: msg_level = self.DebugLevels["debug"]
"convert debuglevel from string to int, if necessary"
try:
msg_level = int(msg_level)
except ValueError:
try:
msg_level = self.DebugLevels[msg_level]
except KeyError:
errorMsg = "Invalid debug level: %s" % msg_level
sys.stderr.write(errorMsg + "\n")
raise "LoggerError", errorMsg
if not msg_level in self.DebugLevels.values():
errorMsg = "Invalid debug level: %s" % msg_level
sys.stderr.write(errorMsg + "\n")
raise "LoggerError", errorMsg
if text is None:
errorMsg = "Empty debug message - this is not allowed"
sys.stderr.write(errorMsg + "\n")
raise "LoggerError", errorMsg
if msg_level <= self.debug_level:
self.logFunc(text, msg_level)
def message2file(self, text, level):
# "level" gets ignored (but syslog needs it)
try:
log_sock = open(self.logFile, "a")
try:
log_sock.writelines("[CryptoBox] - %s\n" % (text, ))
log_sock.close()
return
except IOError:
errorMsg = "Unable to write messages to logfile (%s)." % (self.logFile, )
sys.stderr.write(errorMsg + "\n")
raise "LoggerError", errorMsg
except IOError:
errorMsg = "Unable to open logfile (%s) for writing." % (self.logFile, )
sys.stderr.write("[CryptoBox] - %s\n" % (errorMsg, ))
raise "LoggerError", errorMsg
def message2syslog(self, text, level):
syslog_level = [self.DebugLevels[e]
for e in self.DebugLevels.keys()
if self.DebugLevels[e] == level
][0]
syslog.syslog(syslog_level, text)
# ********************* test class **********************
class CryptoBoxLoggerTest(unittest.TestCase):
logFile = "/tmp/cbox-test.log"
def setUp(self):
if os.path.exists(self.logFile): os.remove(self.logFile)
def tearDown(self):
if os.path.exists(self.logFile): os.remove(self.logFile)
def testInit(self):
"""Initialization should fail for invalid parameters"""
try:
CryptoBoxLogger(syslog.LOG_ERR, 0)
except "LoggerError":
CryptoBoxLogger(syslog.LOG_ERR, 0, self.logFile)
os.remove(self.logFile)
CryptoBoxLogger("info", "file", self.logFile)
self.assertRaises("LoggerError", CryptoBoxLogger, "invalid", 0, self.logFile)
self.assertRaises("LoggerError", CryptoBoxLogger, syslog.LOG_ERR, "invalid", self.logFile)
self.assertRaises("LoggerError", CryptoBoxLogger, 3353, 0, self.logFile)
self.assertRaises("LoggerError", CryptoBoxLogger, -1, 0, self.logFile)
self.assertRaises("LoggerError", CryptoBoxLogger, syslog.LOG_INFO, self.logFile)
self.assertRaises("LoggerError", CryptoBoxLogger, syslog.LOG_CRIT, -1, self.logFile)
self.assertRaises("LoggerError", CryptoBoxLogger, None, 0, self.logFile)
self.assertRaises("LoggerError", CryptoBoxLogger, syslog.LOG_WARNING, None, self.logFile)
self.assertRaises("LoggerError", CryptoBoxLogger, syslog.LOG_EMERG, 0, "/no/existing/path")
def testOutputParams(self):
"""Output should fail for invalid parameters"""
cb = CryptoBoxLogger(syslog.LOG_ERR, 0, self.logFile)
self.assertRaises("LoggerError", cb.printMessage, 3353, "Ausgabe")
self.assertRaises("LoggerError", cb.printMessage, -1, "Ausgabe")
self.assertRaises("LoggerError", cb.printMessage, "invalid", "Ausgabe")
self.assertRaises("LoggerError", cb.printMessage, syslog.LOG_DEBUG, None)
def testFile(self):
"""Do not write messages below specified priority to a file"""
cb = CryptoBoxLogger(syslog.LOG_ERR, 0, self.logFile)
content1 = self.readFile()
self.assertEquals(content1, "")
cb.printMessage(syslog.LOG_ERR, "Ausgabe")
content2 = self.readFile()
self.assertNotEqual(content1, content2)
cb.printMessage(syslog.LOG_DEBUG, "Ausgabe")
self.assertEquals(content2, self.readFile())
cb.printMessage(syslog.LOG_CRIT, "Ausgabe")
self.assertNotEqual(content2, self.readFile())
def testSyslog(self):
"""Check syslog output"""
cb = CryptoBoxLogger(syslog.LOG_DEBUG, "syslog")
cb.printMessage(syslog.LOG_DEBUG, "just a verification check")
"""sorry - we do not have a way to check, if something was written somewhere
so we cannot do other checks beside initialization and writing"""
def readFile(self):
fd = None
try:
fd = open(self.logFile, "r")
text = fd.read()
fd.close()
except IOError:
if fd is not None: fd.close()
text = None
return text
# *************** unit testing *********************
if __name__ == "__main__":
try:
devnull = open(os.devnull, "w")
sys.stderr = devnull
except IOError:
pass
unittest.main()

View file

@ -15,6 +15,7 @@ import sys
import subprocess import subprocess
import pwd import pwd
import grp import grp
import types
allowedProgs = { allowedProgs = {
"sfdisk": "/sbin/sfdisk", "sfdisk": "/sbin/sfdisk",
@ -50,7 +51,7 @@ def isWriteableBlock(device):
trustGIDs.append(allowedGID) trustGIDs.append(allowedGID)
DEF_TYPES = { "pipe":1, "char":2, "dir":4, "block":6, "file":8, "link":10, "socket":12} DEF_TYPES = { "pipe":1, "char":2, "dir":4, "block":6, "file":8, "link":10, "socket":12}
if dev_type != DEF_TYPES["block"]: return False if dev_type != DEF_TYPES["block"]: return False
# replace this access check by "os.access" ## could the following check be replaced by os.access?
"does the owner id match?" "does the owner id match?"
if owner_uid == trustUID: if owner_uid == trustUID:
"is the write bit (2) set for the owner?" "is the write bit (2) set for the owner?"
@ -62,9 +63,49 @@ def isWriteableBlock(device):
def run_cryptsetup(device, args): def run_cryptsetup(device, args):
print "ok - you are free to call cryptsetup ..." """execute cryptsetup as root
print " not yet implemented ..."
return True @args: list of arguments - they will be treated accordingly to the first element
of this list (the action)"""
if not args: raise "WrongArguments", "no action for cryptsetup supplied"
if type(args) != types.ListType: raise "WrongArguments", "invalid arguments supplied"
try:
action = args[0]
del args[0]
cs_args = [allowedProgs["cryptsetup"]]
cs_args.extend(args)
if action == "luksFormat":
cs_args.append(action)
cs_args.append(device)
elif action == "luksUUID":
cs_args.append(action)
cs_args.append(device)
elif action == "luksOpen":
if len(cs_args) < 2: raise "WrongArguments", "missing arguments"
cs_args.insert(-1, action)
cs_args.insert(-1, device)
elif action == "luksClose":
if len(cs_args) < 2: raise "WrongArguments", "missing arguments"
cs_args.insert(-1, action)
elif action == "luksAddKey":
cs_args.append(action)
cs_args.append(device)
elif action == "luksDelKey":
if len(cs_args) < 2: raise "WrongArguments", "missing arguments"
cs_args.insert(-1, action)
cs_args.insert(-1, device)
elif action == "isLuks":
cs_args.append(action)
cs_args.append(device)
else: raise "WrongArguments", "invalid action supplied"
except TypeError:
raise "WrongArguments", "invalid arguments supplied"
# execute cryptsetup with the given parameters
proc = subprocess.Popen(
shell = False,
args = cs_args)
proc.communicate()
return proc.returncode == 0
def run_sfdisk(device, args): def run_sfdisk(device, args):
@ -103,7 +144,7 @@ if __name__ == "__main__":
if len(sys.argv) < 2: if len(sys.argv) < 2:
sys.stderr.write("Not enough arguments supplied (%s)!\n" % " ".join(sys.argv)) sys.stderr.write("Not enough arguments supplied (%s)!\n" % " ".join(sys.argv))
sys.exit(1) sys.exit(100)
progRequest = sys.argv[0] progRequest = sys.argv[0]
deviceRequest = sys.argv[1] deviceRequest = sys.argv[1]
@ -112,21 +153,27 @@ if __name__ == "__main__":
if not progRequest in allowedProgs.keys(): if not progRequest in allowedProgs.keys():
sys.stderr.write("Invalid program requested: %s\n" % progRequest) sys.stderr.write("Invalid program requested: %s\n" % progRequest)
sys.exit(2) sys.exit(100)
if not os.path.exists(deviceRequest): if not os.path.exists(deviceRequest):
sys.stderr.write("The specified device (%s) does not exist!\n" % deviceRequest) sys.stderr.write("The specified device (%s) does not exist!\n" % deviceRequest)
sys.exit(3) sys.exit(100)
if not isWriteableBlock(deviceRequest): if not isWriteableBlock(deviceRequest):
sys.stderr.write("This device (%s) must be a writeable block device!\n" % deviceRequest) sys.stderr.write("This device (%s) must be a writeable block device!\n" % deviceRequest)
sys.exit(4) sys.exit(100)
if progRequest == "cryptsetup": runner = run_cryptsetup if progRequest == "cryptsetup": runner = run_cryptsetup
elif progRequest == "sfdisk": runner = run_sfdisk elif progRequest == "sfdisk": runner = run_sfdisk
else: else:
sys.stderr.write("The interface for this program (%s) is not yet implemented!\n" % progRequest) sys.stderr.write("The interface for this program (%s) is not yet implemented!\n" % progRequest)
sys.exit(1) sys.exit(100)
try:
runner(deviceRequest, sys.argv) if runner(deviceRequest, sys.argv):
sys.exit(0)
else:
sys.exit(1)
except "WrongArguments", errstr:
sys.stderr.write("Execution failed: %s\n" % errstr)
sys.exit(100)

View file

@ -39,16 +39,14 @@ Level = debug
# where to write the log messages to? # where to write the log messages to?
# possible values are: file # possible values are: file
# syslog support will be added later # syslog support will be added later
Facility = file Destination = file
# depending on the choosen facility (see above) you may select a # depending on the choosen destination (see above) you may select
# destination. Possible values for the different facilities are: # details. Possible values for the different destinations are:
# file: $FILENAME # file: $FILENAME
# syslog: $LOG_FACILITY # syslog: $LOG_FACILITY
# The log file will get created as root and then handed over to the #Details = /var/log/cryptobox.log
# cryptobox user (see above) Details = ./cryptobox.log
#Destination = /var/log/cryptobox.log
Destination = ./cryptobox.log
[Settings] [Settings]
#default stylesheet #default stylesheet