Java教程
OSSIM-agent源代码分析（四）
本文主要是介绍OSSIM-agent源代码分析（四），对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！
2021SC@SDUSC
OSSIM-agent源代码分析（四）
OSSIM-agent-config.py下
简述
OSSIM Agent的主要职责是收集网络上存在的各种设备发送的所有数据，然后按照一种标准方式（standardized way）有序的发送给OSSIM Server，Agent收集到数据后在发送给Server之前要对这些数据进行标准化处理，这样Server就可以依一种统一的方式来处理这些信息，并且也简化了Server的处理过程
agent的配置文件是config.py，作为大量数据的标准化处理位置，配置文件的对各种数据、文件的处理方式的
相关方法
在plugin.cfg文件中显示，各种常量的设置
_NEEDED_CONFIG_ENTRIES = {
'config': ['type', 'source', 'enable']
}
_EXIT_IF_MALFORMED_CONFIG = False
TRANSLATION_SECTION = 'translation'
TRANSLATION_FUNCTION = 'translate'
TRANSLATION_DEFAULT = '_DEFAULT_'
SECTIONS_NOT_RULES = ["config", "info", TRANSLATION_SECTION]
CONCAT_FUNCTION = 'CONCAT'
_MAP_REPLACE_TRANSLATIONS = 4
_MAP_REPLACE_USER_FUNCTIONS = 8
_MAP_REPLACE_CUSTOM_USER_FUNCTIONS = 16
_MAP_REPLACE_CONCAT = 32 #00100000
_MAP_REPLACE_TRANSLATE2 = 64 #1000000
检查所选部分是否为可以的翻译部分
__regexIsTranslationSection = re.compile("translation-\S+", re.UNICODE)
__regex_check_for_translate2_function = re.compile("\{translate2\((?P<variable>\$[^\)]+),(?P<translate_section>\$[^\)]+)\)\}", re.UNICODE)
设置配置、翻译等具体的规则
def rules(self):
rules = {}
for section in sorted(self.sections()):
regexp = self.get(section,"regexp")
if self.get("config","source") == "log" and (regexp is None or regexp == ""):
continue
if Plugin.__regexIsTranslationSection.match(section.lower()) is not None:
if section.lower() not in Plugin.SECTIONS_NOT_RULES:
logger.info("Loading new translation section... %s" % section.lower())
Plugin.SECTIONS_NOT_RULES.append(section.lower())
if section.lower() not in Plugin.SECTIONS_NOT_RULES:
rules[section] = self.hitems(section,True)
return rules
通过正则表达式和位置变换进行变量替换
def _replace_array_variables(self, value, groups):
for i in range(2):
search = re.findall("\{\$[^\}\{]+\}", value, re.UNICODE)
rvalue = value
if search != []:
for string in search:
var = string[2:-1]
var_position = 0
try:
var_position= int(var)
if var_position < len(groups):
rvalue = rvalue.replace(string, str(groups[var_position]))
except ValueError,e:
rvalue = value
return rvalue
替换变量、用户功能和自定义用户功能
def get_replace_array_value(self, value, groups):
rvalue = ""
rvalue = self._replace_array_variables(value, groups)
if rvalue==value:
rvalue = self._replace_user_array_functions(value, groups)
if rvalue == value:
rvalue = self._replace_custom_user_function_array(value,groups)
return rvalue
使用位置参数替换自定义用户函数，并且检查所有变量是否有替换，以及调用函数在ParserUtil.py
def _replace_custom_user_function_array(self,value,groups):
search = re.findall("(\{:(\w+)\((\$[^\)]+)?\)\})", value, re.UNICODE)
if search != []:
for string in search:
(string_matched, func, variables) = string
vars = split_variables(variables)
for var in vars:
var_pos =0
try:
var_pos = int(var)
except TypeError:
logger.warning("Can not replace '%s'" % var)
return value
if var_pos >= len(groups):
logger.warning("Can not replace '%s' " % var)
return value
func_name = "%s_%s" % (func,self.get("DEFAULT", "plugin_id"))
if func_name != Plugin.TRANSLATION_FUNCTION and \
hasattr(Plugin, func_name):
args = ",".join(["groups[%s]"%s for s in vars])
try:
cmd = "value = value.replace(string_matched, str(self.%s(%s)))" %(func_name,args)
exec cmd
except TypeError, e:
logger.error(e)
else:
logger.warning( "Function '%s' is not implemented" % (func))
value = value.replace(string_matched, str(groups[var]))
return value
替换位置函数，检查所有变量是否有替换，然后调用ParserUtil.py函数、getattr等函数执行替换功能
def _replace_user_array_functions(self, value, groups):
if value is None:
return None
search = re.findall("(\{(\w+)\((\$[^\)]+)\)\})", value, re.UNICODE)
if search != []:
for string in search:
(string_matched, func, variables) = string
vars = split_variables(variables)
for var in vars:
if len(groups) - 1 < int(var):
return value
if func != Plugin.TRANSLATION_FUNCTION and \
hasattr(ParserUtil, func):
f = getattr(ParserUtil, func)
args = ""
for v in vars:
args += "str(groups[%s])," % (str(v))
try:
exec "value = value.replace(string_matched," + \
"str(f(" + args + ")))"
except TypeError, e:
logger.error(e)
else:
logger.debug("Tranlation fucntion...")
for v in vars:
.
if int(v) > (len(groups) - 1):
logger.debug("Error var:%d, is greatter than groups size. (%d)" % (int(v), len(groups)))
else:
var_value = groups[int(v)]
if self.has_section(Plugin.TRANSLATION_SECTION):
if self.has_option(Plugin.TRANSLATION_SECTION, var_value):
value = self.get(Plugin.TRANSLATION_SECTION, var_value)
return value
通过正则表达式，查找配置参数中的_CFG(section,option)值，并将其替换为全局配置文件中的值
def replace_config(self, conf):
for section in sorted(self.sections()):
for option in self.options(section):
regexp = self.get(section, option)
search = re.findall("(\\\\_CFG\(([\w-]+),([\w-]+)\))", regexp, re.UNICODE)
if search != []:
for string in search:
(all, arg1, arg2) = string
if conf.has_option(arg1, arg2):
value = conf.get(arg1, arg2)
regexp = regexp.replace(all, value)
self.set(section, option, regexp)
总规则下，regexp条目中替换处理每一个反斜线
def replace_aliases(self, aliases)
for rule in self.rules().iterkeys():
regexp = self.get(rule, 'regexp')
search = re.findall("\\\\\w\w+", regexp, re.UNICODE)
if search != []:
for string in search:
repl = string[1:]
if aliases.has_option("regexp", repl):
value = aliases.get("regexp", repl)
regexp = regexp.replace(string, value)
self.set(rule, "regexp", regexp)
从get_replace_value()调用，通过正则替换变量
def _replace_variables(self, value, groups, rounds=2):
for i in range(rounds):
search = re.findall("\{\$[^\}\{]+\}", value, re.UNICODE)
if search != []:
for string in search:
var = string[2:-1]
if groups.has_key(var):
value = value.replace(string, str(groups[var]))
return value
确定替换变量是否成功，如果失败，则跳过get_replace_value()
def _replace_variables_assess(self, value):
ret = 0
for i in range(2):
search = re.findall("\{\$[^\}\{]+\}", value, re.UNICODE)
if search != []:
ret = i
return ret
通过搜索字符串方式，替换指定插件及其对应插件
def _replace_translations(self, value, groups):
regexp = "(\{(" + Plugin.TRANSLATION_FUNCTION + ")\(\$([^\)]+)\)\})"
search = re.findall(regexp, value, re.UNICODE)
if search != []:
for string in search:
(string_matched, func, var) = string
if groups.has_key(var):
if self.has_section(Plugin.TRANSLATION_SECTION):
if self.has_option(Plugin.TRANSLATION_SECTION,
groups[var]):
value = self.get(Plugin.TRANSLATION_SECTION,
groups[var])
else:
logger.warning("Can not translate '%s' value" % \
(groups[var]))
if self.has_option(Plugin.TRANSLATION_SECTION,
Plugin.TRANSLATION_DEFAULT):
value = self.get(Plugin.TRANSLATION_SECTION,
Plugin.TRANSLATION_DEFAULT)
else:
value = groups[var]
else:
logger.warning("There is no translation section")
value = groups[var]
return value
同上，确定替换变量是否成功，如果失败，则跳过get_replace_value()
def _replace_translations_assess(self, value):
regexp = "(\{(" + Plugin.TRANSLATION_FUNCTION + ")\(\$([^\)]+)\)\})"
search = re.findall(regexp, value, re.UNICODE)
if search != []:
return self._MAP_REPLACE_TRANSLATIONS
return 0
函数被指定为{f($v)}，并且从get_replace_value()调用，执行替换功能
def _replace_user_functions(self, value, groups):
search = re.findall("(\{(\w+)\((\$[^\)]+)\)\})", value, re.UNICODE)
if search != []:
for string in search:
(string_matched, func, variables) = string
vars = split_variables(variables)
for var in vars:
if not groups.has_key(var):
logger.warning("Can not replace '%s'" % (value))
return value
if func != Plugin.TRANSLATION_FUNCTION and \
hasattr(ParserUtil, func):
f = getattr(ParserUtil, func)
args = ""
for i in (range(len(vars))):
args += "groups[vars[%s]]," % (str(i))
try:
cmd = "value = value.replace(string_matched," + \
"str(f(" + args + ")))"
exec cmd
except TypeError, e:
logger.error(e)
else:
logger.warning(
"Function '%s' is not implemented" % (func))
value = value.replace(string_matched, \
str(groups[var]))
return value
同上，确定替换变量是否成功，如果失败，则跳过get_replace_value()
def _replace_user_functions_assess(self, value):
search = re.findall("(\{(\w+)\((\$[^\)]+)\)\})", value, re.UNICODE)
if search != []:
return self._MAP_REPLACE_USER_FUNCTIONS
return 0
检查所有变量是否有替换，如果没有一直替换到全部替换为止
def _replace_custom_user_functions(self,value,groups):
search = re.findall("(\{:(\w+)\((\$[^\)]+)?\)\})", value, re.UNICODE)
if search != []:
for string in search:
(string_matched, func, variables) = string
vars = split_variables(variables)
for var in vars:
if not groups.has_key(var):
logger.warning("Can not replace '%s'" % (var))
return value
func_name = "%s_%s" % (func,self.get("DEFAULT", "plugin_id"))
if func_name != Plugin.TRANSLATION_FUNCTION and \
hasattr(Plugin, func_name):
args = ""
for i in (range(len(vars))):
args += "groups[vars[%s]]," % (str(i))
try:
cmd = "value = value.replace(string_matched, str(self.%s(%s)))" %(func_name,args)
exec cmd
except TypeError, e:
logger.error(e)
else:
logger.warning(
"Function '%s' is not implemented" % (func))
value = value.replace(string_matched, \
str(groups[var]))
return value
替换数组变量连接的值
def __replaceConcatFunction(self,value,groups):
concat =""
m = re.match("\$CONCAT\((?P<params>.*)\)",value)
if m:
mdict = m.groupdict()
if mdict.has_key('params'):
paramlist = mdict['params'].split(',')
for param in paramlist:
if param.startswith('$'):#assume variable
if groups.has_key(param[1:]):
concat+=groups[param[1:]]
else:
concat+=param
else:
concat+=param
return concat
检查是否有必要进行连接函数替换
def __checkReplaceConcatFunction(self,value):
ret = 0
if re.match("\$CONCAT\((.*)\)",value):
ret = self._MAP_REPLACE_CONCAT
return ret
替换传递过来的对象中给定的值
def __replace_translate2_function(self,value,groups):
replaced_value =""
m = self.__regex_check_for_translate2_function.match(value)
if m:
mdict = m.groupdict()
if 'variable' in mdict and 'translate_section' in mdict:
variable = mdict['variable'].replace('$','')
translate_section = mdict['translate_section'].replace('$','')
if variable in groups:
if self.has_section(translate_section):
if self.has_option(translate_section,groups[variable]):
replaced_value = self.get(translate_section, groups[variable])
return replaced_value
else:
logger.warning("Translate2 variable(%s) not found on the translate section %s" % (groups[variable],translate_section))
else:
logger.warning("Translate2 translation_section(%s) not found" % translate_section)
replaced_value = value
检查是否需要换translate2
def __check_for_translate2_function(self,value):
"""Check whether we need to make a translate2 replacement or not.
@param value: the string in the rule
"""
ret = 0
if Plugin.__regex_check_for_translate2_function.match(value) is not None:
ret = self._MAP_REPLACE_TRANSLATE2
return ret
替换concat函数、自定义用户函数、用户函数、翻译、变量等
def get_replace_value(self, value, groups, replace=15):
if replace > 0:
if replace & self._MAP_REPLACE_TRANSLATE2:
value = self.__replace_translate2_function(value, groups)
if replace & 3:
value = self._replace_variables(value, groups, (replace & 3))
if replace & 4:
value = self._replace_translations(value, groups)
if replace & 8:
value = self._replace_user_functions(value, groups)
if replace & self._MAP_REPLACE_CUSTOM_USER_FUNCTIONS:
value = self._replace_custom_user_functions(value,groups)
if replace & self._MAP_REPLACE_CONCAT:
value = self.__replaceConcatFunction(value,groups)
return value
类CommandLineOptions，用于配置各种数据
class CommandLineOptions:
def __init__(self):
self.__options = None
parser = OptionParser(
usage="%prog [-v] [-q] [-d] [-f] [-g] [-c config_file]",
version="OSSIM (Open Source Security Information Management) " + \
"- Agent ")
parser.add_option("-v", "--verbose", dest="verbose",
action="count",
help="verbose mode, makes lot of noise")
parser.add_option("-d", "--daemon", dest="daemon", action="store_true",
help="Run agent in daemon mode")
parser.add_option("-f", "--force", dest="force", action="store_true",
help="Force startup overriding pidfile")
parser.add_option("-s", "--stats", dest="stats", type='choice', choices=['all', 'clients', 'plugins'], default=None,
help="Get stats about the agent")
parser.add_option("-c", "--config", dest="config_file", action="store",
help="read config from FILE", metavar="FILE")
(self.__options, args) = parser.parse_args()
if len(args) > 1:
parser.error("incorrect number of arguments")
if self.__options.verbose and self.__options.daemon:
parser.error("incompatible options -v -d")
def get_options(self):
return self.__options
进行数组等变量创建
def split_variables(string):
return re.findall("(?:\$([^,\s]+))+", string)
def split_sids(string, separator=','):
list = list_tmp = []
list = string.split(separator)
for sid in list:
a = sid.split('-')
if len(a) == 2:
list.remove(sid)
for i in range(int(a[0]), int(a[1]) + 1):
list_tmp.append(str(i))
list.extend(list_tmp)
return list
这篇关于OSSIM-agent源代码分析（四）的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
您可能喜欢
-
JavaMailSender是什么，怎么使用？-icode9专业技术文章分享11-15
-
JWT 用户校验学习：从入门到实践11-15
-
Nest学习：新手入门全面指南11-15
-
RestfulAPI学习：新手入门指南11-15
-
Server Component学习：入门教程与实践指南11-15
-
动态路由入门：新手必读指南11-15
-
JWT 用户校验入门：轻松掌握JWT认证基础11-15
-
Nest后端开发入门指南11-15
-
Nest后端开发入门教程11-15
-
RestfulAPI入门：新手快速上手指南11-15
-
Server Action入门：新手必读教程11-15
-
Server Component入门：新手必读教程11-15
-
抽离公共代码教程：简单易懂的入门指南11-15
-
动态路由教程：轻松入门与实战指南11-15
-
Restful API教程：入门与实践指南11-15
栏目导航
