Test Failed
Push — master ( 8bef30...0a5b5b )
by Matěj
02:23 queued 12s
created

collect_remediations.collect_remediations()   A

Complexity

Conditions 3

Size

Total Lines 15
Code Lines 14

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 0
CRAP Score 12

Importance

Changes 0
Metric Value
cc 3
eloc 14
nop 6
dl 0
loc 15
ccs 0
cts 9
cp 0
crap 12
rs 9.7
c 0
b 0
f 0
1
#!/usr/bin/env python
2
3
import sys
4
import os
5
import os.path
6
import argparse
7
8
import ssg.build_remediations as remediation
9
import ssg.build_yaml
10
import ssg.rules
11
import ssg.jinja
12
import ssg.environment
13
import ssg.utils
14
import ssg.xml
15
16
17
def parse_args():
18
    p = argparse.ArgumentParser(
19
        description="This script collects all remediation scripts, both "
20
        "static and templated, processes them with Jinja and puts them to "
21
        "the given output directory.")
22
    p.add_argument(
23
        "--build-config-yaml", required=True,
24
        help="YAML file with information about the build configuration. "
25
        "e.g.: ~/scap-security-guide/build/build_config.yml"
26
    )
27
    p.add_argument(
28
        "--product-yaml", required=True,
29
        help="YAML file with information about the product we are building. "
30
        "e.g.: ~/scap-security-guide/rhel7/product.yml"
31
    )
32
    p.add_argument(
33
        "--resolved-rules-dir", required=True,
34
        help="Directory with <rule-id>.yml resolved rule YAMLs"
35
    )
36
    p.add_argument(
37
        "--remediation-type", required=True, action="append",
38
        help="language or type of the remediations we are combining."
39
        "example: ansible")
40
    p.add_argument(
41
        "--output-dir", required=True,
42
        help="output directory where all remediations will be saved"
43
    )
44
    p.add_argument(
45
        "--fixes-from-templates-dir", required=True,
46
        help="directory from which we will collect fixes generated from "
47
        "templates")
48
49
    return p.parse_args()
50
51
52
def prepare_output_dirs(output_dir, remediation_types):
53
    output_dirs = dict()
54
    for lang in remediation_types:
55
        language_output_dir = os.path.join(output_dir, lang)
56
        if not os.path.exists(language_output_dir):
57
            os.makedirs(language_output_dir)
58
        output_dirs[lang] = language_output_dir
59
    return output_dirs
60
61
62
def find_remediation(
63
        fixes_from_templates_dir, rule_dir, lang, product, expected_file_name):
64
    language_fixes_from_templates_dir = os.path.join(
65
        fixes_from_templates_dir, lang)
66
    fix_path = None
67
    # first look for a static remediation
68
    rule_dir_remediations = remediation.get_rule_dir_remediations(
69
        rule_dir, lang, product)
70
    if len(rule_dir_remediations) > 0:
71
        # first item in the list has the highest priority
72
        fix_path = rule_dir_remediations[0]
73
    if fix_path is None:
74
        # check if we have a templated remediation instead
75
        if os.path.isdir(language_fixes_from_templates_dir):
76
            templated_fix_path = os.path.join(
77
                language_fixes_from_templates_dir, expected_file_name)
78
            if os.path.exists(templated_fix_path):
79
                fix_path = templated_fix_path
80
    return fix_path
81
82
83
def process_remediation(
84
        rule, fix_path, lang, output_dirs, expected_file_name, env_yaml):
85
    remediation_cls = remediation.REMEDIATION_TO_CLASS[lang]
86
    remediation_obj = remediation_cls(fix_path)
87
    remediation_obj.associate_rule(rule)
88
    fix = remediation.process(remediation_obj, env_yaml)
89
    if fix:
90
        output_file_path = os.path.join(output_dirs[lang], expected_file_name)
91
        remediation.write_fix_to_file(fix, output_file_path)
92
93
94
def collect_remediations(
95
        rule, langs, fixes_from_templates_dir, product, output_dirs,
96
        env_yaml):
97
    rule_dir = os.path.dirname(rule.definition_location)
98
    for lang in langs:
99
        ext = remediation.REMEDIATION_TO_EXT_MAP[lang]
100
        expected_file_name = rule.id_ + ext
101
        fix_path = find_remediation(
102
            fixes_from_templates_dir, rule_dir, lang, product,
103
            expected_file_name)
104
        if fix_path is None:
105
            # neither static nor templated remediation found
106
            continue
107
        process_remediation(
108
            rule, fix_path, lang, output_dirs, expected_file_name, env_yaml)
109
110
111
def main():
112
    args = parse_args()
113
114
    env_yaml = ssg.environment.open_environment(
115
        args.build_config_yaml, args.product_yaml)
116
117
    product = ssg.utils.required_key(env_yaml, "product")
118
    output_dirs = prepare_output_dirs(args.output_dir, args.remediation_type)
119
120
    for rule_file in os.listdir(args.resolved_rules_dir):
121
        rule_path = os.path.join(args.resolved_rules_dir, rule_file)
122
        try:
123
            rule = ssg.build_yaml.Rule.from_yaml(rule_path, env_yaml)
124
        except ssg.build_yaml.DocumentationNotComplete:
125
            # Happens on non-debug build when a rule is
126
            # "documentation-incomplete"
127
            continue
128
        collect_remediations(
129
            rule, args.remediation_type, args.fixes_from_templates_dir,
130
            product, output_dirs, env_yaml)
131
    sys.exit(0)
132
133
134
if __name__ == "__main__":
135
    main()
136