|
|
# vim: fileencoding=utf-8
|
|
|
#
|
|
|
# ToDo: добавить проверку того, полностью ли парсился документ. Если отпарсился
|
|
|
# не весь файл -- выдаем ошибку.
|
|
|
#
|
|
|
from .base_format import BaseFormat
|
|
|
from collections import OrderedDict
|
|
|
from pyparsing import originalTextFor, OneOrMore, Word, alphanums, Literal,\
|
|
|
ZeroOrMore, Forward, Optional, Group, restOfLine,\
|
|
|
cppStyleComment, Keyword, printables, nums, SkipTo
|
|
|
|
|
|
|
|
|
class BINDFormat(BaseFormat):
|
|
|
def __init__(self, document_text: str, ignore_comments=False):
|
|
|
processing_methods = []
|
|
|
|
|
|
super().__init__(processing_methods)
|
|
|
|
|
|
self._ignore_comments = ignore_comments
|
|
|
self._comments_processing = True
|
|
|
self._format = 'bind'
|
|
|
|
|
|
self._last_comments_list = []
|
|
|
self._initialize_parser()
|
|
|
|
|
|
if document_text == '':
|
|
|
self._document_dictionary = OrderedDict()
|
|
|
else:
|
|
|
self._parse_text(document_text)
|
|
|
|
|
|
def _initialize_parser(self):
|
|
|
left_brace = Literal('{')
|
|
|
right_brace = Literal('}')
|
|
|
semicolon = Literal(';')
|
|
|
action_symbols = (Literal('!') |
|
|
|
Literal('-'))
|
|
|
|
|
|
plain_allow = Keyword('allow')
|
|
|
drop_allow = Keyword('!allow')
|
|
|
replace_allow = Keyword('-allow')
|
|
|
allow = (plain_allow | drop_allow | replace_allow)
|
|
|
keys = Keyword('keys')
|
|
|
inet = Keyword('inet')
|
|
|
|
|
|
statement = originalTextFor(
|
|
|
Word(alphanums+'_-',
|
|
|
excludeChars='{};')
|
|
|
)
|
|
|
|
|
|
statement_name = originalTextFor(
|
|
|
Word(printables,
|
|
|
excludeChars='{};')
|
|
|
)
|
|
|
|
|
|
statement_class = originalTextFor(
|
|
|
Word(printables,
|
|
|
excludeChars='{};')
|
|
|
)
|
|
|
|
|
|
parameter_value = originalTextFor(
|
|
|
Word(printables,
|
|
|
excludeChars='{};')
|
|
|
)('parameter')
|
|
|
|
|
|
ip_value = originalTextFor(
|
|
|
Word(nums+':./',
|
|
|
excludeChars=';{}')
|
|
|
)
|
|
|
|
|
|
# Будущий парсер блока.
|
|
|
block = Forward()
|
|
|
|
|
|
# Для парсинга директивы inet_spec.
|
|
|
allow_item = (ip_value | statement_name) + semicolon.suppress()
|
|
|
key_item = statement_name + semicolon.suppress()
|
|
|
|
|
|
allow_group = Group(Group(Optional(action_symbols,
|
|
|
default='')('action')
|
|
|
+ allow)
|
|
|
+ Group(left_brace.suppress()
|
|
|
+ ZeroOrMore(allow_item)
|
|
|
+ right_brace.suppress()))
|
|
|
|
|
|
keys_group = Group(Group(Optional(action_symbols, default='')('action')
|
|
|
+ keys)
|
|
|
+ Group(left_brace.suppress()
|
|
|
+ ZeroOrMore(key_item)
|
|
|
+ right_brace.suppress()))
|
|
|
|
|
|
inet_spec = (Group(Optional(action_symbols, default='')('action')
|
|
|
+ inet + originalTextFor(SkipTo(allow,
|
|
|
include=False)
|
|
|
)('parameter')
|
|
|
)('name')
|
|
|
+ Group(allow_group
|
|
|
+ Optional(keys_group)
|
|
|
+ semicolon.suppress())('content')
|
|
|
).setParseAction(
|
|
|
self._add_inet_specline
|
|
|
)
|
|
|
|
|
|
# Для парсинга комментариев.
|
|
|
python_style_comment = originalTextFor(Literal('#') + restOfLine)
|
|
|
comments = (cppStyleComment |
|
|
|
python_style_comment).setParseAction(
|
|
|
self._create_comment_list
|
|
|
)
|
|
|
|
|
|
# Для парсинга директивы include.
|
|
|
include_line = (Optional(action_symbols, default='')('action')
|
|
|
+ Keyword('include')
|
|
|
+ Word(printables, excludeChars=';{}')
|
|
|
+ Optional(semicolon.suppress())
|
|
|
).setParseAction(
|
|
|
self._add_include_line
|
|
|
)
|
|
|
|
|
|
# Для парсинга простых директив состоящих из одного
|
|
|
# или двух параметров.
|
|
|
plain_line = (Group(Optional(action_symbols, default='')('action')
|
|
|
+ statement)('name')
|
|
|
+ Optional(parameter_value)
|
|
|
+ Optional(semicolon.suppress())).setParseAction(
|
|
|
self._add_plain_line
|
|
|
)
|
|
|
|
|
|
# Метод для парсинга IP адресов.
|
|
|
ip_line = (Group(Optional(action_symbols, default='')('action')
|
|
|
+ ip_value
|
|
|
+ Optional(semicolon.suppress()))('IP')
|
|
|
).setParseAction(self._add_ipline)
|
|
|
|
|
|
# Парсеры параметров.
|
|
|
param_line = (include_line |
|
|
|
ip_line |
|
|
|
plain_line)
|
|
|
|
|
|
# Парсер блока параметров.
|
|
|
param_block = (Group(Optional(action_symbols, default='')('action')
|
|
|
+ statement + Optional(statement_name)
|
|
|
+ Optional(statement_class))('name')
|
|
|
+ block('content')
|
|
|
+ Optional(semicolon.suppress())).setParseAction(
|
|
|
self._add_param_block
|
|
|
)
|
|
|
|
|
|
# Виды блочных директив.
|
|
|
block_types = (inet_spec | param_block)
|
|
|
|
|
|
# Парсер параметров с комментариями.
|
|
|
# Note: Применение parser.ignore(comments) является причиной странного
|
|
|
# поведения парсера, при котором невозможно многократное повторное
|
|
|
# применение формата после установки флага ignore_comments.
|
|
|
if self._ignore_comments:
|
|
|
param_line_with_comments = (ZeroOrMore(comments).suppress()(
|
|
|
'comments'
|
|
|
)
|
|
|
+ param_line('value')
|
|
|
).setParseAction(
|
|
|
self._add_comments_to_paramline
|
|
|
)
|
|
|
else:
|
|
|
param_line_with_comments = (ZeroOrMore(comments)('comments')
|
|
|
+ param_line('value')
|
|
|
).setParseAction(
|
|
|
self._add_comments_to_paramline
|
|
|
)
|
|
|
|
|
|
# Парсер блока с комментариями.
|
|
|
if self._ignore_comments:
|
|
|
param_block_with_comments = (ZeroOrMore(comments).suppress()(
|
|
|
'comments'
|
|
|
)
|
|
|
+ block_types('value')
|
|
|
).setParseAction(
|
|
|
self._add_comments_to_block
|
|
|
)
|
|
|
else:
|
|
|
param_block_with_comments = (ZeroOrMore(comments)('comments')
|
|
|
+ block_types('value')
|
|
|
).setParseAction(
|
|
|
self._add_comments_to_block
|
|
|
)
|
|
|
|
|
|
# Парсер содержимого блоков.
|
|
|
block_item = (param_block_with_comments |
|
|
|
param_line_with_comments)
|
|
|
|
|
|
# Для парсинга всего блока с любым содержимым.
|
|
|
block << Group(left_brace.suppress() + ZeroOrMore(block_item)
|
|
|
+ right_brace.suppress())
|
|
|
|
|
|
# Парсер всего документа.
|
|
|
self._document_parser = OneOrMore(block_item)
|
|
|
|
|
|
def _parse_text(self, text):
|
|
|
parsing_result = self._document_parser.parseString(text, parseAll=True)
|
|
|
list_of_elements = parsing_result.asList()
|
|
|
for part in list_of_elements:
|
|
|
self._join_dictionary(self._document_dictionary,
|
|
|
part)
|
|
|
|
|
|
def _join_dictionary(self, out_dictionary, dictionary_to_add):
|
|
|
for key in dictionary_to_add:
|
|
|
if dictionary_to_add == OrderedDict():
|
|
|
return
|
|
|
if key in out_dictionary and\
|
|
|
isinstance(dictionary_to_add[key], OrderedDict) and\
|
|
|
isinstance(out_dictionary[key], OrderedDict):
|
|
|
self._join_dictionary(out_dictionary[key],
|
|
|
dictionary_to_add[key])
|
|
|
else:
|
|
|
out_dictionary[key] = dictionary_to_add[key]
|
|
|
|
|
|
def _add_plain_line(self, current_parse):
|
|
|
name = tuple(current_parse.name.asList())
|
|
|
if not current_parse.parameter == '':
|
|
|
value = current_parse.parameter
|
|
|
else:
|
|
|
if current_parse.name.action == '-':
|
|
|
return OrderedDict()
|
|
|
value = ''
|
|
|
|
|
|
return OrderedDict({name: [value]})
|
|
|
|
|
|
def _add_include_line(self, current_parse):
|
|
|
name = current_parse.asList()
|
|
|
return OrderedDict({tuple(name): ['']})
|
|
|
|
|
|
def _add_ipline(self, current_parse):
|
|
|
ip_value = current_parse.IP
|
|
|
return OrderedDict({tuple(ip_value): ['']})
|
|
|
|
|
|
def _add_inet_specline(self, current_parse):
|
|
|
# Удаляем пробельные символы из второго параметра директивы.
|
|
|
current_parse.name.parameter = current_parse.name.parameter.strip()
|
|
|
block_name = tuple(current_parse.name.asList())
|
|
|
block_content = current_parse.content.asList()
|
|
|
content = OrderedDict({'#': []})
|
|
|
for item in block_content:
|
|
|
current_keyword, values = item
|
|
|
current_keyword = tuple(current_keyword)
|
|
|
content[current_keyword] = values
|
|
|
return OrderedDict({block_name: content})
|
|
|
|
|
|
def _add_param_block(self, current_parse):
|
|
|
block_name = tuple(current_parse.name.asList())
|
|
|
block_content = current_parse.content.asList()
|
|
|
content = OrderedDict({'#': []})
|
|
|
|
|
|
for item in block_content:
|
|
|
self._join_dictionary(content, item)
|
|
|
return OrderedDict({block_name: content})
|
|
|
|
|
|
def _add_comments_to_paramline(self, current_parse):
|
|
|
[parameter] = current_parse.value.asList()
|
|
|
comments = current_parse.comments
|
|
|
if parameter == OrderedDict():
|
|
|
if not comments == '':
|
|
|
self._last_comments_list.extend(comments.asList())
|
|
|
return OrderedDict()
|
|
|
|
|
|
name = next(iter(parameter))
|
|
|
value = parameter[name]
|
|
|
|
|
|
if not comments == '':
|
|
|
comments_list = comments.asList()
|
|
|
parameter[name] = (self._last_comments_list
|
|
|
+ comments_list + value)
|
|
|
self._last_comments_list = []
|
|
|
return parameter
|
|
|
|
|
|
def _add_comments_to_block(self, current_parse):
|
|
|
[value] = current_parse.value
|
|
|
[block_name] = value
|
|
|
block = value[block_name]
|
|
|
|
|
|
if not current_parse.comments == '':
|
|
|
block['#'] = (self._last_comments_list
|
|
|
+ current_parse.comments.asList())
|
|
|
self._last_comments_list = []
|
|
|
else:
|
|
|
block.pop('#')
|
|
|
return value
|
|
|
|
|
|
def _create_comment_list(self, current_parse):
|
|
|
comments_list = []
|
|
|
comments = current_parse.asList()
|
|
|
for comment in comments:
|
|
|
lines = comment.splitlines()
|
|
|
for line in lines:
|
|
|
comments_list.append(line.strip())
|
|
|
return comments_list
|
