utils.build_stig_control.check_output() - Code Metrics - Inspection of "Merge pull request #7904 from Mab879/make_stig_con..." - ComplianceAsCode/content - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 972b5e...7501b0 )

by Jan

created 2021-11-29 16:54 UTC

utils.build_stig_control.check_output() A

↳ Parent: utils.build_stig_control

Complexity

Conditions

Size

Total Lines	6
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	6

Importance

Changes

Metric	Value
cc	2
eloc	5
nop	1
dl	0
loc	6
ccs	0
cts	5
cp	0
crap	6
rs	10
c	0
b	0
f	0

#!/usr/bin/env python

from __future__ import print_function

import argparse
import json
import os
import re
from pathlib import Path
import sys
import xml.etree.ElementTree as ET
import yaml

import ssg.build_yaml
import ssg.environment
import ssg.rules
import ssg.yaml


SSG_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
BUILD_OUTPUT = os.path.join(SSG_ROOT, "build", "stig_control.yml")
RULES_JSON = os.path.join(SSG_ROOT, "build", "rule_dirs.json")
BUILD_CONFIG = os.path.join(SSG_ROOT, "build", "build_config.yml")


def check_output(output: str) -> None:
    pat = re.compile(r'.*\/?[a-z_0-9]+\.yml')
    if not pat.match(output):
        sys.stderr.write('Output must only contain lowercase letters, underscores, and numbers.'
                         ' The file must also end with .yml\n')
        exit(1)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("-r", "--root", type=str, action="store", default=SSG_ROOT,
                        help="Path to SSG root directory (defaults to %s)" % SSG_ROOT)
    parser.add_argument("-o", "--output", type=str, action="store", default=BUILD_OUTPUT,
                        help=f"File to write yaml output to (defaults to {BUILD_OUTPUT}). "
                             f"Must end in '.yml' and only contain "
                             f"lowercase letters, underscores, and numbers.")
    parser.add_argument("-p", "--product", type=str, action="store", required=True,
                        help="What product to get STIGs for")
    parser.add_argument("-m", "--manual", type=str, action="store", required=True,
                        help="Path to XML XCCDF manual file to use as the source of the STIGs")
    parser.add_argument("-j", "--json", type=str, action="store", default=RULES_JSON,
                        help=f"Path to the rules_dir.json (defaults to {RULES_JSON})")
    parser.add_argument("-c", "--build-config-yaml", default=BUILD_CONFIG,
                        help="YAML file with information about the build configuration")
    parser.add_argument("-ref", "--reference", type=str, default="stigid",
                        help="Reference system to check for, defaults to stigid")
    parser.add_argument('-s', '--split', action='store_true',
                        help='Splits the each ID into its own file.')

    args = parser.parse_args()
    check_output(args.output)
    return args


def handle_rule_yaml(args, rule_id, rule_dir, guide_dir, env_yaml):
    rule_obj = {'id': rule_id, 'dir': rule_dir, 'guide': guide_dir}
    rule_file = ssg.rules.get_rule_dir_yaml(rule_dir)

    rule_yaml = ssg.build_yaml.Rule.from_yaml(rule_file, env_yaml=env_yaml)
    rule_yaml.normalize(args.product)
    rule_obj['references'] = rule_yaml.references
    return rule_obj


def get_platform_rules(args):
    rules_json_file = open(args.json, 'r')
    rules_json = json.load(rules_json_file)
    platform_rules = list()
    for rule in rules_json.values():
        if args.product in rule['products']:
            platform_rules.append(rule)
    if not rules_json_file.closed:
        rules_json_file.close()
    return platform_rules


def get_implemented_stigs(args):
    platform_rules = get_platform_rules(args)

    product_dir = os.path.join(args.root, "products", args.product)
    product_yaml_path = os.path.join(product_dir, "product.yml")
    env_yaml = ssg.environment.open_environment(args.build_config_yaml, str(product_yaml_path))

    known_rules = dict()
    for rule in platform_rules:
        try:
            rule_obj = handle_rule_yaml(args, rule['id'],
                                        rule['dir'], rule['guide'], env_yaml)
        except ssg.yaml.DocumentationNotComplete:
            sys.stderr.write('Rule %s throw DocumentationNotComplete' % rule['id'])
            # Happens on non-debug build when a rule is "documentation-incomplete"
            continue

        if args.reference in rule_obj['references'].keys():
            refs = rule_obj['references'][args.reference]
            if ',' in refs:
                refs = refs.split(',')
            else:
                refs = [refs]
            for ref in refs:
                if ref in known_rules:
                    known_rules[ref].append(rule['id'])
                else:
                    known_rules[ref] = [rule['id']]
    return known_rules


def check_files(args):
    if not os.path.exists(args.json):
        sys.stderr.write('Unable to find %s\n' % args.json)
        sys.stderr.write('Hint: run ./utils/rule_dir_json.py\n')
        exit(-1)

    if not os.path.exists(args.build_config_yaml):
        sys.stderr.write('Unable to find %s\n' % args.build_config_yaml)
        sys.stderr.write('Hint: build the project,\n')
        exit(-1)


def get_controls(known_rules, ns, root):
    controls = list()
    for group in root.findall('checklist:Group', ns):
        for stig in group.findall('checklist:Rule', ns):
            stig_id = stig.find('checklist:version', ns).text
            control = dict()
            control['id'] = stig_id
            control['levels'] = [stig.attrib['severity']]
            control['title'] = stig.find('checklist:title', ns).text
            if stig_id in known_rules.keys():
                control['rules'] = known_rules.get(stig_id)
                control['status'] = 'automated'
            else:
                control['status'] = 'pending'
            controls.append(control)
    return controls


def main():
    args = parse_args()
    check_files(args)

    ns = {'checklist': 'http://checklists.nist.gov/xccdf/1.1'}
    known_rules = get_implemented_stigs(args)
    tree = ET.parse(args.manual)
    root = tree.getroot()
    output = dict()
    output['policy'] = root.find('checklist:title', ns).text
    output['title'] = root.find('checklist:title', ns).text
    output['id'] = 'stig_%s' % args.product
    output['source'] = 'https://public.cyber.mil/stigs/downloads/'
    output['levels'] = list()
    for level in ['high', 'medium', 'low']:
        output['levels'].append({'id': level})
    controls = get_controls(known_rules, ns, root)

    if args.split:
        with open(args.output, 'w') as f:
            f.write(yaml.dump(output, sort_keys=False))
        print(f'Wrote main control file to {args.output}')
        output_path = Path(args.output)
        output_dir_name = output_path.stem
        output_root = output_path.parent
        output_dir = os.path.join(output_root, output_dir_name)
        if not os.path.exists(output_dir):
            os.mkdir(output_dir)
        for control in controls:
            out = dict()
            out['controls'] = control
            filename = f"{control['id']}.yml"
            output_filename = os.path.join(output_dir, filename)
            with open(output_filename, 'w') as f:
                f.write(yaml.dump(out, sort_keys=False))
        print(f'Wrote SRG files to {output_dir}')
        exit(0)
    else:
        output['controls'] = controls
        with open(args.output, 'w') as f:
            f.write(yaml.dump(output, sort_keys=False))
        print(f'Wrote all SRGs out to {args.output}')


if __name__ == "__main__":
    main()


1			#!/usr/bin/env python
2
3			from __future__ import print_function
4
5			import argparse
6			import json
7			import os
8			import re
9			from pathlib import Path
10			import sys
11			import xml.etree.ElementTree as ET
12			import yaml
13
14			import ssg.build_yaml
15			import ssg.environment
16			import ssg.rules
17			import ssg.yaml
18
19
20			SSG_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
21			BUILD_OUTPUT = os.path.join(SSG_ROOT, "build", "stig_control.yml")
22			RULES_JSON = os.path.join(SSG_ROOT, "build", "rule_dirs.json")
23			BUILD_CONFIG = os.path.join(SSG_ROOT, "build", "build_config.yml")
24
25
26			def check_output(output: str) -> None:
27			pat = re.compile(r'.*\/?[a-z_0-9]+\.yml')
28			if not pat.match(output):
29			sys.stderr.write('Output must only contain lowercase letters, underscores, and numbers.'
30			' The file must also end with .yml\n')
31			exit(1)
32
33
34			def parse_args() -> argparse.Namespace:
35			parser = argparse.ArgumentParser()
36			parser.add_argument("-r", "--root", type=str, action="store", default=SSG_ROOT,
37			help="Path to SSG root directory (defaults to %s)" % SSG_ROOT)
38			parser.add_argument("-o", "--output", type=str, action="store", default=BUILD_OUTPUT,
39			help=f"File to write yaml output to (defaults to {BUILD_OUTPUT}). "
40			f"Must end in '.yml' and only contain "
41			f"lowercase letters, underscores, and numbers.")
42			parser.add_argument("-p", "--product", type=str, action="store", required=True,
43			help="What product to get STIGs for")
44			parser.add_argument("-m", "--manual", type=str, action="store", required=True,
45			help="Path to XML XCCDF manual file to use as the source of the STIGs")
46			parser.add_argument("-j", "--json", type=str, action="store", default=RULES_JSON,
47			help=f"Path to the rules_dir.json (defaults to {RULES_JSON})")
48			parser.add_argument("-c", "--build-config-yaml", default=BUILD_CONFIG,
49			help="YAML file with information about the build configuration")
50			parser.add_argument("-ref", "--reference", type=str, default="stigid",
51			help="Reference system to check for, defaults to stigid")
52			parser.add_argument('-s', '--split', action='store_true',
53			help='Splits the each ID into its own file.')
54
55			args = parser.parse_args()
56			check_output(args.output)
57			return args
58
59
60			def handle_rule_yaml(args, rule_id, rule_dir, guide_dir, env_yaml):
61			rule_obj = {'id': rule_id, 'dir': rule_dir, 'guide': guide_dir}
62			rule_file = ssg.rules.get_rule_dir_yaml(rule_dir)
63
64			rule_yaml = ssg.build_yaml.Rule.from_yaml(rule_file, env_yaml=env_yaml)
65			rule_yaml.normalize(args.product)
66			rule_obj['references'] = rule_yaml.references
67			return rule_obj
68
69
70			def get_platform_rules(args):
71			rules_json_file = open(args.json, 'r')
72			rules_json = json.load(rules_json_file)
73			platform_rules = list()
74			for rule in rules_json.values():
75			if args.product in rule['products']:
76			platform_rules.append(rule)
77			if not rules_json_file.closed:
78			rules_json_file.close()
79			return platform_rules
80
81
82			def get_implemented_stigs(args):
83			platform_rules = get_platform_rules(args)
84
85			product_dir = os.path.join(args.root, "products", args.product)
86			product_yaml_path = os.path.join(product_dir, "product.yml")
87			env_yaml = ssg.environment.open_environment(args.build_config_yaml, str(product_yaml_path))
88
89			known_rules = dict()
90			for rule in platform_rules:
91			try:
92			rule_obj = handle_rule_yaml(args, rule['id'],
93			rule['dir'], rule['guide'], env_yaml)
94			except ssg.yaml.DocumentationNotComplete:
95			sys.stderr.write('Rule %s throw DocumentationNotComplete' % rule['id'])
96			# Happens on non-debug build when a rule is "documentation-incomplete"
97			continue
98
99			if args.reference in rule_obj['references'].keys():
100			refs = rule_obj['references'][args.reference]
101			if ',' in refs:
102			refs = refs.split(',')
103			else:
104			refs = [refs]
105			for ref in refs:
106			if ref in known_rules:
107			known_rules[ref].append(rule['id'])
108			else:
109			known_rules[ref] = [rule['id']]
110			return known_rules
111
112
113			def check_files(args):
114			if not os.path.exists(args.json):
115			sys.stderr.write('Unable to find %s\n' % args.json)
116			sys.stderr.write('Hint: run ./utils/rule_dir_json.py\n')
117			exit(-1)
118
119			if not os.path.exists(args.build_config_yaml):
120			sys.stderr.write('Unable to find %s\n' % args.build_config_yaml)
121			sys.stderr.write('Hint: build the project,\n')
122			exit(-1)
123
124
125			def get_controls(known_rules, ns, root):
126			controls = list()
127			for group in root.findall('checklist:Group', ns):
128			for stig in group.findall('checklist:Rule', ns):
129			stig_id = stig.find('checklist:version', ns).text
130			control = dict()
131			control['id'] = stig_id
132			control['levels'] = [stig.attrib['severity']]
133			control['title'] = stig.find('checklist:title', ns).text
134			if stig_id in known_rules.keys():
135			control['rules'] = known_rules.get(stig_id)
136			control['status'] = 'automated'
137			else:
138			control['status'] = 'pending'
139			controls.append(control)
140			return controls
141
142
143			def main():
144			args = parse_args()
145			check_files(args)
146
147			ns = {'checklist': 'http://checklists.nist.gov/xccdf/1.1'}
148			known_rules = get_implemented_stigs(args)
149			tree = ET.parse(args.manual)
150			root = tree.getroot()
151			output = dict()
152			output['policy'] = root.find('checklist:title', ns).text
153			output['title'] = root.find('checklist:title', ns).text
154			output['id'] = 'stig_%s' % args.product
155			output['source'] = 'https://public.cyber.mil/stigs/downloads/'
156			output['levels'] = list()
157			for level in ['high', 'medium', 'low']:
158			output['levels'].append({'id': level})
159			controls = get_controls(known_rules, ns, root)
160
161			if args.split:
162			with open(args.output, 'w') as f:
163			f.write(yaml.dump(output, sort_keys=False))
164			print(f'Wrote main control file to {args.output}')
165			output_path = Path(args.output)
166			output_dir_name = output_path.stem
167			output_root = output_path.parent
168			output_dir = os.path.join(output_root, output_dir_name)
169			if not os.path.exists(output_dir):
170			os.mkdir(output_dir)
171			for control in controls:
172			out = dict()
173			out['controls'] = control
174			filename = f"{control['id']}.yml"
175			output_filename = os.path.join(output_dir, filename)
176			with open(output_filename, 'w') as f:
177			f.write(yaml.dump(out, sort_keys=False))
178			print(f'Wrote SRG files to {output_dir}')
179			exit(0)
180			else:
181			output['controls'] = controls
182			with open(args.output, 'w') as f:
183			f.write(yaml.dump(output, sort_keys=False))
184			print(f'Wrote all SRGs out to {args.output}')
185
186
187			if __name__ == "__main__":
188			main()
189

ComplianceAsCode / content

Push — master ( 972b5e...7501b0 )

utils.build_stig_control.check_output() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like