Commit 4c66de50 authored by Ziirish's avatar Ziirish

add the ability to rename/move options within the configuration

parent 2db8e801
Pipeline #1625 passed with stages
in 10 minutes and 42 seconds
......@@ -21,48 +21,42 @@ class BUIConfig(dict):
logger = logger
mtime = 0
def __init__(self, config=None, explain=False, defaults=None):
def __init__(self, config=None, defaults=None):
"""Wrapper around the ConfigObj class
:param config: Configuration to parse
:type config: str, list or File
:param explain: Whether to explain the parsing errors or not
:type explain: bool
:param defaults: Default options
:type defaults: dict
"""
if defaults is not None:
self.defaults = defaults
if config:
self.parse(config, explain, defaults)
self.parse(config, defaults)
def parse(self, config, explain=False, defaults=None):
def parse(self, config, defaults=None):
"""Parse the conf
:param config: Configuration to parse
:type config: str, list or File
:param explain: Whether to explain the parsing errors or not
:type explain: bool
:param defaults: Default options
:type defaults: dict
"""
self.conf = {}
self.conffile = config
self.section = None
self.defaults = defaults
if defaults is not None or not hasattr(self, 'defaults'):
self.defaults = defaults
self.validator = validate.Validator()
try:
self.conf = configobj.ConfigObj(config, encoding='utf-8')
self.mtime = os.path.getmtime(self.conffile)
except configobj.ConfigObjError as exp:
# We were unable to parse the config
self.logger.critical('Unable to convert configuration')
if explain:
self._explain(exp)
else:
raise exp
self.logger.critical('Unable to parse configuration')
raise exp
@property
def options(self):
......@@ -148,13 +142,63 @@ class BUIConfig(dict):
if ori:
with codecs.open(conffile, 'w', 'utf-8', errors='ignore') as config:
for line in ori:
if re.match(r'^\s*(#|;)+\s*\[{}\]'.format(old_section), line):
if re.match(r'^\s*(#|;)*\s*\[{}\]'.format(old_section), line):
config.write('{}\n'.format(line.replace(old_section, new_section)))
ret = True
else:
config.write('{}\n'.format(line))
return ret
def _rename_option_full(self, orig_option, dest_option, orig_section, dest_section):
"""Rename a given option and possibly moves it to another section
:return: True if the option have been successfully renamed/moved
:rtype: bool
:raises ValueError: if the ``orig_section`` does not exist
:raises KeyError: if the ``orig_option`` does not exist in the ``orig_section``
"""
if not self.section_exists(orig_section):
raise ValueError("No such section: {}".format(orig_section))
orig = self.conf[orig_section]
if orig_option not in orig:
raise KeyError("No such option in the [{}] section: {}".format(orig_section, orig_option))
# adding the new section if it is missing
if orig_section != dest_section and not self.lookup_section(dest_section):
self._refresh(True)
dest = self.conf[dest_section]
comments = orig.comments[orig_option]
inline_comments = orig.inline_comments[orig_option]
# copy value and comments from orig to dest
dest[dest_option] = orig[orig_option]
dest.comments[dest_option] = comments
dest.inline_comments[dest_option] = inline_comments
# remove orig key
del orig[orig_option]
# save
self.conf.write()
return True
def rename_option(self, orig_option, dest_option, section):
"""Rename a given option"""
# this is useless
if orig_option == dest_option:
return False
return self._rename_option_full(orig_option, dest_option, section, section)
def move_option(self, option, orig_section, dest_section):
"""Move an option to another section, if you need to rename the option use the
_rename_option_full function instead"""
# useless
if orig_section == dest_section:
return False
return self._rename_option_full(option, option, orig_section, dest_section)
def changed(self, id):
"""Check if the conf has changed"""
# don't use delta for cases where we run several gunicorn workers
......@@ -177,19 +221,6 @@ class BUIConfig(dict):
"""Set the default section"""
self.section = section
@staticmethod
def _explain(exception):
"""Explain parsing errors
:param exception: Exception object
:type exception: :class:`configobj.ConfigObjError`
"""
message = '\n'
for error in exception.errors:
message += error.message + '\n'
raise configobj.ConfigObjError(message.rstrip('\n'))
def safe_get(
self,
key,
......
......@@ -121,7 +121,7 @@ class BUIAgent(BUIbackend):
# Raise exception if errors are encountered during parsing
self.conf = config
self.conf.parse(conf, True, BUI_DEFAULTS)
self.conf.parse(conf, BUI_DEFAULTS)
self.conf.default_section('Global')
self.port = self.conf.safe_get('port', 'integer')
self.bind = self.conf.safe_get('bind')
......
......@@ -102,7 +102,7 @@ class MonitorPool:
# Raise exception if errors are encountered during parsing
self.conf = config
self.conf.parse(conf, True, BUI_DEFAULTS)
self.conf.parse(conf, BUI_DEFAULTS)
self.conf.default_section('Global')
self.port = self.conf.safe_get('port', 'integer')
self.bind = self.conf.safe_get('bind')
......
......@@ -47,7 +47,7 @@ BUI_DEFAULTS = {
'refresh': 180,
'liverefresh': 5,
'ignore_labels': ["color:.*"],
'format_labels': ["s/^os:\s*//"],
'format_labels': [r"s/^os:\s*//"],
'default_strip': 0,
},
'Security': {
......@@ -137,7 +137,7 @@ class BUIServer(Flask):
raise IOError('No configuration file found')
# Raise exception if errors are encountered during parsing
self.conf.parse(conf, True, BUI_DEFAULTS)
self.conf.parse(conf, BUI_DEFAULTS)
self.conf.default_section('Global')
self.config['BUI_BIND'] = self.conf.safe_get('bind')
......
......@@ -74,7 +74,7 @@ class SessionManager(object):
"""anonymize ip address while running the demo"""
# Do nothing if not in demo mode
if self.app.config['BUI_DEMO']:
if re.match('^\d+\.\d+\.\d+\.\d+$', ip):
if re.match(r'^\d+\.\d+\.\d+\.\d+$', ip):
spl = ip.split('.')
ip = '{}.x.x.x'.format(spl[0])
else:
......
......@@ -17,3 +17,5 @@ omit =
*/burpui/engines/agent.py
*/burpui/engines/worker.py
*/burpui/engines/monitor.py
*/burpui/misc/auth/ldap.py
*/burpui/misc/auth/local.py
......@@ -2,17 +2,6 @@
# @version@ - 0.3.0
# @release@ - stable
[Global]
# On which port is the application listening
port = 5001
# On which address is the application listening
# '::' is the default for all IPv6
bind = ::
# enable SSL
ssl = false
# ssl cert
sslcert = /etc/burp/ssl_cert-server.pem
# ssl key
sslkey = /etc/burp/ssl_cert-server.key
backend = burp1
# authentication plugin (mandatory)
# list the misc/auth directory to see the available backends
......
......@@ -2,17 +2,6 @@
# @version@ - 0.3.0
# @release@ - stable
[Global]
# On which port is the application listening
port = 5001
# On which address is the application listening
# '::' is the default for all IPv6
bind = ::
# enable SSL
ssl = false
# ssl cert
sslcert = /etc/burp/ssl_cert-server.pem
# ssl key
sslkey = /etc/burp/ssl_cert-server.key
backend = burp1
# authentication plugin (mandatory)
# list the misc/auth directory to see the available backends
......
......@@ -2,17 +2,6 @@
# @version@ - 0.3.0
# @release@ - stable
[Global]
# On which port is the application listening
port = 5001
# On which address is the application listening
# '::' is the default for all IPv6
bind = ::
# enable SSL
ssl = false
# ssl cert
sslcert = /etc/burp/ssl_cert-server.pem
# ssl key
sslkey = /etc/burp/ssl_cert-server.key
backend = burp1
# authentication plugin (mandatory)
# list the misc/auth directory to see the available backends
......
......@@ -2,17 +2,6 @@
# @version@ - 0.3.0
# @release@ - stable
[Global]
# On which port is the application listening
port = 5001
# On which address is the application listening
# '::' is the default for all IPv6
bind = ::
# enable SSL
ssl = false
# ssl cert
sslcert = /etc/burp/ssl_cert-server.pem
# ssl key
sslkey = /etc/burp/ssl_cert-server.key
backend = burp1
# authentication plugin (mandatory)
# list the misc/auth directory to see the available backends
......
......@@ -2,17 +2,6 @@
# @version@ - 0.3.0
# @release@ - stable
[Global]
# On which port is the application listening
port = 5001
# On which address is the application listening
# '::' is the default for all IPv6
bind = ::
# enable SSL
ssl = false
# ssl cert
sslcert = /etc/burp/ssl_cert-server.pem
# ssl key
sslkey = /etc/burp/ssl_cert-server.key
backend = burp1
# authentication plugin (mandatory)
# list the misc/auth directory to see the available backends
......
......@@ -2,17 +2,6 @@
# @version@ - 0.3.0
# @release@ - stable
[Global]
# On which port is the application listening
port = 5001
# On which address is the application listening
# '::' is the default for all IPv6
bind = ::
# enable SSL
ssl = false
# ssl cert
sslcert = /etc/burp/ssl_cert-server.pem
# ssl key
sslkey = /etc/burp/ssl_cert-server.key
backend = burp1
# authentication plugin (mandatory)
# list the misc/auth directory to see the available backends
......
......@@ -2,17 +2,6 @@
# @version@ - 0.3.0
# @release@ - stable
[Global]
# On which port is the application listening
port = 5001
# On which address is the application listening
# '::' is the default for all IPv6
bind = ::
# enable SSL
ssl = false
# ssl cert
sslcert = /etc/burp/ssl_cert-server.pem
# ssl key
sslkey = /etc/burp/ssl_cert-server.key
backend = burp1
# authentication plugin (mandatory)
# list the misc/auth directory to see the available backends
......
......@@ -2,17 +2,6 @@
# @version@ - 0.3.0
# @release@ - stable
[Global]
# On which port is the application listening
port = 5001
# On which address is the application listening
# '::' is the default for all IPv6
bind = ::
# enable SSL
ssl = false
# ssl cert
sslcert = /etc/burp/ssl_cert-server.pem
# ssl key
sslkey = /etc/burp/ssl_cert-server.key
backend = burp1
# authentication plugin (mandatory)
# list the misc/auth directory to see the available backends
......
......@@ -2,17 +2,6 @@
# @version@ - 0.3.0
# @release@ - stable
[Global]
# On which port is the application listening
port = 5001
# On which address is the application listening
# '::' is the default for all IPv6
bind = ::
# enable SSL
ssl = false
# ssl cert
sslcert = /etc/burp/ssl_cert-server.pem
# ssl key
sslkey = /etc/burp/ssl_cert-server.key
backend = burp1
# authentication plugin (mandatory)
# list the misc/auth directory to see the available backends
......
......@@ -2,17 +2,6 @@
# @version@ - 0.3.0
# @release@ - stable
[Global]
# On which port is the application listening
port = 5001
# On which address is the application listening
# '::' is the default for all IPv6
bind = ::
# enable SSL
ssl = false
# ssl cert
sslcert = /etc/burp/ssl_cert-server.pem
# ssl key
sslkey = /etc/burp/ssl_cert-server.key
backend = burp1
# authentication plugin (mandatory)
# list the misc/auth directory to see the available backends
......
import os
import pytest
import configobj
import validate
from tempfile import mkstemp
from burpui.config import BUIConfig
TEST_CONFIG = b"""
[Global]
# backend comment
backend = something
timeout = 12
duplicate = nyan
#[Test]
[Production]
duplicate = cat
run = true
sql = none
array = some, VALUES
"""
TEST_CONFIG_FAILURE = b"""
[I is a wrong file
hi ha ho
"""
def test_config_init():
casters = ['string_lower_list', 'force_string', 'boolean_or_string']
fd, tmpfile = mkstemp()
os.write(fd, TEST_CONFIG)
os.close(fd)
fd, wrong = mkstemp()
os.write(fd, TEST_CONFIG_FAILURE)
os.close(fd)
config = BUIConfig(tmpfile)
with pytest.raises(configobj.ConfigObjError):
fail = BUIConfig(wrong, defaults={})
assert config.safe_get('backend', section='Global') == 'something'
assert config.safe_get('timeout', 'integer', 'Global') == 12
config.default_section('Production')
assert config.safe_get('duplicate') == 'cat'
assert config.safe_get('duplicate', section='Global') == 'nyan'
assert config.safe_get('run', 'boolean_or_string') is True
assert config.safe_get('sql', 'boolean_or_string') == 'none'
array = config.safe_get('array', 'string_lower_list')
assert array[1] == 'values'
assert array[0] == 'some'
assert isinstance(config.safe_get('array'), list)
assert config.safe_get('array', 'force_string') == 'some,VALUES'
for cast in casters:
# safe_get is safe and shouldn't raise any exception
assert config.safe_get('i iz not in ze config!', cast) is None
os.unlink(tmpfile)
os.unlink(wrong)
def test_config_reload():
fd, tmpfile = mkstemp()
os.write(fd, TEST_CONFIG)
os.close(fd)
config = BUIConfig(tmpfile)
assert 'last' not in config.options.get('Production', {})
with open(tmpfile, 'a') as cfg:
print("last = ohai", file=cfg)
config.mtime = -1
assert 'last' in config.options.get('Production', {})
assert config.options.get('Production', {}).get('last') == 'ohai'
os.unlink(tmpfile)
def test_config_sections():
fd, tmpfile = mkstemp()
os.write(fd, TEST_CONFIG)
os.close(fd)
config = BUIConfig(tmpfile)
with open(tmpfile) as cfg:
lines = [x.rstrip() for x in cfg.readlines()]
assert '[Unknown]' not in lines
assert '[Test]' not in lines
assert not config.lookup_section('Unknown')
with open(tmpfile) as cfg:
lines = [x.rstrip() for x in cfg.readlines()]
assert '[Unknown]' in lines
assert lines[-1] == '[Unknown]'
assert not config.lookup_section('Test')
with open(tmpfile) as cfg:
lines = [x.rstrip() for x in cfg.readlines()]
assert '[Test]' in lines
assert lines[-1] != '[Test]'
assert config.lookup_section('Production')
os.unlink(tmpfile)
def test_config_rename_section():
fd, tmpfile = mkstemp()
os.write(fd, TEST_CONFIG)
os.close(fd)
config = BUIConfig(tmpfile)
with open(tmpfile) as cfg:
lines = [x.rstrip() for x in cfg.readlines()]
assert '[Production2]' not in lines
assert not config.rename_section('Unknown', 'Test')
assert config.rename_section('Production', 'Production2')
with open(tmpfile) as cfg:
lines = [x.rstrip() for x in cfg.readlines()]
assert '[Production2]' in lines
os.unlink(tmpfile)
def test_config_rename_option():
fd, tmpfile = mkstemp()
os.write(fd, TEST_CONFIG)
os.close(fd)
config = BUIConfig(tmpfile)
config.default_section('Global')
with pytest.raises(KeyError):
config.rename_option('unknown', 'yeah', 'Global')
with pytest.raises(ValueError):
config.rename_option('test', 'truc', 'Unknown')
assert 'back' not in config.options.get('Global', {})
assert not config.rename_option('backend', 'backend', 'Global')
assert config.rename_option('backend', 'back', 'Global')
assert config.safe_get('back') == 'something'
os.unlink(tmpfile)
def test_config_move_option():
fd, tmpfile = mkstemp()
os.write(fd, TEST_CONFIG)
os.close(fd)
config = BUIConfig(tmpfile)
assert 'New' not in config.options
assert 'backend' not in config.options.get('New', {})
assert not config.move_option('backend', 'Global', 'Global')
assert config.move_option('backend', 'Global', 'New')
assert config.safe_get('backend', section='New') == 'something'
os.unlink(tmpfile)
def test_config_safe_get():
fd, tmpfile = mkstemp()
os.write(fd, TEST_CONFIG)
os.close(fd)
config = BUIConfig(tmpfile)
assert config.safe_get('timeout', 'idontknow', 'Global') == '12'
assert config.safe_get('test', section='hahaha') is None
os.unlink(tmpfile)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment