Passed
Pull Request — master (#4216)
by Matěj
02:33 queued 10s
created

combine_remediations.collect_fixes()   B

Complexity

Conditions 6

Size

Total Lines 21
Code Lines 14

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 0
CRAP Score 42

Importance

Changes 0
Metric Value
cc 6
eloc 14
nop 4
dl 0
loc 21
ccs 0
cts 14
cp 0
crap 42
rs 8.6666
c 0
b 0
f 0
1
#!/usr/bin/env python2
2
3
import sys
4
import os
5
import os.path
6
import re
7
import errno
8
import argparse
9
import codecs
10
11
import ssg.build_remediations as remediation
12
import ssg.rules
13
import ssg.jinja
14
import ssg.yaml
15
import ssg.utils
16
import ssg.xml
17
18
19
def parse_args():
20
    p = argparse.ArgumentParser()
21
    p.add_argument(
22
        "--build-config-yaml", required=True,
23
        help="YAML file with information about the build configuration. "
24
        "e.g.: ~/scap-security-guide/build/build_config.yml"
25
    )
26
    p.add_argument(
27
        "--product-yaml", required=True,
28
        help="YAML file with information about the product we are building. "
29
        "e.g.: ~/scap-security-guide/rhel7/product.yml"
30
    )
31
    p.add_argument(
32
        "--resolved-rules-dir", required=True,
33
        help="Directory with <rule-id>.yml resolved rule YAMLs"
34
    )
35
    p.add_argument("--remediation-type", required=True,
36
                   help="language or type of the remediations we are combining."
37
                   "example: ansible")
38
    p.add_argument(
39
        "--output-dir", required=True,
40
        help="output directory where all remediations will be saved"
41
    )
42
    p.add_argument("fix_dirs", metavar="FIX_DIR", nargs="+",
43
                   help="directory(ies) from which we will collect "
44
                   "remediations to combine.")
45
46
    return p.parse_args()
47
48
49
def main():
50
    args = parse_args()
51
52
    env_yaml = ssg.yaml.open_environment(
53
        args.build_config_yaml, args.product_yaml)
54
55
    product = ssg.utils.required_key(env_yaml, "product")
56
57
    product_dir = os.path.dirname(args.product_yaml)
58
    relative_guide_dir = ssg.utils.required_key(env_yaml, "benchmark_root")
59
    guide_dir = os.path.abspath(os.path.join(product_dir, relative_guide_dir))
60
61
    # As fixes is continually updated, the last seen fix that is applicable for a
62
    # given fix_name is chosen to replace newer fix_names
63
    remediation_cls = remediation.REMEDIATION_TO_CLASS[args.remediation_type]
64
65
    rule_id_to_remediation_map = collect_fixes(
66
        product, guide_dir, args.fix_dirs, args.remediation_type)
67
68
    fixes = dict()
69
    for rule_id, fix_path in rule_id_to_remediation_map.items():
70
        remediation_obj = remediation_cls(fix_path)
71
        rule_path = os.path.join(args.resolved_rules_dir, rule_id + ".yml")
72
        remediation_obj.load_rule_from(rule_path)
73
        # Fixes gets updated with the contents of the fix, if it is applicable
74
        remediation.process(remediation_obj, env_yaml, fixes, rule_id)
75
76
    remediation.write_fixes_to_dir(fixes, args.remediation_type,
77
                                   args.output_dir)
78
79
    sys.stderr.write("Collected %d %s remediations.\n" % (len(fixes), args.remediation_type))
80
81
    sys.exit(0)
82
83
84
def collect_fixes(product, guide_dir, fix_dirs, remediation_type):
85
    # path -> remediation
86
    # rule ID -> assoc rule
87
    rule_id_to_remediation_map = dict()
88
    for fixdir in fix_dirs:
89
        if os.path.isdir(fixdir):
90
            for filename in os.listdir(fixdir):
91
                file_path = os.path.join(fixdir, filename)
92
                rule_id, _ = os.path.splitext(filename)
93
                rule_id_to_remediation_map[rule_id] = file_path
94
95
    # Walk the guide last, looking for rule folders as they have the highest priority
96
    for _dir_path in ssg.rules.find_rule_dirs(guide_dir):
97
        rule_id = ssg.rules.get_rule_dir_id(_dir_path)
98
99
        contents = remediation.get_rule_dir_remediations(_dir_path, remediation_type, product)
100
        for _path in reversed(contents):
101
            # To be compatible with the later checks, use the rule_id
102
            # (i.e., the value of _dir) to create the fix_name
103
            rule_id_to_remediation_map[rule_id] = _path
104
    return rule_id_to_remediation_map
105
106
107
if __name__ == "__main__":
108
    main()
109