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.

951 lines
32 KiB

9 months ago
#!/usr/bin/env python
#
# Simplified parsing of bind configuration, with include support and nested sections.
import re
import string
class ConfigParseError(Exception):
"""Generic error when parsing config file."""
def __init__(self, error=None, parent=None):
# IOError on python3 includes path, on python2 it does not
message = "Cannot open the configuration file \"{path}\": {error}".format(
path=error.filename, error=str(error))
if parent:
message += "; included from \"{0}\"".format(parent)
super(ConfigParseError, self).__init__(message)
self.error = error
self.parent = parent
pass
class ConfigFile(object):
"""Representation of single configuration file and its contents."""
def __init__(self, path):
"""Load config file contents from path.
:param path: Path to file
"""
self.path = path
self.load(path)
self.status = None
def __str__(self):
return self.buffer
def __repr__(self):
return 'ConfigFile {0} ({1})'.format(
self.path, self.buffer)
def load(self, path):
with open(path, 'r') as f:
self.buffer = self.original = f.read()
def is_modified(self):
return self.original == self.buffer
def root_section(self):
return ConfigSection(self, None, 0, len(self.buffer))
class MockConfig(ConfigFile):
"""Configuration file with contents defined on constructor.
Intended for testing the library.
"""
DEFAULT_PATH = '/etc/named/mock.conf'
def __init__(self, contents, path=DEFAULT_PATH):
self.original = contents
super(MockConfig, self).__init__(path)
def load(self, path):
self.buffer = self.original
class ConfigSection(object):
"""Representation of section or key inside single configuration file.
Section means statement, block, quoted string or any similar."""
TYPE_BARE = 1
TYPE_QSTRING = 2
TYPE_BLOCK = 3
TYPE_IGNORED = 4 # comments and whitespaces
def __init__(self, config, name=None, start=None, end=None, kind=None, parser=None):
"""
:param config: config file inside which is this section
:type config: ConfigFile
:param kind: type of this section
"""
self.config = config
self.name = name
self.start = start
self.end = end
self.ctext = self.original_value() # a copy for modification
self.parser = parser
if kind is None:
if self.config.buffer.startswith('{', self.start):
self.kind = self.TYPE_BLOCK
elif self.config.buffer.startswith('"', self.start):
self.kind = self.TYPE_QSTRING
else:
self.kind = self.TYPE_BARE
else:
self.kind = kind
self.statements = []
def __repr__(self):
text = self.value()
path = self.config.path
return 'ConfigSection#{kind}({path}:{start}-{end}: "{text}")'.format(
path=path, start=self.start, end=self.end,
text=text, kind=self.kind
)
def __str__(self):
return self.value()
def copy(self):
return ConfigSection(self.config, self.name, self.start, self.end, self.kind)
def type(self):
return self.kind
def value(self):
return self.ctext
def original_value(self):
return self.config.buffer[self.start:self.end+1]
def invalue(self):
"""Return just inside value of blocks and quoted strings."""
t = self.type()
if t in (self.TYPE_QSTRING, self.TYPE_BLOCK):
return self.ctext[1:-1]
return self.value()
def children(self, comments=False):
"""Return list of items inside this section."""
start = self.start
if self.type() == self.TYPE_BLOCK:
start += 1
return list(IscIterator(self.parser, self, comments, start))
def serialize(self):
return self.value()
class IscIterator(object):
"""Iterator for walking over parsed configuration.
Creates sequence of ConfigSection objects for a given file.
That means a stream of objects.
"""
def __init__(self, parser, section, comments=False, start=None):
"""Create iterator.
:param comments: Include comments and whitespaces
:param start: Index for starting, None means beginning of section
"""
self.parser = parser
self.section = section
self.current = None
self.key_wanted = True
self.comments = comments
self.waiting = None
if start is None:
start = section.start
self.start = start
def __iter__(self):
self.current = None
self.key_wanted = True
self.waiting = None
return self
def __next__(self):
index = self.start
cfg = self.section.config
if self.waiting:
self.current = self.waiting
self.waiting = None
return self.current
if self.current is not None:
index = self.current.end+1
if self.key_wanted:
val = self.parser.find_next_key(cfg, index, self.section.end)
self.key_wanted = False
else:
val = self.parser.find_next_val(cfg, None, index, self.section.end, end_report=True)
if val is not None and val.value() in self.parser.CHAR_DELIM:
self.key_wanted = True
if val is None:
if self.current is not None and self.current.end < self.section.end and self.comments:
self.current = ConfigSection(self.section.config, None,
index, self.section.end, ConfigSection.TYPE_IGNORED)
return self.current
raise StopIteration
if index != val.start and self.comments:
# Include comments and spaces as ignored section
self.waiting = val
val = ConfigSection(val.config, None, index, val.start-1, ConfigSection.TYPE_IGNORED)
self.current = val
return val
next = __next__ # Python2 compat
class IscVarIterator(object):
"""Iterator for walking over parsed configuration.
Creates sequence of ConfigVariableSection objects for a given
file or section.
"""
def __init__(self, parser, section, comments=False, start=None):
"""Create iterator."""
self.parser = parser
self.section = section
self.iter = IscIterator(parser, section, comments, start)
def __iter__(self):
return self
def __next__(self):
vl = []
try:
statement = next(self.iter)
while statement:
vl.append(statement)
if self.parser.is_terminal(statement):
return ConfigVariableSection(vl, None, parent=self.section)
statement = next(self.iter)
except StopIteration:
if vl:
return ConfigVariableSection(vl, None, parent=self.section)
raise StopIteration
next = __next__ # Python2 compat
class ConfigVariableSection(ConfigSection):
"""Representation for key and values of variable length.
Intended for view and zone.
"""
def __init__(self, sectionlist, name, zone_class=None, parent=None, parser=None):
"""Creates variable block for zone or view.
:param sectionlist: list of ConfigSection, obtained from IscConfigParser.find_values()
"""
last = next(reversed(sectionlist))
first = sectionlist[0]
self.values = sectionlist
super(ConfigVariableSection, self).__init__(
first.config, name, start=first.start, end=last.end, parser=parser
)
if name is None:
try:
self.name = self.var(1).invalue()
except IndexError:
pass
# For optional dns class, like IN or CH
self.zone_class = zone_class
self.parent = parent
def key(self):
if self.zone_class is None:
return self.name
return self.zone_class + '_' + self.name
def firstblock(self):
"""Return first block section in this tool."""
return self.vartype(0, self.TYPE_BLOCK)
def var(self, i):
"""Return value by index, ignore spaces."""
n = 0
for v in self.values:
if v.type() != ConfigSection.TYPE_IGNORED:
if n == i:
return v
n += 1
raise IndexError
def vartype(self, i, vtype):
n = 0
for v in self.values:
if v.type() == vtype:
if n == i:
return v
n += 1
raise IndexError
def serialize(self):
s = ''
for v in self.values:
s += v.serialize()
return s
def serialize_skip(self, replace_ignored=None):
"""
Create single string from section, but skip whitespace on start.
:type section: ConfigVariableSection
:param replace_ignored: Specify replaced text for whitespace
Allows normalizing with replace ignored sections.
Is intended to strip possible comments between parts.
"""
s = ''
nonwhite = None
for v in self.values:
if nonwhite is None:
if v.type() != self.TYPE_IGNORED:
nonwhite = v
s += v.serialize()
elif replace_ignored is not None and v.type() == self.TYPE_IGNORED:
s += replace_ignored
else:
s += v.serialize()
return s
class ModifyState(object):
"""Object keeping state of modifications when walking configuration file statements.
It would keep modified configuration file and position of last found statement.
"""
def __init__(self):
self.value = ''
self.lastpos = 0
def append_before(self, section):
"""Appends content from last seen section to beginning of current one.
It adds also whitespace on beginning of statement,
which is usually not interesting for any changes.
:type section: ConfigVariableSection
"""
end = section.start
first = section.values[0]
if first.type() == first.TYPE_IGNORED:
end = first.end
cfg = section.config.buffer
self.value += cfg[self.lastpos:end+1]
self.lastpos = end+1
def move_after(self, section):
"""Set position to the end of section."""
self.lastpos = section.end+1
def finish(self, section):
"""Append remaining part of file to modified state."""
if self.lastpos < section.end:
self.value += section.config.buffer[self.lastpos:section.end+1]
self.lastpos = section.end
def content(self):
"""Get content of (modified) section.
Would be valid after finish() was called.
"""
return self.value
@staticmethod
def callback_comment_out(section, state):
"""parser.walk callback for commenting out the section."""
state.append_before(section)
state.value += '/* ' + section.serialize_skip(' ') + ' */'
state.move_after(section)
@staticmethod
def callback_remove(section, state):
"""parser.walk callback for skipping a section."""
state.append_before(section)
state.move_after(section)
# Main parser class
class IscConfigParser(object):
"""Parser file with support of included files.
Reads ISC BIND configuration file and tries to skip commented blocks, nested sections and similar stuff.
Imitates what isccfg does in native code, but without any use of native code.
"""
CONFIG_FILE = "/etc/named.conf"
FILES_TO_CHECK = []
CHAR_DELIM = ";" # Must be single character
CHAR_CLOSING = CHAR_DELIM + "})]"
CHAR_CLOSING_WHITESPACE = CHAR_CLOSING + string.whitespace
CHAR_KEYWORD = string.ascii_letters + string.digits + '-_.:'
CHAR_STR_OPEN = '"'
def __init__(self, config=None):
"""Construct parser.
:param config: path to file or already loaded ConfigFile instance
Initialize contents from path to real config or already loaded ConfigFile class.
"""
if isinstance(config, ConfigFile):
self.FILES_TO_CHECK = [config]
self.load_included_files()
elif config is not None:
self.load_config(config)
#
# function for parsing of config files
#
def is_comment_start(self, istr, index=0):
if istr[index] == "#" or (
index+1 < len(istr) and istr[index:index+2] in ["//", "/*"]):
return True
return False
def _find_end_of_comment(self, istr, index=0):
"""Returns index where the comment ends.
:param istr: input string
:param index: begin search from the index; from the start by default
Support usual comments till the end of line (//, #) and block comment
like (/* comment */). In case that index is outside of the string or end
of the comment is not found, return -1.
In case of block comment, returned index is position of slash after star.
"""
length = len(istr)
if index >= length or index < 0:
return -1
if istr[index] == "#" or istr[index:].startswith("//"):
return istr.find("\n", index)
if index+2 < length and istr[index:index+2] == "/*":
res = istr.find("*/", index+2)
if res != -1:
return res + 1
return -1
def is_opening_char(self, c):
return c in "\"'{(["
def _remove_comments(self, istr, space_replace=False):
"""Removes all comments from the given string.
:param istr: input string
:param space_replace When true, replace comments with spaces. Skip them by default.
:return: istr without comments
"""
ostr = ""
length = len(istr)
index = 0
while index < length:
if self.is_comment_start(istr, index):
index = self._find_end_of_comment(istr, index)
if index == -1:
index = length
if space_replace:
ostr = ostr.ljust(index)
if index < length and istr[index] == "\n":
ostr += "\n"
elif istr[index] in self.CHAR_STR_OPEN:
end_str = self._find_closing_char(istr, index)
if end_str == -1:
ostr += istr[index:]
break
ostr += istr[index:end_str+1]
index = end_str
else:
ostr += istr[index]
index += 1
return ostr
def _replace_comments(self, istr):
"""Replaces all comments by spaces in the given string.
:param istr: input string
:returns: string of the same length with comments replaced
"""
return self._remove_comments(istr, True)
def find_next_token(self, istr, index=0, end_index=-1, end_report=False):
"""
Return index of another interesting token or -1 when there is not next.
:param istr: input string
:param index: begin search from the index; from the start by default
:param end_index: stop searching at the end_index or end of the string
In case that initial index contains already some token, skip to another.
But when searching starts on whitespace or beginning of the comment,
choose the first one.
The function would be confusing in case of brackets, but content between
brackets is not evaluated as new tokens.
E.g.:
"find { me };" : 5
" me" : 1
"find /* me */ me " : 13
"/* me */ me" : 9
"me;" : 2
"{ me }; me" : 6
"{ me } me" : 8
"me } me" : 3
"}} me" : 1
"me" : -1
"{ me } " : -1
"""
length = len(istr)
if length < end_index or end_index < 0:
end_index = length
if index >= end_index or index < 0:
return -1
# skip to the end of the current token
if istr[index] == '\\':
index += 2
elif self.is_opening_char(istr[index]):
index = self._find_closing_char(istr, index, end_index)
if index != -1:
index += 1
elif self.is_comment_start(istr, index):
index = self._find_end_of_comment(istr, index)
if index != -1:
index += 1
elif istr[index] not in self.CHAR_CLOSING_WHITESPACE:
# so we have to skip to the end of the current token
index += 1
while index < end_index:
if (istr[index] in self.CHAR_CLOSING_WHITESPACE
or self.is_comment_start(istr, index)
or self.is_opening_char(istr[index])):
break
index += 1
elif end_report and istr[index] in self.CHAR_DELIM:
# Found end of statement. Report delimiter
return index
elif istr[index] in self.CHAR_CLOSING:
index += 1
# find next token (can be already under the current index)
while 0 <= index < end_index:
if istr[index] == '\\':
index += 2
continue
if self.is_comment_start(istr, index):
index = self._find_end_of_comment(istr, index)
if index == -1:
break
elif self.is_opening_char(istr[index]) or istr[index] not in string.whitespace:
return index
index += 1
return -1
def _find_closing_char(self, istr, index=0, end_index=-1):
"""
Returns index of equivalent closing character.
:param istr: input string
It's similar to the "find" method that returns index of the first character
of the searched character or -1. But in this function the corresponding
closing character is looked up, ignoring characters inside strings
and comments. E.g. for
"(hello (world) /* ) */ ), he would say"
index of the third ")" is returned.
"""
important_chars = { # TODO: should be that rather global var?
"{": "}",
"(": ")",
"[": "]",
"\"": "\"",
self.CHAR_DELIM: None,
}
length = len(istr)
if 0 <= end_index < length:
length = end_index
if length < 2:
return -1
if index >= length or index < 0:
return -1
closing_char = important_chars.get(istr[index], self.CHAR_DELIM)
if closing_char is None:
return -1
isString = istr[index] in "\""
index += 1
curr_c = ""
while index < length:
curr_c = istr[index]
if curr_c == '//':
index += 2
elif self.is_comment_start(istr, index) and not isString:
index = self._find_end_of_comment(istr, index)
if index == -1:
return -1
elif not isString and self.is_opening_char(curr_c):
deep_close = self._find_closing_char(istr[index:])
if deep_close == -1:
break
index += deep_close
elif curr_c == closing_char:
if curr_c == self.CHAR_DELIM:
index -= 1
return index
index += 1
return -1
def find_key(self, istr, key, index=0, end_index=-1, only_first=True):
"""
Return index of the key or -1.
:param istr: input string; it could be whole file or content of a section
:param key: name of the searched key in the current scope
:param index: start searching from the index
:param end_index: stop searching at the end_index or end of the string
Function is not recursive. Searched key has to be in the current scope.
Attention:
In case that input string contains data outside of section by mistake,
the closing character is ignored and the key outside of scope could be
found. Example of such wrong input could be:
key1 "val"
key2 { key-ignored "val-ignored" };
};
controls { ... };
In this case, the key "controls" is outside of original scope. But for this
cases you can set end_index to value, where searching should end. In case
you set end_index higher then length of the string, end_index will be
automatically corrected to the end of the input string.
"""
length = len(istr)
keylen = len(key)
if length < end_index or end_index < 0:
end_index = length
if index >= end_index or index < 0:
return -1
while index != -1:
if istr.startswith(key, index):
if index+keylen < end_index and istr[index+keylen] not in self.CHAR_KEYWORD:
# key has been found
return index
while not only_first and index != -1 and istr[index] != self.CHAR_DELIM:
index = self.find_next_token(istr, index)
index = self.find_next_token(istr, index)
return -1
def find_next_key(self, cfg, index=0, end_index=-1, end_report=False):
"""Modernized variant of find_key.
:type cfg: ConfigFile
:param index: Where to start search
:rtype: ConfigSection
Searches for first place of bare keyword, without quotes or block.
"""
istr = cfg.buffer
length = len(istr)
if length < end_index or end_index < 0:
end_index = length
if index > end_index or index < 0:
raise IndexError("Invalid cfg index")
while index != -1:
keystart = index
while istr[index] in self.CHAR_KEYWORD and index < end_index:
index += 1
if keystart < index <= end_index and istr[index] not in self.CHAR_KEYWORD:
# key has been found
return ConfigSection(cfg, istr[keystart:index], keystart, index-1)
if istr[index] in self.CHAR_DELIM:
return ConfigSection(cfg, istr[index], index, index)
index = self.find_next_token(istr, index, end_index, end_report)
return None
def find_next_val(self, cfg, key=None, index=0, end_index=-1, end_report=False):
"""Find following token.
:param cfg: input token
:type cfg: ConfigFile
:returns: ConfigSection object or None
:rtype: ConfigSection
"""
start = self.find_next_token(cfg.buffer, index, end_index, end_report)
if start < 0:
return None
if end_index < 0:
end_index = len(cfg.buffer)
# remains = cfg.buffer[start:end_index]
if not self.is_opening_char(cfg.buffer[start]):
return self.find_next_key(cfg, start, end_index, end_report)
end = self._find_closing_char(cfg.buffer, start, end_index)
if end == -1 or (0 < end_index < end):
return None
return ConfigSection(cfg, key, start, end)
def find_val(self, cfg, key, index=0, end_index=-1):
"""Find value of keyword specified by key.
:param cfg: ConfigFile
:param key: name of searched key (str)
:param index: start of search in cfg (int)
:param end_index: end of search in cfg (int)
:returns: ConfigSection object or None
:rtype: ConfigSection
"""
if not isinstance(cfg, ConfigFile):
raise TypeError("cfg must be ConfigFile parameter")
if end_index < 0:
end_index = len(cfg.buffer)
key_start = self.find_key(cfg.buffer, key, index, end_index)
if key_start < 0 or key_start+len(key) >= end_index:
return None
return self.find_next_val(cfg, key, key_start+len(key), end_index)
def find_val_section(self, section, key):
"""Find value of keyword in section.
:param section: section object returned from find_val
Section is object found by previous find_val call.
"""
if not isinstance(section, ConfigSection):
raise TypeError("section must be ConfigSection")
return self.find_val(section.config, key, section.start+1, section.end)
def find_values(self, section, key):
"""Find key in section and list variable parameters.
:param key: Name to statement to find
:returns: List of all found values in form of ConfigSection. First is key itself.
Returns all sections of keyname. They can be mix of "quoted strings", {nested blocks}
or just bare keywords. First key is section of key itself, final section includes ';'.
Makes it possible to comment out whole section including terminal character.
"""
if isinstance(section, ConfigFile):
cfg = section
index = 0
end_index = len(cfg.buffer)
elif isinstance(section, ConfigSection):
cfg = section.config
index = section.start+1
end_index = section.end
if end_index > index:
end_index -= 1
else:
raise TypeError('Unexpected type')
if key is None:
v = self.find_next_key(cfg, index, end_index)
else:
key_start = self.find_key(cfg.buffer, key, index, end_index)
key_end = key_start+len(key)-1
if key_start < 0 or key_end >= end_index:
return None
# First value is always just keyword
v = ConfigSection(cfg, key, key_start, key_end)
values = []
while isinstance(v, ConfigSection):
values.append(v)
if v.value() == self.CHAR_DELIM:
break
v = self.find_next_val(cfg, key, v.end+1, end_index, end_report=True)
return values
def find(self, key_string, cfg=None, delimiter='.'):
"""Helper searching for values under requested sections.
Search for statement under some sections. It is inspired by xpath style paths,
but searches section in bind configuration.
:param key_string: keywords delimited by dots. For example options.dnssec-lookaside
:type key_string: str
:param cfg: Search only in given config file
:type cfg: ConfigFile
:returns: list of ConfigVariableSection
"""
keys = key_string.split(delimiter)
if cfg is not None:
return self._find_values_simple(cfg.root_section(), keys)
items = []
for cfgs in self.FILES_TO_CHECK:
items.extend(self._find_values_simple(cfgs.root_section(), keys))
return items
def is_terminal(self, section):
""".Returns true when section is final character of one statement."""
return section.value() in self.CHAR_DELIM
def _variable_section(self, vl, parent=None, offset=1):
"""Create ConfigVariableSection with a name and optionally class.
Intended for view and zone in bind.
:returns: ConfigVariableSection
"""
vname = self._list_value(vl, 1).invalue()
vclass = None
v = self._list_value(vl, 2)
if v.type() != ConfigSection.TYPE_BLOCK and self._list_value(vl, 2):
vclass = v.value()
return ConfigVariableSection(vl, vname, vclass, parent)
def _list_value(self, vl, i):
n = 0
for v in vl:
if v.type() != ConfigSection.TYPE_IGNORED:
if n == i:
return v
n += 1
raise IndexError
def _find_values_simple(self, section, keys):
found_values = []
sect = section.copy()
while sect is not None:
vl = self.find_values(sect, keys[0])
if vl is None:
break
if len(keys) <= 1:
variable = self._variable_section(vl, section)
found_values.append(variable)
sect.start = variable.end+1
else:
for v in vl:
if v.type() == ConfigSection.TYPE_BLOCK:
vl2 = self._find_values_simple(v, keys[1:])
if vl2 is not None:
found_values.extend(vl2)
sect.start = vl[-1].end+1
return found_values
def walk(self, section, callbacks, state=None, parent=None, start=0):
"""Walk over section also with nested blocks.
:param section: Section to iterate, usually ConfigFile.root_section()
:param callbacks: Set of callbacks with name: f(section, state) parameters, indexed by statement name
:param start: Offset from beginning of section
Call specified actions specified in callbacks, which can react on desired statements.
Pass state and matching section to callback.
"""
if start == 0 and section.type() == ConfigSection.TYPE_BLOCK:
start = 1
it = IscVarIterator(self, section, True, start=section.start+start)
for statement in it:
try:
name = statement.var(0).value()
if name in callbacks:
f = callbacks[name]
f(statement, state)
except IndexError:
pass
for child in statement.values:
if child.type() == ConfigSection.TYPE_BLOCK:
self.walk(child, callbacks, state, parent=statement)
return state
#
# CONFIGURATION fixes PART - END
#
def is_file_loaded(self, path=""):
"""
Checks if the file with a given 'path' is already loaded in FILES_TO_CHECK.
"""
for f in self.FILES_TO_CHECK:
if f.path == path:
return True
return False
def new_config(self, path, parent=None):
config = ConfigFile(path)
self.FILES_TO_CHECK.append(config)
return config
def on_include_error(self, e):
"""Handle IO errors on file reading.
Override to create custom error handling."""
raise e
def load_included_files(self):
"""Add included list to parser.
Finds the configuration files that are included in some configuration
file, reads it, closes and adds into the FILES_TO_CHECK list.
"""
# TODO: use parser instead of regexp
pattern = re.compile(r'include\s*"(.+?)"\s*;')
# find includes in all files
for ch_file in self.FILES_TO_CHECK:
nocomments = self._remove_comments(ch_file.buffer)
includes = re.findall(pattern, nocomments)
for include in includes:
# don't include already loaded files -> prevent loops
if self.is_file_loaded(include):
continue
try:
self.new_config(include)
except IOError as e:
self.on_include_error(ConfigParseError(e, include))
def load_main_config(self):
"""Loads main CONFIG_FILE."""
try:
self.new_config(self.CONFIG_FILE)
except IOError as e:
raise ConfigParseError(e)
def load_config(self, path=None):
"""Loads main config file with all included files."""
if path is not None:
self.CONFIG_FILE = path
self.load_main_config()
self.load_included_files()
pass