Test Failed
Push — master ( 5d561a...0b14cb )
by Matěj
01:45 queued 15s
created

ssg.products   A

Complexity

Total Complexity 27

Size/Duplication

Total Lines 155
Duplicated Lines 0 %

Test Coverage

Coverage 84.52%

Importance

Changes 0
Metric Value
eloc 102
dl 0
loc 155
ccs 71
cts 84
cp 0.8452
rs 10
c 0
b 0
f 0
wmc 27

7 Functions

Rating   Name   Duplication   Size   Complexity  
A _validate_product_oval_feed_url() 0 10 3
A get_all() 0 24 3
A load_product_yaml() 0 27 1
A product_yaml_path() 0 2 1
F _get_implied_properties() 0 41 15
A get_profiles_directory() 0 5 2
A get_profile_files_from_root() 0 8 2
1 2
from __future__ import absolute_import
2 2
from __future__ import print_function
3
4 2
import os
5 2
from collections import namedtuple
6 2
from glob import glob
7
8 2
from .build_cpe import ProductCPEs
9 2
from .constants import (DEFAULT_PRODUCT, product_directories,
10
                        DEFAULT_GID_MIN,
11
                        DEFAULT_UID_MIN,
12
                        DEFAULT_GRUB2_BOOT_PATH,
13
                        DEFAULT_GRUB2_UEFI_BOOT_PATH,
14
                        DEFAULT_DCONF_GDM_DIR,
15
                        DEFAULT_AIDE_CONF_PATH,
16
                        DEFAULT_AIDE_BIN_PATH,
17
                        DEFAULT_SSH_DISTRIBUTED_CONFIG,
18
                        PKG_MANAGER_TO_SYSTEM,
19
                        PKG_MANAGER_TO_CONFIG_FILE,
20
                        XCCDF_PLATFORM_TO_PACKAGE,
21
                        SSG_REF_URIS)
22 2
from .utils import merge_dicts, required_key
23 2
from .yaml import open_raw
24
25
26 2
def _validate_product_oval_feed_url(contents):
27 2
    if "oval_feed_url" not in contents:
28 2
        return
29 2
    url = contents["oval_feed_url"]
30 2
    if not url.startswith("https"):
31
        msg = (
32
            "OVAL feed of product '{product}' is not available through an encrypted channel: {url}"
33
            .format(product=contents["product"], url=url)
34
        )
35
        raise ValueError(msg)
36
37
38 2
def _get_implied_properties(existing_properties):
39 2
    result = existing_properties.copy()
40 2
    if "pkg_manager" in existing_properties:
41 2
        pkg_manager = existing_properties["pkg_manager"]
42 2
        if "pkg_system" not in existing_properties:
43 2
            result["pkg_system"] = PKG_MANAGER_TO_SYSTEM[pkg_manager]
44 2
        if "pkg_manager_config_file" not in existing_properties:
45 2
            if pkg_manager in PKG_MANAGER_TO_CONFIG_FILE:
46 2
                result["pkg_manager_config_file"] = PKG_MANAGER_TO_CONFIG_FILE[pkg_manager]
47
48 2
    if "gid_min" not in existing_properties:
49 2
        result["gid_min"] = DEFAULT_GID_MIN
50
51 2
    if "uid_min" not in existing_properties:
52 2
        result["uid_min"] = DEFAULT_UID_MIN
53
54 2
    if "auid" not in existing_properties:
55 2
        result["auid"] = existing_properties.get("uid_min", DEFAULT_UID_MIN)
56
57 2
    if "grub2_boot_path" not in existing_properties:
58 2
        result["grub2_boot_path"] = DEFAULT_GRUB2_BOOT_PATH
59
60 2
    if "grub2_uefi_boot_path" not in existing_properties:
61 2
        result["grub2_uefi_boot_path"] = DEFAULT_GRUB2_UEFI_BOOT_PATH
