Test Failed
Push — master ( 45848e...931d35 )
by Jan
03:00 queued 11s
created

ssg.build_sce.load_sce_and_metadata_parsed()   B

Complexity

Conditions 6

Size

Total Lines 21
Code Lines 18

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 0
CRAP Score 42

Importance

Changes 0
Metric Value
cc 6
eloc 18
nop 1
dl 0
loc 21
ccs 0
cts 18
cp 0
crap 42
rs 8.5666
c 0
b 0
f 0
1
from __future__ import absolute_import
2
from __future__ import print_function
3
4
import os
5
import os.path
6
import json
7
import sys
8
9
from .build_yaml import Rule, DocumentationNotComplete
10
from .constants import MULTI_PLATFORM_LIST
11
from .jinja import process_file_with_macros
12
from .rule_yaml import parse_prodtype
13
from .rules import get_rule_dir_id, get_rule_dir_sces, find_rule_dirs_in_paths
14
from . import utils
15
16
17
def load_sce_and_metadata(file_path, local_env_yaml):
18
    """
19
    For the given SCE audit file (file_path) under the specified environment
20
    (local_env_yaml), parse the file while expanding Jinja macros and read any
21
    metadata headers the file contains. Note that the last keyword of a
22
    specified type is the recorded one.
23
24
    Returns (audit_content, metadata).
25
    """
26
27
    raw_content = process_file_with_macros(file_path, local_env_yaml)
28
    return load_sce_and_metadata_parsed(raw_content)
29
30
31
def load_sce_and_metadata_parsed(raw_content):
32
    metadata = dict()
33
    sce_content = []
34
35
    for line in raw_content.split("\n"):
36
        found_metadata = False
37
        keywords = ['platform', 'check-import', 'check-export', 'complex-check']
38
        for keyword in keywords:
39
            if line.startswith('# ' + keyword + ' = '):
40
                # Strip off the initial comment marker
41
                _, value = line[2:].split('=', maxsplit=1)
42
                values = value.strip()
43
                if ',' in values:
44
                    values.split(',')
45
                metadata[keyword] = values
46
                found_metadata = True
47
                break
48
        if not found_metadata:
49
            sce_content.append(line)
50
51
    return "\n".join(sce_content), metadata
52
53
54
def _check_is_applicable_for_product(metadata, product):
55
    """
56
    Validates whether or not the specified check is applicable for this
57
    product. Different from build_ovals.py in that this operates directly
58
    on the parsed metadata and doesn't have to deal with matching XML
59
    elements.
60
    """
61
62
    if 'platform' not in metadata:
63
        return True
64
65
    product, product_version = utils.parse_name(product)
66
67
    multi_product = 'multi_platform_{0}'.format(product)
68
    if product in ['macos', 'ubuntu']:
69
        product_version = product_version[:2] + "." + product_version[2:]
70
71
    return ('multi_platform_all' in metadata['platform'] or
72
            (multi_product in metadata['platform'] and
73
             product in MULTI_PLATFORM_LIST) or
74
            (product in metadata['platform'] and
75
             product_version in metadata['platform']))
76
77
78
def _check_is_loaded(already_loaded, filename):
79
    # Right now this check doesn't look at metadata or anything
80
    # else. Eventually we might add versions to the entry or
81
    # something.
82
    return filename in already_loaded
83
84
85
def checks(env_yaml, yaml_path, sce_dirs, template_builder, output):
86
    """
87
    Walks the build system and builds all SCE checks (and metadata entry)
88
    into the output directory.
89
    """
90
    product = utils.required_key(env_yaml, "product")
91
    included_checks_count = 0
92
    reversed_dirs = sce_dirs[::-1]
93
    already_loaded = dict()
94
    local_env_yaml = dict()
95
    local_env_yaml.update(env_yaml)
96
97
    # We maintain the same search structure as build_ovals.py even though we
98
    # don't currently have any content under shared/checks/sce.
99
    product_dir = os.path.dirname(yaml_path)
100
    relative_guide_dir = utils.required_key(env_yaml, "benchmark_root")
101
    guide_dir = os.path.abspath(os.path.join(product_dir, relative_guide_dir))
102
    additional_content_directories = env_yaml.get("additional_content_directories", [])
103
    add_content_dirs = [
104
        os.path.abspath(os.path.join(product_dir, rd))
105
        for rd in additional_content_directories
106
    ]
107
108
    # First walk all rules under the product. These have higher priority than any
109
    # out-of-tree SCE checks.
110
    for _dir_path in find_rule_dirs_in_paths([guide_dir] + add_content_dirs):
