Updated many commands to work (better) with OVAL files. Import will now
import plain OVAL files to /var/lib/secstate/oval and OVAL files can be
selected/deselected.
Also fixed a couple issues with the audit command when not using a
profile.
---
src/bin/secstate | 6 +-
src/secstate/main.py | 289 +++++++++++++++++++++++++++++---------------------
src/secstate/util.py | 36 +++---
3 files changed, 191 insertions(+), 140 deletions(-)
diff --git a/src/bin/secstate b/src/bin/secstate
index e786fd7..3dce12c 100644
--- a/src/bin/secstate
+++ b/src/bin/secstate
@@ -111,8 +111,8 @@ def import_content(arguments):
(options, args) = parser.parse_args(arguments)
for arg in args:
- (benchmark, def_model) = sec_instance.import_content(arg,
sec_instance.config.get('secstate', 'data_dir'), options.cpe,
options.puppet)
- if (benchmark == None) or (def_model == None):
+ (benchmark, def_model) = sec_instance.import_content(arg, options.cpe,
options.puppet)
+ if (benchmark == None) and (def_model == None):
return -1
oscap.oval_definition_model_free(def_model)
oscap.xccdf_benchmark_free(benchmark)
@@ -152,7 +152,7 @@ def audit(arguments):
default=sec_instance.config.get('secstate',
'oval_interpreter'),
help="Specifies the interpreter to use when auditing the
system")
parser.add_option('-p', '--profile', action='store',
type='string', dest='profile',
- default="Custom", help="Specifies the profile to use
when auditing the system")
+ default=None, help="Specifies the profile to use when auditing
the system")
parser.add_option('-r', '--results', action='store',
dest='results', default=False,
help="Export results to the specified file")
parser.add_option('-v', '--verbose', action='store_true',
dest='verbose', default=False,
diff --git a/src/secstate/main.py b/src/secstate/main.py
index 5642e8d..bb6e6d6 100644
--- a/src/secstate/main.py
+++ b/src/secstate/main.py
@@ -40,6 +40,7 @@ class Secstate:
def __init__(self, conf_file):
self.setConfigFile(conf_file)
self.content = self.get_content_dict()
+ self.content_configs = self.get_content_configs()
self.log = self.getLogger()
self.benchmark_dir = self.config.get('secstate',
'benchmark_dir')
@@ -92,6 +93,15 @@ class Secstate:
fp.close()
return content
+
+ def get_content_configs(self):
+ configs = {}
+ conf_dir = self.config.get('secstate', 'conf_dir')
+ for conf_file in os.listdir(conf_dir):
+ id = os.path.splitext(conf_file)[0]
+ configs[id] = os.path.join(conf_dir, conf_file)
+
+ return configs
def combine_def_models(self, target, source):
"""
@@ -155,24 +165,27 @@ class Secstate:
def_model = oscap.oval_definition_model_import(oval_file)
if def_model == None:
self.log.error("Error importing OVAL content: %('file')s" %
{'file':oval_file})
- return None
+ return (None, None)
if not oscap.oval_definition_model_is_valid(def_model):
self.log.error("Definition model is invalid")
oscap.oval_definition_model_free(def_model)
- return None
+ return (None, None)
if store_path:
shutil.copy(oval_file, store_path)
config = ConfigParser.ConfigParser()
- config.add_section('selected')
- config.set('selected', 'selected', True)
- conf_file = open(self.config.get('secstate', 'conf_dir') +
os.path.splitext(os.path.basename(oval_file)), 'w')
+ config.optionxform = str
+ id = os.path.splitext(os.path.basename(oval_file))[0]
+ config.add_section(id)
+ config.set(id, 'selected', True)
+ config.set(id, 'file', os.path.join(store_path, oval_file))
+ conf_file = open(os.path.join(self.config.get('secstate',
'conf_dir'), id + ".cfg"), 'w')
config.write(conf_file)
conf_file.close()
- return None
+ return (None, None)
- return def_model
+ return (None, def_model)
def import_benchmark(self, benchmark_file, oval_path="", store_path=None):
@@ -229,7 +242,7 @@ class Secstate:
shutil.copy(benchmark_file, directory)
config = ConfigParser.ConfigParser()
config.add_section(id)
- config.set(id, 'file', os.path.basename(benchmark_file))
+ config.set(id, 'file', os.path.join(directory,
os.path.basename(benchmark_file)))
config.set(id, 'selected', True)
conf_file = open(os.path.join(self.config.get('secstate',
'conf_dir'), id + ".cfg"), 'w')
config.write(conf_file)
@@ -309,7 +322,7 @@ class Secstate:
return (benchmark, def_model)
- def import_content(self, content, store_path=None, cpe=False, puppet=False,
changes=True):
+ def import_content(self, content, cpe=False, puppet=False, changes=True, save=True):
"""
Function: Validates XCCDF/OVAL content and optionally saves it to the data
store
Input: File containing content
@@ -317,12 +330,13 @@ class Secstate:
"""
xccdf = False
oval = False
+ store_path = None
if self.content.has_key(content):
- (benchmark, oval) = self.import_benchmark(os.path.join(self.benchmark_dir,
content, self.content[content]), oval_path=os.path.join(self.benchmark_dir, content))
+ (benchmark, oval) = self.import_content(os.path.join(self.benchmark_dir,
content, self.content[content]), save=False)
if changes and (benchmark != None):
#benchmark = apply_changes(benchmark, os.path.join(self.benchmark_dir,
content, str(content + ".cfg")))
- benchmark = apply_changes_profile(benchmark,
os.path.join(self.config.get('secstate', 'conf_dir'), content +
".cfg"))
+ benchmark = apply_changes_profile(benchmark,
self.content_configs[content])
return (benchmark, oval)
@@ -330,17 +344,21 @@ class Secstate:
if file_type[0] == "text/xml":
if is_benchmark(content):
xccdf = True
+ if save:
+ store_path = self.config.get('secstate',
'benchmark_dir')
else:
oval = True
+ if save:
+ store_path = self.config.get('secstate', 'oval_dir')
if oval:
- return self.import_oval(content, self.config.get('secstate',
'oval_dir'))
+ return self.import_oval(content, store_path)
if xccdf:
return self.import_benchmark(content, store_path=store_path,
oval_path=os.path.dirname(content))
else:
- return self.import_zipped_content(content, file_type, store_path, puppet)
+ return self.import_zipped_content(content, file_type, puppet)
def export(self, benchmark_id, new_file, original=False):
if not self.content.has_key(benchmark_id):
@@ -403,7 +421,7 @@ class Secstate:
sel_dict = {'selected':selected, 'message':message}
bench_cfg = ConfigParser.ConfigParser()
bench_cfg.optionxform = str
- conf_path = os.path.join(self.config.get('secstate', 'conf_dir'),
benchmark_id + ".cfg")
+ conf_path = self.content_configs[benchmark_id]
if os.path.isfile(conf_path):
try:
fp = open(conf_path)
@@ -421,45 +439,51 @@ class Secstate:
return False
(benchmark, oval) = self.import_content(benchmark_id)
- oscap.oval_definition_model_free(oval)
if benchmark == None:
- self.log.error("Error opening benchmark: %(file)s" %
{'file':benchmark_id})
- return False
-
- if item_id == benchmark_id:
- bench_cfg.set(benchmark_id, 'selected', selected)
- self.log.debug("Setting %(id)s to %(val)s" %
{'id':benchmark_id,
- 'val':selected})
- item = oscap.xccdf_benchmark_to_item(benchmark)
+ if oval == None:
+ self.log.error("Error opening benchmark: %(file)s" %
{'file':benchmark_id})
+ return False
+ else:
+ bench_cfg.set(benchmark_id, 'selected', selected)
+ self.log.debug("Set Oval file %(file)s to %(sel)s" %
{'file':benchmark_id,
+
'sel':selected})
else:
- item = oscap.xccdf_benchmark_get_item(benchmark, item_id)
-
- if item == None:
- self.log.error("Benchmark %(bench_id)s does not contain
%(item_id)s" % {'bench_id':benchmark_id,
-
'item_id':item_id})
- return False
+ oscap.oval_definition_model_free(oval)
- item_type = oscap.xccdf_item_get_type(item)
- if item_type == oscap.XCCDF_PROFILE:
- bench_cfg.set(benchmark_id, 'profile', item_id)
- self.log.debug("Setting active profile to %(id)s" %
{'id':item_id})
- else:
- if bench_cfg.has_option(benchmark_id, 'profile'):
- active_profile = bench_cfg.get(benchmark_id, 'profile')
- if active_profile != "Custom":
- bench_cfg.set(benchmark_id, 'extends', active_profile)
+ if item_id == benchmark_id:
+ bench_cfg.set(benchmark_id, 'selected', selected)
+ self.log.debug("Setting %(id)s to %(val)s" %
{'id':benchmark_id,
+
'val':selected})
+ item = oscap.xccdf_benchmark_to_item(benchmark)
+ else:
+ item = oscap.xccdf_benchmark_get_item(benchmark, item_id)
- if item_type != oscap.XCCDF_BENCHMARK:
- bench_cfg.set('selections', item_id, json.dumps(sel_dict))
- self.log.debug("Setting %(id)s to %(val)s" %
{'id':item_id,
- 'val':selected})
+ if item == None:
+ self.log.error("Benchmark %(bench_id)s does not contain
%(item_id)s" % {'bench_id':benchmark_id,
+
'item_id':item_id})
+ return False
- if recurse:
- if (item_type == oscap.XCCDF_GROUP) or (item_type ==
oscap.XCCDF_BENCHMARK):
- for sub in xccdf_get_items(benchmark, oscap.XCCDF_ITEM,
oscap.xccdf_item_get_content(item)):
- bench_cfg.set('selections', oscap.xccdf_item_get_id(sub),
json.dumps(sel_dict))
- self.log.debug("Setting %(id)s to %(val)s" %
{'id':oscap.xccdf_item_get_id(sub),
-
'val':selected})
+ item_type = oscap.xccdf_item_get_type(item)
+ if item_type == oscap.XCCDF_PROFILE:
+ bench_cfg.set(benchmark_id, 'profile', item_id)
+ self.log.debug("Setting active profile to %(id)s" %
{'id':item_id})
+ else:
+ if bench_cfg.has_option(benchmark_id, 'profile'):
+ active_profile = bench_cfg.get(benchmark_id, 'profile')
+ if active_profile != "Custom":
+ bench_cfg.set(benchmark_id, 'extends', active_profile)
+
+ if item_type != oscap.XCCDF_BENCHMARK:
+ bench_cfg.set('selections', item_id, json.dumps(sel_dict))
+ self.log.debug("Setting %(id)s to %(val)s" %
{'id':item_id,
+
'val':selected})
+
+ if recurse:
+ if (item_type == oscap.XCCDF_GROUP) or (item_type ==
oscap.XCCDF_BENCHMARK):
+ for sub in xccdf_get_items(benchmark, oscap.XCCDF_ITEM,
oscap.xccdf_item_get_content(item)):
+ bench_cfg.set('selections',
oscap.xccdf_item_get_id(sub), json.dumps(sel_dict))
+ self.log.debug("Setting %(id)s to %(val)s" %
{'id':oscap.xccdf_item_get_id(sub),
+
'val':selected})
fp = open(conf_path, 'w')
bench_cfg.write(fp)
@@ -476,32 +500,47 @@ class Secstate:
"""
def_model = None
benchmark = None
+ config = None
for arg in args:
(benchmark, def_model) = self.import_content(arg)
- if (benchmark == None) or (def_model == None):
+ if (benchmark == None) and (def_model == None):
self.log.error("Error importing benchmark: %(bench)s" %
{'bench':arg})
return False
- if interpreter == "openscap":
- sess = oscap.oval_agent_new_session(def_model)
-
+ else:
if self.content.has_key(arg):
- if profile == None:
- config = ConfigParser.ConfigParser()
- fp = open(self.content[arg][1])
- config.readfp(fp)
- profile = config.get(arg, 'profile')
- fp.close()
+ config = ConfigParser.ConfigParser()
+ fp = open(self.content_configs[arg])
+ config.readfp(fp)
+ fp.close()
+ if not all and (not config.getboolean(arg, 'selected')) and
(len(args) > 1):
+ print "Skipping %(id)s" % {'id':arg}
+ ret = True
+ continue
+
+ if interpreter == "openscap":
+ sess = oscap.oval_agent_new_session(def_model)
+
+ if benchmark != None:
+ # Set profile to default found in config file
+ if (profile == None) and (config != None):
+ if config.has_option(arg, 'profile'):
+ profile = config.get(arg, 'profile')
+
+ if profile!= None:
+ if oscap.xccdf_benchmark_get_item(benchmark, profile) ==
None:
+ self.log.error("Profile %(prof)s does not
exist." % {'prof':profile})
+ return False
+
+ ret = (evaluate_xccdf(benchmark, def_model, arg, sess,
s_profile=profile, verbose=verbose) == 0)
- if oscap.xccdf_benchmark_get_item(benchmark, profile) == None:
- self.log.error("Profile %(prof)s does not exist." %
{'prof':profile})
- return False
- ret = (evaluate_xccdf(benchmark, def_model, arg, sess, s_profile=profile,
verbose=verbose, all=all) == 0)
+ elif def_model != None:
+ ret = (evaluate_oval(sess, f_results, verbose) == 0)
- if ret == False:
- self.log.error("Error auditing %(arg)s" % {'arg':arg})
- return ret
+ if ret == False:
+ self.log.error("Error auditing %(arg)s" %
{'arg':arg})
+ return ret
def search(self, search_string, verbose=False):
"""
@@ -621,63 +660,74 @@ class Secstate:
return True
def sublist(self, benchmark, bench_cfg, def_model, arg, recurse, show_all, tabs=0):
- benchmark_id = oscap.xccdf_benchmark_get_id(benchmark)
tabstr = "\t" * tabs
selected = ""
- is_selected = False
- item = None
- if arg == benchmark_id:
- item = oscap.xccdf_benchmark_to_item(benchmark)
- is_selected = bench_cfg.getboolean(arg, 'selected')
- else:
- item = oscap.xccdf_benchmark_get_item(benchmark, arg)
- if item == None:
- defn = oscap.oval_definition_model_get_definition(def_model, arg)
- if defn == None:
- self.log.error("No item '%(id)s'" %
{'id':arg})
- return
- print "%(indent)sDefinition - ID: %(id)s, Title:
'%(title)s'" % {'indent':tabstr, 'id':arg,
-
'title':oscap.oval_definition_get_title(defn)}
- is_selected = oscap.xccdf_item_get_selected(item)
-
- selects = {}
- if bench_cfg.has_option(benchmark_id, 'profile'):
- prof = oscap.xccdf_benchmark_get_item(benchmark, bench_cfg.get(benchmark_id,
"profile"))
- if prof == None:
- self.log.error("Error loading profile %(prof)s" %
{'prof':profile})
- return False
- prof = oscap.xccdf_item_to_profile(prof)
- select_it = oscap.xccdf_profile_get_selects(prof)
- for select in xccdf_select_generator(select_it):
- selects[oscap.xccdf_select_get_item(select)] =
oscap.xccdf_select_get_selected(select)
-
- try:
- is_selected = selects[oscap.xccdf_item_get_id(item)]
- except KeyError, e:
- pass
-
- titles = oscap.xccdf_item_get_title(item)
- for title in oscap_text_generator(titles):
- if show_all:
- if is_selected:
- selected = "[X]"
+ if benchmark == None:
+ if self.content.has_key(arg):
+ if bench_cfg.getboolean(arg, 'selected'):
+ if show_all:
+ selected = "[X]"
else:
selected = "[ ]"
-
- if not is_selected:
- if not recurse or (tabs == 0):
- selected = "[ ]"
+ print "%(indent)s%(sel)sOVAL File - ID: %(id)s" %
{'indent':tabstr, 'sel':selected, 'id':arg}
+ else:
+ defn = oscap.oval_definition_model_get_definition(def_model, arg)
+ if defn != None:
+ print "%(indent)sDefinition - ID: %(id)s, Title:
'%(title)s'" % {'indent':tabstr, 'id':arg,
+
'title':oscap.oval_definition_get_title(defn)}
+ else:
+ benchmark_id = oscap.xccdf_benchmark_get_id(benchmark)
+ is_selected = False
+ item = None
+ if arg == benchmark_id:
+ item = oscap.xccdf_benchmark_to_item(benchmark)
+ is_selected = bench_cfg.getboolean(arg, 'selected')
+ else:
+ item = oscap.xccdf_benchmark_get_item(benchmark, arg)
+ if item == None:
+ self.sublist(None, bench_cfg, def_model, arg, recurse, show_all,
tabs)
- print "%(indent)s%(sel)s%(type)s - ID: %(id)s, Title:
'%(title)s'" % {'indent':tabstr, 'sel':selected,
-
'type':item_get_type_str(item), 'id':arg,
-
'title':oscap.oscap_text_get_text(title)}
- if recurse:
- type = oscap.xccdf_item_get_type(item)
- if (type == oscap.XCCDF_GROUP) or (type == oscap.XCCDF_BENCHMARK):
- content = oscap.xccdf_item_get_content(item)
- for sub in xccdf_item_generator(content):
- self.sublist(benchmark, bench_cfg, def_model,
oscap.xccdf_item_get_id(sub), recurse, show_all, tabs+1)
+ is_selected = oscap.xccdf_item_get_selected(item)
+
+ selects = {}
+ if bench_cfg.has_option(benchmark_id, 'profile'):
+ prof = oscap.xccdf_benchmark_get_item(benchmark,
bench_cfg.get(benchmark_id, "profile"))
+ if prof == None:
+ self.log.error("Error loading profile %(prof)s" %
{'prof':profile})
+ return False
+ prof = oscap.xccdf_item_to_profile(prof)
+
+ select_it = oscap.xccdf_profile_get_selects(prof)
+ for select in xccdf_select_generator(select_it):
+ selects[oscap.xccdf_select_get_item(select)] =
oscap.xccdf_select_get_selected(select)
+
+ try:
+ is_selected = selects[oscap.xccdf_item_get_id(item)]
+ except KeyError, e:
+ pass
+
+ titles = oscap.xccdf_item_get_title(item)
+ for title in oscap_text_generator(titles):
+ if show_all:
+ if is_selected:
+ selected = "[X]"
+ else:
+ selected = "[ ]"
+
+ if not is_selected:
+ if not recurse or (tabs == 0):
+ selected = "[ ]"
+
+ print "%(indent)s%(sel)s%(type)s - ID: %(id)s, Title:
'%(title)s'" % {'indent':tabstr, 'sel':selected,
+
'type':item_get_type_str(item), 'id':arg,
+
'title':oscap.oscap_text_get_text(title)}
+ if recurse:
+ type = oscap.xccdf_item_get_type(item)
+ if (type == oscap.XCCDF_GROUP) or (type == oscap.XCCDF_BENCHMARK):
+ content = oscap.xccdf_item_get_content(item)
+ for sub in xccdf_item_generator(content):
+ self.sublist(benchmark, bench_cfg, def_model,
oscap.xccdf_item_get_id(sub), recurse, show_all, tabs+1)
def list_content(self, arg=None, recurse=False, show_all=False):
@@ -685,11 +735,12 @@ class Secstate:
for key in self.content:
(benchmark, def_model) = self.import_content(key)
if benchmark == None:
- self.log.error("Error loading benchmark: %(id)s" %
{'id':key})
- return False
+ if def_model == None:
+ self.log.error("Error loading benchmark: %(id)s" %
{'id':key})
+ return False
config = ConfigParser.ConfigParser()
- fp = open(os.path.join(self.config.get('secstate',
'conf_dir'), key + ".cfg"))
+ fp = open(self.content_configs[key])
config.readfp(fp)
fp.close()
diff --git a/src/secstate/util.py b/src/secstate/util.py
index d1a16df..aaec22a 100644
--- a/src/secstate/util.py
+++ b/src/secstate/util.py
@@ -26,6 +26,7 @@ import time
import re
import ConfigParser
import json
+import tempfile
import openscap as oscap
@@ -182,7 +183,7 @@ def output_callback(id, result, usr):
print "Rule '%(id)s' result: %(res)s" % {'id':id,
'res':oscap.xccdf_test_result_type_get_text(result)}
return 0
-def evaluate_xccdf(benchmark, def_model, url_XCCDF, sess, s_profile=None, all=False,
verbose=False):
+def evaluate_xccdf(benchmark, def_model, url_XCCDF, sess, s_profile=None,
verbose=False):
policy = None
policy_model = oscap.xccdf_policy_model_new(benchmark)
@@ -214,10 +215,7 @@ def evaluate_xccdf(benchmark, def_model, url_XCCDF, sess,
s_profile=None, all=Fa
oscap.oval_agent_cb_data_set_callback_py(usr, None, None)
oscap.oval_agent_cb_data_set_usr(usr, policy_model)
- if all:
- callback_func = oscap.oval_agent_eval_rule_py
- else:
- callback_func = xccdf_callback
+ callback_func = oscap.oval_agent_eval_rule_py
oscap.xccdf_policy_model_register_output_callback_py(policy_model, output_callback,
res_dict)
oscap.xccdf_policy_model_register_callback_py(policy_model,
"http://oval.mitre.org/XMLSchema/oval-definitions-5",
@@ -261,15 +259,17 @@ def evaluate_xccdf(benchmark, def_model, url_XCCDF, sess,
s_profile=None, all=Fa
def oval_callback(id, result, usr):
if result == oscap.OVAL_RESULT_TRUE:
- usr['True'] += 1
+ usr['true'] += 1
elif result == oscap.OVAL_RESULT_FALSE:
- usr['False'] += 1
+ usr['false'] += 1
+ elif result == oscap.OVAL_RESULT_ERROR:
+ usr['err'] += 1
elif result == oscap.OVAL_RESULT_UNKNOWN:
- usr['Unknown'] += 1
+ usr['unknown'] += 1
elif result == oscap.OVAL_RESULT_NOT_EVALUATED:
- usr['Not Evaluated'] += 1
+ usr['neval'] += 1
elif result == oscap.OVAL_RESULT_NOT_APPLICABLE:
- usr['Not Applicable'] += 1
+ usr['na'] += 1
if usr['verbose']:
print "Definintion '%(id)s' result: %(res)s" %
{'id':id, 'res':oscap.oval_result_get_text(result)}
@@ -308,12 +308,11 @@ def output_oval_results(res_model, res_directives, xml, html):
return True
-def evaluate_oval(def_model, sess, f_xml, f_html, verbose=True):
- res_model = oscap.oval_agent_get_results_model(sess)
+def evaluate_oval(sess, f_results, verbose=False):
usr = {'false':0,
'true':0,
- 'invalid':0,
+ 'err':0,
'unknown':0,
'neval':0,
'na':0,
@@ -327,17 +326,19 @@ def evaluate_oval(def_model, sess, f_xml, f_html, verbose=True):
if oscap.oscap_err():
sys.stderr.write("Error: (%(code)d) %(desc)s" %
{'code':oscap.oscap_err_code(),
'desc':oscap.oscap_err_desc()})
- return False
+ return None
+
+ res_model = oscap.oval_agent_get_results_model(sess)
print "--Results--\n" \
"True:\t\t%(true)s\n" \
"False:\t\t%(false)s\n" \
- "Invalid:\t\t%(invalid)s\n" \
+ "Error:\t\t%(err)s\n" \
"Unknown:\t\t%(unknown)s\n" \
"Not Evaluated:\t%(neval)s\n" \
"Not Applicable:\t%(na)s\n" % usr
- if f_xml or f_html:
+ if f_results:
res_direct = oscap.oval_result_directives_new(res_model)
oscap.oval_result_directives_set_reported(res_direct, oscap.OVAL_RESULT_TRUE |
oscap.OVAL_RESULT_FALSE |
@@ -351,8 +352,7 @@ def evaluate_oval(def_model, sess, f_xml, f_html, verbose=True):
output_oval_results(res_model, res_direct, f_xml, f_html)
oscap.oval_result_directives_free(res_direct)
- oscap.oval_agent_destroy_session(sess)
- return ret
+ return res_model
def is_benchmark(benchmark):
tree = xml.dom.minidom.parse(benchmark)
--
1.7.1.1