62
63 2
    if "dconf_gdm_dir" not in existing_properties:
64 2
        result["dconf_gdm_dir"] = DEFAULT_DCONF_GDM_DIR
65
66 2
    if "aide_conf_path" not in existing_properties:
67 2
        result["aide_conf_path"] = DEFAULT_AIDE_CONF_PATH
68
69 2
    if "aide_bin_path" not in existing_properties:
70 2
        result["aide_bin_path"] = DEFAULT_AIDE_BIN_PATH
71
72 2
    if "sshd_distributed_config" not in existing_properties:
73 2
        result["sshd_distributed_config"] = DEFAULT_SSH_DISTRIBUTED_CONFIG
74
75 2
    if "product" not in existing_properties:
76 2
        result["product"] = DEFAULT_PRODUCT
77
78 2
    return result
79
80
81 2
def product_yaml_path(ssg_root, product):
82
    return os.path.join(ssg_root, "products", product, "product.yml")
83
84
85 2
def load_product_yaml(product_yaml_path):
86
    """
87
    Reads a product data from disk and returns it.
88
    The returned product dictionary also contains derived useful information.
89
    """
90
91 2
    product_yaml = open_raw(product_yaml_path)
92 2
    _validate_product_oval_feed_url(product_yaml)
93
94
    # The product directory is necessary to get absolute paths to benchmark, profile and
95
    # cpe directories, which are all relative to the product directory
96 2
    product_yaml["product_dir"] = os.path.dirname(product_yaml_path)
97
98 2
    platform_package_overrides = product_yaml.get("platform_package_overrides", {})
99
    # Merge common platform package mappings, while keeping product specific mappings
100 2
    product_yaml["platform_package_overrides"] = merge_dicts(XCCDF_PLATFORM_TO_PACKAGE,
101
                                                             platform_package_overrides)
102 2
    product_yaml.update(_get_implied_properties(product_yaml))
103
104
    # The product_yaml should be aware of the ProductCPEs
105 2
    product_yaml["product_cpes"] = ProductCPEs(product_yaml)
106
107 2
    reference_uris = product_yaml.get("reference_uris", {})
108 2
    product_yaml["reference_uris"] = merge_dicts(SSG_REF_URIS,
109
                                                 reference_uris)
110
111 2
    return product_yaml
112
113
114 2
def get_all(ssg_root):
115
    """
116
    Analyzes all products in the SSG root and sorts them into two categories:
117
    those which use linux_os and those which use their own directory. Returns
118
    a namedtuple of sets, (linux, other).
119
    """
120
121 2
    linux_products = set()
122 2
    other_products = set()
123
124 2
    for product in product_directories:
125 2
        product_yaml_path = os.path.join(ssg_root, "products", product, "product.yml")
126 2
        product_yaml = load_product_yaml(product_yaml_path)
127
128 2
        guide_dir = os.path.join(product_yaml["product_dir"], product_yaml['benchmark_root'])
129 2
        guide_dir = os.path.abspath(guide_dir)
130
131 2
        if 'linux_os' in guide_dir:
132 2
            linux_products.add(product)
133
        else:
134 2
            other_products.add(product)
135
136 2
    products = namedtuple('products', ['linux', 'other'])
137 2
    return products(linux_products, other_products)
138
139
140 2
def get_profiles_directory(env_yaml):
141
    profiles_root = None
142
    if env_yaml:
143
        profiles_root = required_key(env_yaml, "profiles_root")
144
    return profiles_root
145
146
147 2
def get_profile_files_from_root(env_yaml, product_yaml):
148
    profile_files = []
149
    if env_yaml:
150
        profiles_root = get_profiles_directory(env_yaml)
151
        base_dir = os.path.dirname(product_yaml)
152
        profile_files = sorted(glob("{base_dir}/{profiles_root}/*.profile"
153
                               .format(profiles_root=profiles_root, base_dir=base_dir)))
154
    return profile_files
155