You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

247 lines
7.8 KiB

# Copyright 2006 sense.lab e.V.
# This file is part of the CryptoBox.
# The CryptoBox is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
# The CryptoBox is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with the CryptoBox; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
This is the web interface for a fileserver managing encrypted filesystems.
__revision__ = "$Id"
import sys
import cryptobox.core.container as cbxContainer
from cryptobox.core.exceptions import CBEnvironmentError, CBConfigUndefinedError
import re
import os
import as cbxTools
import subprocess
class CryptoBox:
'''this class rules them all!
put things like logging, conf and other stuff in here,
that might be used by more classes, it will be passed on to them'''
def __init__(self, config_file=None):
import cryptobox.core.settings as cbxSettings
self.log = self.__get_startup_logger()
self.prefs = cbxSettings.CryptoBoxSettings(config_file)
self.__containers = []
self.busy_devices = {}
def __get_startup_logger(self):
"""Initialize the configured logging facility of the CryptoBox.
use it with: 'self.log.[debug|info|warning|error|critical](logmessage)'
all classes should get the logging instance during __init__:
self.log = logging.getLogger("CryptoBox")
first we output all warnings/errors to stderr
as soon as we opened the config file successfully, we redirect debug output
to the configured destination
import logging
## basicConfig(...) needs python >= 2.4
log_handler = logging.getLogger("CryptoBox")
format = '%(asctime)s CryptoBox %(levelname)s: %(message)s',
stderr = sys.stderr)
log_handler.setLevel(logging.ERROR)"loggingsystem is up'n running")
## from now on everything can be logged via self.log...
raise CBEnvironmentError("couldn't initialise the loggingsystem. I give up.")
return log_handler
def __run_tests(self):
"""Do some initial tests.
def __run_test_root_priv(self):
"""Try to run 'super' with 'CryptoBoxRootActions'.
devnull = open(os.devnull, "w")
except IOError:
raise CBEnvironmentError("could not open %s for writing!" % os.devnull)
prog_super = self.prefs["Programs"]["super"]
except KeyError:
raise CBConfigUndefinedError("Programs", "super")
prog_rootactions = self.prefs["Programs"]["CryptoBoxRootActions"]
except KeyError:
raise CBConfigUndefinedError("Programs", "CryptoBoxRootActions")
proc = subprocess.Popen(
shell = False,
stdout = devnull,
stderr = devnull,
args = [prog_super, prog_rootactions, "check"])
except OSError:
raise CBEnvironmentError(
"failed to execute 'super' (%s)" % self.prefs["Programs"]["super"])
if proc.returncode != 0:
raise CBEnvironmentError("failed to call CryptoBoxRootActions ("
+ prog_rootactions + ") via 'super' - maybe you did not add the "
+ "appropriate line to '/etc/'?")
def reread_container_list(self):
"""Reinitialize the list of available containers.
This should be called whenever the available containers may have changed.
E.g.: after partitioning and after device addition/removal
self.log.debug("rereading container list")
self.__containers = []
for device in cbxTools.get_available_partitions():
if self.is_device_allowed(device) and not self.is_config_partition(device):
self.__containers.append(cbxContainer.CryptoBoxContainer(device, self))
## sort by container name
self.__containers.sort(cmp = lambda x, y: x.get_name() < y.get_name() and -1 or 1)
def is_config_partition(self, device):
"""Check if a given partition contains configuration informations.
The check is done by comparing the label of the filesystem with a string.
proc = subprocess.Popen(
shell = False,
stdout = subprocess.PIPE,
stderr = subprocess.PIPE,
args = [
"-c", os.path.devnull,
"-o", "value",
"-s", "LABEL",
(output, error) = proc.communicate()
return output.strip() == self.prefs["Main"]["ConfigVolumeLabel"]
def is_device_allowed(self, devicename):
"check if a device is white-listed for being used as cryptobox containers"
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 ...
real_device = os.path.realpath(devicename)
if a_dev and'^' + a_dev, real_device):
return True
return False
def get_log_data(self, lines=None, max_size=None):
"""get the most recent log entries of the cryptobox
the maximum number and size of these entries can be limited by
'lines' and 'max_size'
# return nothing if the currently selected log output is not a file
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 []
fdesc = open(log_file, "r")
if max_size:, 2) # seek relative to the end of the file
content = fdesc.readlines()
except IOError:
self.log.warn("failed to read the log file (%s)" % log_file)
return []
if lines:
content = content[-lines:]
return content
def get_container_list(self, filter_type=None, filter_name=None):
"retrieve the list of all containers of this cryptobox"
result = self.__containers[:]
if filter_type != None:
if filter_type in range(len(cbxContainer.CONTAINERTYPES)):
return [e for e in self.__containers if e.get_type() == filter_type]
else:"invalid filter_type (%d)" % filter_type)
if filter_name != None:
result = [e for e in self.__containers if e.get_name() == filter_name]
return result
except AttributeError:
return []
def get_container(self, device):
"retrieve the container element for this device"
all = [e for e in self.get_container_list() if e.device == device]
if all:
return all[0]
return None
def send_event_notification(self, event, event_infos):
"""call all available scripts in the event directory with some event information"""
event_dir = self.prefs["Locations"]["EventDir"]
for fname in os.listdir(event_dir):
real_fname = os.path.join(event_dir, fname)
if os.path.isfile(real_fname) and os.access(real_fname, os.X_OK):
cmd_args = [ self.prefs["Programs"]["super"],
"event", real_fname, event]
proc = subprocess.Popen(
shell = False,
stdout = subprocess.PIPE,
stderr = subprocess.PIPE,
args = cmd_args)
(stdout, stderr) = proc.communicate()
if proc.returncode != 0:
"an event script (%s) failed (exitcode=%d) to handle an event (%s): %s" %
(real_fname, proc.returncode, event, stderr.strip()))
else:"event handler (%s) finished successfully: %s" %
(real_fname, event))
if __name__ == "__main__":