111
        rule_id = get_rule_dir_id(_dir_path)
112
113
        rule_path = os.path.join(_dir_path, "rule.yml")
114
        try:
115
            rule = Rule.from_yaml(rule_path, env_yaml)
116
        except DocumentationNotComplete:
117
            # Happens on non-debug builds when a rule isn't yet completed. We
118
            # don't want to build the SCE check for this rule yet so skip it
119
            # and move on.
120
            continue
121
122
        prodtypes = parse_prodtype(rule.prodtype)
123
        if prodtypes and 'all' not in prodtypes and product not in prodtypes:
124
            # The prodtype exists, isn't all and doesn't contain this current
125
            # product, so we're best to skip this rule altogether.
126
            continue
127
128
        local_env_yaml['rule_id'] = rule.id_
129
        local_env_yaml['rule_title'] = rule.title
130
        local_env_yaml['products'] = prodtypes  # default is all
131
132
        for _path in get_rule_dir_sces(_dir_path, product):
133
            # To be compatible with later checks, use the rule_id (i.e., the
134
            # value of _dir) to recreate the expected filename if this OVAL
135
            # was in a rule directory. However, note that unlike
136
            # build_oval.checks(...), we have to get this script's extension
137
            # first.
138
            _, ext = os.path.splitext(_path)
139
            filename = "{0}{1}".format(rule_id, ext)
140
141
            sce_content, metadata = load_sce_and_metadata(_path, local_env_yaml)
142
            metadata['filename'] = filename
143
144
            if not _check_is_applicable_for_product(metadata, product):
145
                continue
146
            if _check_is_loaded(already_loaded, rule_id):
147
                continue
148
149
            with open(os.path.join(output, filename), 'w') as output_file:
150
                print(sce_content, file=output_file)
151
152
            included_checks_count += 1
153
            already_loaded[rule_id] = metadata
154
155
        if rule.template:
156
            langs = template_builder.get_resolved_langs_to_generate(rule)
157
            if 'sce-bash' in langs:
158
                # Here we know the specified rule has a template and this
159
                # template actually generates (bash) SCE content. We
160
                # prioritize bespoke SCE content over templated content,
161
                # however, while we add this to our metadata, we do not
162
                # bother (yet!) with generating the SCE content. This is done
163
                # at a later time by build-scripts/build_templated_content.py.
164
                if _check_is_loaded(already_loaded, rule_id):
165
                    continue
166
167
                # While we don't _write_ it, we still need to parse SCE
168
                # metadata from the templated content. Render it internally.
169
                raw_sce_content = template_builder.get_lang_for_rule(
170
                    rule_id, rule.title, rule.template, 'sce-bash')
171
172
                ext = '.sh'
173
                filename = rule_id + ext
174
175
                # Load metadata and infer correct file name.
176
                sce_content, metadata = load_sce_and_metadata_parsed(raw_sce_content)
177
                metadata['filename'] = filename
178
179
                # Skip the check if it isn't applicable for this product.
180
                if not _check_is_applicable_for_product(metadata, product):
181
                    continue
182
183
                with open(os.path.join(output, filename), 'w') as output_file:
184
                    print(sce_content, file=output_file)
185
186
                # Finally, include it in our loaded content
187
                included_checks_count += 1
188
                already_loaded[rule_id] = metadata
189
190
    # Finally take any shared SCE checks and build them as well. Note that
191
    # there's no way for shorthand generation to include them if they do NOT
192
    # align with a particular rule_id, so it is suggested that the former
193
    # method be used.
194
    for sce_dir in reversed_dirs:
195
        if not os.path.isdir(sce_dir):
196
            continue
197
198
        for filename in sorted(os.listdir(sce_dir)):
199
            rule_id, _ = os.path.splitext(filename)
200
201
            sce_content, metadata = load_sce_and_metadata(filename, env_yaml)
202
            metadata['filename'] = filename
203
204
            if not _check_is_applicable_for_product(metadata, product):
205
                continue
206
            if _check_is_loaded(already_loaded, rule_id):
207
                continue
208
209
            with open(os.path.join(output, filename), 'w') as output_file:
210
                print(sce_content, file=output_file)
211
212
            included_checks_count += 1
213
            already_loaded[rule_id] = metadata
214
215
    # Finally, write out our metadata to disk so that we can reference it in
216
    # later build stages (such as during building shorthand content).
217
    metadata_path = os.path.join(output, 'metadata.json')
218
    json.dump(already_loaded, open(metadata_path, 'w'))
219
220
    return already_loaded
221