502 lines
23 KiB
Python
502 lines
23 KiB
Python
from validator_common import *
|
|
from copy import deepcopy
|
|
from merger import MergerStorage
|
|
from merger import Merger as MergerImpl
|
|
from merger import MergerBranch as MergerBranchImpl
|
|
from merger import MergerLastBranch as MergerLastBranchImpl
|
|
from merger import MergerRejectBranch as MergerRejectBranchImpl
|
|
from merger import MergerRule as MergerRuleImpl
|
|
import itertools
|
|
import allen_ops
|
|
import options
|
|
|
|
class MergerValidator(object):
|
|
def __init__(self, parser, gr_filter_validator):
|
|
self.parser = parser
|
|
self.gr_filter_validator = gr_filter_validator
|
|
self.mergers = deepcopy(parser.mergers)
|
|
# The last field returns a list of the present fields for each branch
|
|
# ('rec_id', 'Last', 'First', 'records', 'srcip', 'dstip', 'bytes', 'n', 'flags', 'srcports')
|
|
# ('rec_id', 'Last', 'First', 'records', 'srcip', 'dstip', 'bytes', 'n', 'flags', 'dstports')
|
|
self.branches_fields = gr_filter_validator.branches_fields
|
|
# A simple dictionary mapptin of branch name to a GroupFilter
|
|
# {'A': <groupfilter.GroupFilter object at 0x9c3d66c>, 'B': <groupfilter.GroupFilter object at 0x9c43ccc>}
|
|
self.br_name_to_gr_filter = gr_filter_validator.br_name_to_gr_filter
|
|
# Checks that all the defined merger modules are actually exported
|
|
# Returns a dictionay of a merger name and module implementation
|
|
self.megers_export_modules = self.find_mergers_export_modules()
|
|
# Returns the size of the field type of the 'records' field, 4 bytes
|
|
self.id_size = self.get_id_size()
|
|
self.impl = self.get_mergers_impl()
|
|
|
|
# Returns the size of the field type of the 'records' field, 4 bytess
|
|
def get_id_size(self):
|
|
rec_reader = self.gr_filter_validator.impl[0].records
|
|
field_types = dict(zip(rec_reader.group_record_fields,
|
|
rec_reader.group_record_types))
|
|
id_size = field_types['records'].itemsize
|
|
return id_size
|
|
|
|
# Check for duplicate merger names
|
|
def check_duplicate_merger_names(self):
|
|
duplicates = {}
|
|
for merger in self.mergers:
|
|
old_val = duplicates.setdefault(merger.name, 0)
|
|
duplicates[merger.name] = old_val + 1
|
|
|
|
duplicate_names = [k for k,v in duplicates.iteritems() if v > 1]
|
|
if len(duplicate_names) > 0:
|
|
msg = "Merger(s) %s"%duplicate_names
|
|
msg += " is/are all defined more than once."
|
|
raise SyntaxError(msg)
|
|
|
|
# Check for duplicate module names
|
|
def check_duplicate_module_names(self, merger):
|
|
duplicates = {}
|
|
for module in merger.modules:
|
|
old_val = duplicates.setdefault(module.name, 0)
|
|
duplicates[module.name] = old_val + 1
|
|
|
|
duplicate_names = [k for k,v in duplicates.iteritems() if v > 1]
|
|
if len(duplicate_names) > 0:
|
|
msg = "Module(s) %s"%duplicate_names
|
|
msg += " is/are all defined more than once in merger"
|
|
msg += " %s."%merger.name
|
|
raise SyntaxError(msg)
|
|
|
|
|
|
# Checks that all the defined merger modules are actually exported
|
|
# Returns a dictionay of a merger name and module implementation
|
|
def find_mergers_export_modules(self):
|
|
merger_to_export_module = {}
|
|
for merger in self.mergers:
|
|
exp = None
|
|
for module in merger.modules:
|
|
if merger.export == module.name:
|
|
exp = module
|
|
break
|
|
|
|
if exp:
|
|
merger_to_export_module[merger.name] = exp
|
|
# print merger_to_export_module
|
|
else:
|
|
msg = "Merger %s"%merger.name
|
|
msg += " export module %s is not defined."%merger.export
|
|
|
|
return merger_to_export_module
|
|
|
|
#--------------------------------------ALLEN CHECKS-------------------------------------#
|
|
#All the operations on rules are around a sample set like: {'M': Module('m1', 38, [[Rule('EQ', 40, [Field('A.srcip'), Field('B.dstip')], False)], [Rule('EQ', 41, [Field('A.srcports'), Field('B.dstports')], False)], [Rule('LT', 42, [Field('A.bytes'), Field('B.bytes')], False)], [AllenRule('oi', 43, [Field('B'), Field('A')], False), AllenRule('d', 43, [Field('B'), Field('A')], False)]], ['B', 'A'])}
|
|
|
|
#Returns only the Allen rules
|
|
def iterate_module_allen_op_groups(self, merger):
|
|
for module in merger.modules:
|
|
for rules in module.rules:
|
|
if type(rules[0]) is not AllenRule:
|
|
continue
|
|
else:
|
|
for op in rules:
|
|
yield op
|
|
|
|
# Orders allen operations and the branches that they influence in a reverse order, if not already so
|
|
def order_allen_ops_args(self, merger):
|
|
order = self.get_merger_branches_order(merger)#Orders merger branches, exported module's branches being first
|
|
arg_combinaions = tuple(itertools.combinations(order, 2))#combinations('ABCD', 2) --> AB AC AD BC BD CD
|
|
for allen_op in self.iterate_module_allen_op_groups(merger):#Returns only the Allen rules
|
|
first, second = allen_op.args[:2] # Returns Field('B') Field('A')
|
|
op = allen_op.op # operations like oi, d
|
|
if (first.name, second.name) not in arg_combinaions:
|
|
allen_op.args = [second, first] + allen_op.args[2:]# reverse names
|
|
allen_op.op = allen_ops.inv_op_str(op)# and operations
|
|
|
|
# A number of different checks of the AllenRule
|
|
def check_allen_ops(self, merger):
|
|
allen_arg_pairs = []
|
|
arg_pairs_to_line = {}
|
|
for module in merger.modules:
|
|
for rules in module.rules:
|
|
if type(rules[0]) is not AllenRule:
|
|
continue
|
|
|
|
first_arg = rules[0].args[0].name # Get the branch names influenced by the AllenRule
|
|
second_arg = rules[0].args[1].name
|
|
line = rules[0].line
|
|
order = (first_arg, second_arg)
|
|
allen_arg_pairs.append(order)# [('B', 'A')]
|
|
|
|
self.check_allen_satisfiability(arg_pairs_to_line, order, line)
|
|
self.check_allen_consistency(first_arg, second_arg, rules)
|
|
self.check_allen_deltas(rules)
|
|
|
|
self.check_allen_reachability(allen_arg_pairs, merger)
|
|
|
|
# The following 3 methods run different tests on the allen arguments and rules
|
|
def check_allen_satisfiability(self, arg_pairs_to_line, order, line):
|
|
if arg_pairs_to_line.has_key(order):
|
|
msg = "Unsatisfiable Allen op group. "
|
|
msg += "All allen ops concerning a pair of branches should"
|
|
msg += " be connected with and OR into a single group "
|
|
msg += "within a single module.\n"
|
|
msg += "Argument pair %s on line %s"%(order, line)
|
|
msg += " is also used on line %s."%arg_pairs_to_line[order]
|
|
raise SyntaxError(msg)
|
|
else:
|
|
arg_pairs_to_line[order] = line
|
|
def check_allen_consistency(self, first_arg, second_arg, rules):
|
|
for al_op in rules:
|
|
first = al_op.args[0].name
|
|
second = al_op.args[1].name
|
|
|
|
if (first != first_arg or second != second_arg):
|
|
msg = "Inconsistent group of Allen statements "
|
|
msg += "on line %s"%rules[0].line
|
|
msg += ": %s, %s.\n"%(first, second)
|
|
msg += "All branches in this group should have "
|
|
msg += "%s and %s"%(first_arg, second_arg)
|
|
msg += " as left and righthand side arguments "
|
|
msg += "respectively."
|
|
raise SyntaxError(msg)
|
|
def check_allen_deltas(self, rules):
|
|
for al_op in rules:
|
|
if al_op.op == 'LT' or al_op.op == 'GT':
|
|
if len(al_op.args) < 3:
|
|
msg = "Allen op < or > on line %s "%al_op.line
|
|
msg += " should have delta explicitly stated."
|
|
raise SyntaxError(msg)
|
|
# A check for reachability of subsequent branches from the first one
|
|
def check_allen_reachability(self, allen_arg_pairs, merger):
|
|
br_order = self.get_merger_branches_order(merger)
|
|
# check reachability through allen index from initial branch
|
|
# of export module:
|
|
reachable = br_order[0:1] # list of first branch of exp module
|
|
unreachable = br_order[1:]
|
|
change = True
|
|
while(change):
|
|
change = False
|
|
for arg1, arg2 in allen_arg_pairs:
|
|
if arg1 in reachable and arg2 in unreachable:
|
|
unreachable.remove(arg2)
|
|
reachable.append(arg2)
|
|
change = True
|
|
if len(unreachable) > 0:
|
|
msg = "Branch(es): %s"%unreachable
|
|
msg += " in merger %s"%merger.name
|
|
msg += " is/are unreachable through an allen op or chain of"
|
|
msg += " allen ops from the first branch of the exported module"
|
|
raise SyntaxError(msg)
|
|
#--------------------------------------END ALLEN CHECKS---------------------------------#
|
|
|
|
# Orders the merger modules s.t. the exported module comes first
|
|
def order_modules(self):
|
|
for merger in self.mergers:
|
|
exp_module = self.megers_export_modules[merger.name]
|
|
new_modules_order = [exp_module]
|
|
new_modules_order += [m for m in merger.modules if m != exp_module]
|
|
merger.modules = new_modules_order
|
|
|
|
# Checks that the modules are interconnected among each other with at least one branch
|
|
def check_for_disjoint_modules(self):
|
|
for merger in self.mergers:
|
|
exp_module = self.megers_export_modules[merger.name]
|
|
exp_branches = set(exp_module.branches)
|
|
for module in merger.modules:
|
|
branches = set(module.branches)
|
|
# NOTE & is set intersection
|
|
if len(exp_branches & branches) < 1:
|
|
msg = "Merger module %s.%s"%(merger.name,module.name)
|
|
msg += " in has no overlaping branches with the"
|
|
msg += " export module."
|
|
raise SyntaxError(msg)
|
|
|
|
|
|
# Check the validity of the AllenRule, by seeing if the branch names are all defined
|
|
def check_branch_id_ref(self, rule, module_branches):
|
|
for arg in rule.args:
|
|
if type(arg) is Field:
|
|
id_ref = arg.name
|
|
if id_ref not in self.br_name_to_gr_filter.keys():
|
|
msg = 'Branch %s referenced on line'%id_ref
|
|
msg += ' %s is not defined.'%rule.line
|
|
raise SyntaxError(msg)
|
|
if id_ref not in module_branches:
|
|
msg = 'Branch %s referenced on line'%id_ref
|
|
msg += " %s "%rule.line
|
|
msg += "is not in module's branches statement."
|
|
raise SyntaxError(msg)
|
|
|
|
# Check the validity of the Rule, GrouperRule and statements like A.bytes
|
|
def check_qid_field_ref(self, rule, module_branches):
|
|
for arg in rule.args:
|
|
if type(arg) is Field:
|
|
qid_field = arg.name
|
|
branch, _, field = qid_field.partition('.') #Separates statements like A.bytes
|
|
try:
|
|
if field not in self.branches_fields[branch]:
|
|
msg = 'Wrong field %s on line %s. '%(qid_field,
|
|
rule.line)
|
|
msg += 'Branch %s does not have field %s.'%(branch,
|
|
field)
|
|
raise SyntaxError(msg)
|
|
except KeyError:
|
|
msg = 'Branch %s referenced on line'%branch
|
|
msg += ' %s is not defined'%rule.line
|
|
raise SyntaxError(msg)
|
|
if branch not in module_branches:
|
|
msg = 'Branch %s referenced on line'%branch
|
|
msg += " %s "%rule.line
|
|
msg += "is not in module's branches statement."
|
|
raise SyntaxError(msg)
|
|
|
|
# Orders merger branches with the exported module's branches being first
|
|
def get_merger_branches_order(self, merger):
|
|
br_order = []
|
|
# first add export module
|
|
for module in merger.modules:
|
|
if module.name == merger.export:
|
|
for br in module.branches:
|
|
if br not in br_order:
|
|
br_order.append(br)
|
|
|
|
# add all the others:
|
|
for module in merger.modules:
|
|
for br in module.branches:
|
|
if br not in br_order:
|
|
br_order.append(br)
|
|
|
|
return br_order
|
|
|
|
#
|
|
def order_merger_rules(self, merger):
|
|
"""
|
|
Produces mapping between incrementally larger available branches tuples
|
|
(A,B,C,etc) ordered as they will appear in the implementation.
|
|
"""
|
|
br_order = self.get_merger_branches_order(merger)
|
|
needed_brs_to_rule = {}
|
|
for module in merger.modules:
|
|
replace_with_vals(module)
|
|
replace_bound_rules(module)
|
|
for rules in module.rules:
|
|
rule_branches = self.get_rule_needed_branches(rules[0])
|
|
|
|
ordered_branches = tuple(br for br in br_order
|
|
if br in rule_branches)
|
|
|
|
if len(rules) > 1:
|
|
rule = Rule('or_op', 0, rules)
|
|
else:
|
|
rule = rules[0]
|
|
needed_brs_to_rule.setdefault(ordered_branches,
|
|
[]).append(rule)
|
|
|
|
avail_to_rules = {}
|
|
tup = ()
|
|
# create sets - needed for the set intersection operation
|
|
needed_sets = map(set, needed_brs_to_rule.keys())
|
|
# incrementaly add branches to the tuple of available branches
|
|
# and check which rules have their branch needs satisfied
|
|
for br in br_order:
|
|
tup += (br,)
|
|
# find how many of the needed branches are in this tuple
|
|
# of branches. It makes elementwise intesection of the sets
|
|
# of the needed branches and the tuple of available branches
|
|
intersect = map(set(tup).intersection , needed_sets )
|
|
for el, intersection, key in zip(needed_sets , intersect,
|
|
needed_brs_to_rule.keys()):
|
|
if len(intersection) == len(el):
|
|
# Lenght is the same, which means all needed branches
|
|
# are present. Remove these elements, take the rules from
|
|
# the needed_brs_to_rule and delete the key their to
|
|
# keep the zip() in sync
|
|
needed_sets.remove(el)
|
|
avail_to_rules[tup] = needed_brs_to_rule[key]
|
|
del needed_brs_to_rule[key]
|
|
return avail_to_rules
|
|
|
|
#
|
|
def get_rule_needed_branches(self, rule):
|
|
args_list = set()
|
|
for sub_rule in iterate_subrules(rule):
|
|
for arg in sub_rule.args:
|
|
if type(arg) is Field:
|
|
args_list.add(arg.name)
|
|
|
|
for arg in rule.args:
|
|
if type(arg) is Field:
|
|
args_list.add(arg.name)
|
|
|
|
if type(rule) is AllenRule:
|
|
return list(args_list)
|
|
|
|
else:
|
|
return [qid.partition('.')[0] for qid in args_list]
|
|
|
|
|
|
# Validates the correctness of the merger stage
|
|
def validate(self):
|
|
self.check_duplicate_merger_names()
|
|
for merger in self.mergers:
|
|
self.check_duplicate_module_names(merger)
|
|
for module in merger.modules:
|
|
# Checks the whole rule list to see that all
|
|
# the rules fall into [Rule, GrouperRule, AllenRule]
|
|
# Returns the actual rules
|
|
for rule in iterate_rules(module):
|
|
# Checks that all the rule entries are correctly specified
|
|
if type(rule) is AllenRule:
|
|
self.check_branch_id_ref(rule, module.branches)
|
|
else:
|
|
self.check_qid_field_ref(rule, module.branches)
|
|
# Orders allen operations and the branches that they influence in a reverse order
|
|
self.order_allen_ops_args(merger)
|
|
# Performs several checks on the branches and the operations (consistency, reachability, etc.)
|
|
self.check_allen_ops(merger)
|
|
|
|
# Orders the merger modules s.t. the exported module comes first
|
|
self.order_modules()
|
|
# Checks that the modules are interconnected among each other with at least one branch
|
|
self.check_for_disjoint_modules()
|
|
|
|
|
|
# Get the allen indexing operations for each branch.
|
|
def get_branches_allen_index_ops(self, merger):
|
|
"""
|
|
Get the allen indexing operations for each branch.
|
|
"""
|
|
br_to_allen_ind_ops = {}
|
|
for module in merger.modules:
|
|
for rules in module.rules:
|
|
if type(rules[0]) != AllenRule:
|
|
continue
|
|
br = rules[0].args[0].name
|
|
br_to_allen_ind_ops.setdefault(br, []).append(rules)
|
|
return br_to_allen_ind_ops
|
|
|
|
#
|
|
def get_rule_impl(self, rule, br_to_record):
|
|
if type(rule) == AllenRule:
|
|
op = find_op(rule, module='allen_ops')
|
|
args = [ (arg.name, None)
|
|
if type(arg) == Field else arg
|
|
for arg in rule.args]
|
|
else:
|
|
args = []
|
|
op = find_op(rule)
|
|
for arg in rule.args:
|
|
if type(arg) == Rule:
|
|
arg_impl = self.get_rule_impl(arg, br_to_record)
|
|
elif type(arg) == Field:
|
|
branch, _, field = arg.name.partition('.')
|
|
arg_impl = (branch, field)
|
|
else:
|
|
arg_impl = arg
|
|
|
|
args.append(arg_impl)
|
|
return MergerRuleImpl(op, args, br_to_record)
|
|
|
|
# Create indexing rules implementation for AllenRules
|
|
def get_index_rule_impl(self, rules):
|
|
res = []
|
|
for or_rules in rules:
|
|
or_rules_impl = []
|
|
for rule in or_rules:
|
|
op = find_op(rule, 'allen_index')
|
|
args = [arg.name if type(arg) == Field else arg
|
|
for arg in rule.args]
|
|
# replace with values
|
|
args = [arg.value if type(arg) == Arg else arg
|
|
for arg in args]
|
|
#[<allen_index.oi object at 0x9f5adcc>, <allen_index.d object at 0x9f5ae0c>]
|
|
or_rules_impl.append(op(*args))
|
|
res.append(or_rules_impl)
|
|
return res
|
|
|
|
# Creates a file MergedM.h5 for further storage of the merged files
|
|
def get_merger_table_impl(self, merger):
|
|
fields = self.megers_export_modules[merger.name].branches
|
|
types = [UIntCol(self.id_size) for _ in fields]
|
|
field_types = dict(zip(fields,types))
|
|
recordClass = record.get_record_class(fields, types)
|
|
# TODO fix file names
|
|
fname = options.temp_path + options.merger_file_prefix
|
|
fname += merger.name + ".h5"
|
|
if options.delete_temp_files: if_exists_delete(fname)
|
|
pytables.create_table_file(fname, field_types)
|
|
mergerTable = FlowRecordsTable(fname)
|
|
|
|
return MergerStorage(merger.name, mergerTable, recordClass)
|
|
|
|
# Actual implementation of the merger stage
|
|
def get_merger_impl(self, merger):
|
|
# Create merger storage
|
|
merger_table = self.get_merger_table_impl(merger)
|
|
|
|
# Create indexing rules implementation
|
|
br_to_index_rule_impl = {}
|
|
|
|
# {'B': [[AllenRule('oi', 43, [Field('B'), Field('A')], False), AllenRule('d', 43, [Field('B'), Field('A')], False)]]}
|
|
for br, rules in self.get_branches_allen_index_ops(merger).iteritems():
|
|
br_to_index_rule_impl[br] = self.get_index_rule_impl(rules)# List of allen index rules implemented
|
|
|
|
for br in self.get_merger_branches_order(merger):#orders branches with the exported branch being first
|
|
if br not in br_to_index_rule_impl.keys():
|
|
br_to_index_rule_impl[br] = []
|
|
|
|
# some "globals" shared among branches or needed for their creation
|
|
needed_brs = self.order_merger_rules(merger) # Re-orders the rules as they will appear in the implementation
|
|
tup = () # tuple of available branches
|
|
name = merger.name
|
|
br_order = self.get_merger_branches_order(merger) # Returns reversely-ordered branch names of the merger
|
|
export_branches = self.megers_export_modules[merger.name].branches # Returns branch names contained in the export module
|
|
br_to_record = {}
|
|
name_to_branch = {}
|
|
merger_impl = None
|
|
for br_name in br_order: # For each branch in the ordered branch set
|
|
tup += (br_name,)
|
|
next_branches_names = [br for br in br_order if br not in tup]
|
|
records = self.br_name_to_gr_filter[br_name] # Group-filters associated with each branch
|
|
index_rules = br_to_index_rule_impl[br_name] # Allen index rule associated with each branch
|
|
index = records.index # Time index object
|
|
if len(tup)<2: # If tuple contains only one branch, then execute the initial Merger class
|
|
# first branch
|
|
rules = []
|
|
impl = MergerImpl(name, br_name, records, name_to_branch,
|
|
next_branches_names, export_branches,
|
|
br_to_record, index, index_rules, rules,
|
|
merger_table)
|
|
merger_impl = impl
|
|
else:
|
|
unimpl_rules = needed_brs[tup]
|
|
rules = [self.get_rule_impl(rule, br_to_record)
|
|
for rule in unimpl_rules]
|
|
if br_name not in export_branches:
|
|
# Reject branch
|
|
impl = MergerRejectBranchImpl(br_name, records,
|
|
name_to_branch, next_branches_names,
|
|
export_branches, br_to_record, index,
|
|
index_rules, rules, merger_table)
|
|
|
|
elif not next_branches_names:
|
|
# Last non-rejecting branch
|
|
impl = MergerLastBranchImpl(br_name, records,
|
|
name_to_branch, next_branches_names,
|
|
export_branches, br_to_record, index,
|
|
index_rules, rules, merger_table)
|
|
|
|
else:
|
|
# For normal middle branches execute the MergerBranch class
|
|
impl = MergerBranchImpl(br_name, records, name_to_branch,
|
|
next_branches_names, export_branches,
|
|
br_to_record, index, index_rules,
|
|
rules, merger_table)
|
|
|
|
name_to_branch[br_name] = impl
|
|
|
|
return merger_impl
|
|
|
|
def get_mergers_impl(self):
|
|
self.validate()
|
|
mergers_impl = [self.get_merger_impl(merger)
|
|
for merger in self.mergers]
|
|
|
|
return mergers_impl
|