Test Failed
Push — master ( aa538d...df2cf1 )
by Jan
02:59 queued 10s
created

ssg.products   A

Complexity

Total Complexity 25

Size/Duplication

Total Lines 148
Duplicated Lines 0 %

Test Coverage

Coverage 83.75%

Importance

Changes 0
Metric Value
eloc 97
dl 0
loc 148
ccs 67
cts 80
cp 0.8375
rs 10
c 0
b 0
f 0
wmc 25

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