1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349
|
from __future__ import absolute_import
from __future__ import print_function
import os
import sys
import yaml
from .controls import ControlsManager, Policy
from .products import (
get_profile_files_from_root,
load_product_yaml,
product_yaml_path,
)
if sys.version_info >= (3, 9):
dict_type = dict # Python 3.9+ supports built-in generics
list_type = list
tuple_type = tuple
else:
from typing import Dict as dict_type # Fallback for older versions
from typing import List as list_type
from typing import Tuple as tuple_type
class ProfileSelections:
"""
A class to represent profile with sections of rules and variables.
Attributes:
-----------
profile_id : str
The unique identifier for the profile.
profile_title : str
The profile title associated with the profile id.
product_id : str
The product id associated with the profile.
product_title : str
The product title associated with the product id.
"""
def __init__(self, profile_id, profile_title, product_id, product_title):
self.profile_id = profile_id
self.profile_title = profile_title
self.product_id = product_id
self.product_title = product_title
self.rules = []
self.unselected_rules = []
self.variables = {}
def _load_product_yaml(content_dir: str, product: str) -> object:
"""
Load the product YAML file and return its content as a Python object.
Args:
content_dir (str): The directory where the content is stored.
product (str): The name of the product.
Returns:
object: The loaded YAML content as a Python object.
"""
file_yaml_path = product_yaml_path(content_dir, product)
return load_product_yaml(file_yaml_path)
def _load_yaml_profile_file(file_path: str) -> dict_type:
"""
Load the content of a YAML file intended to profiles definitions.
It is not necessary to process macros in this case.
Args:
file_path (str): The path to the YAML file.
Returns:
dict: The content of the YAML file as a dictionary.
"""
with open(file_path, 'r') as file:
try:
return yaml.safe_load(file)
except yaml.YAMLError as e:
print(f"Error loading YAML profile file {file_path}: {e}")
return {}
def _get_extended_profile_path(profiles_files: list, profile_name: str) -> str:
"""
Retrieve the full path of a profile file from a list of profile file paths.
Args:
profiles_files (list of str): A list of file paths where profile files are located.
profile_name (str): The name of the profile to search for.
Returns:
str: The full path of the profile file if found, otherwise None.
"""
profile_file = f"{profile_name}.profile"
profile_path = next((path for path in profiles_files if profile_file in path), None)
return profile_path
def _process_profile_extension(profile: ProfileSelections, profile_yaml: dict,
profiles_files: list, policies: dict) -> ProfileSelections:
"""
Processes the extension of a profile by recursively checking if the profile extends another
profile and updating the profile selections accordingly.
Args:
profile (ProfileSelections): The profile object to be processed.
profile_yaml (dict): The YAML content of the current profile.
profiles_files (list): List of profile file paths.
policies (dict): The policies defined in the current profile.
Returns:
ProfileSelections: The updated profile object.
"""
extended_profile = profile_yaml.get("extends")
if isinstance(extended_profile, str):
extended_profile = _get_extended_profile_path(profiles_files, extended_profile)
if extended_profile is not None:
profile_yaml = _load_yaml_profile_file(extended_profile)
return _process_profile(profile, profile_yaml, profiles_files, policies)
return profile
def _parse_control_line(control_line: str) -> tuple_type[str, str]:
"""
Parses a control line string and returns a tuple containing the first and third parts of the
string, separated by a colon. If the string does not contain three parts, the second element
of the tuple defaults to 'all'.
Args:
control_line (str): The control line string to be parsed.
Returns:
tuple[str, str]: A tuple containing the first part of the control line and either the
third part or 'all' if the third part is not present.
"""
parts = control_line.split(":")
if len(parts) == 3:
return parts[0], parts[2]
return parts[0], 'all'
def _process_selected_variable(profile: ProfileSelections, variable: str) -> None:
"""
Processes a selected variable and updates the profile's variables.
Args:
profile (ProfileSelections): The profile object containing variables.
variable (str): The variable in the format 'name=value'.
Raises:
ValueError: If the variable is not in the correct format.
"""
variable_name, variable_value = variable.split('=', 1)
if variable_name not in profile.variables:
profile.variables[variable_name] = variable_value
def _process_selected_rule(profile: ProfileSelections, rule: str) -> None:
"""
Adds a rule to the profile's selected rules if it is not already selected or unselected.
Args:
profile (ProfileSelections): The profile containing selected and unselected rules.
rule (str): The rule to be added to the profile's selected rules.
Returns:
None
"""
if rule not in profile.unselected_rules and rule not in profile.rules:
profile.rules.append(rule)
def _process_control(profile: ProfileSelections, control: object) -> None:
"""
Processes a control by iterating through its rules and applying the appropriate processing
function. Not that at this level rules list in control can include both variables and rules.
The function distinguishes between variable and rules based on the presence of an '='
character in the rule.
Args:
profile (ProfileSelections): The profile selections to be processed.
control: The control object containing rules to be processed.
"""
for rule in control.rules:
if "=" in rule:
_process_selected_variable(profile, rule)
else:
_process_selected_rule(profile, rule)
def _update_profile_with_policy(profile: ProfileSelections, policy: Policy, level: str) -> None:
"""
Updates the given profile with controls from the specified policy based on the provided level.
Args:
profile (ProfileSelections): The profile to be updated.
policy (Policy): The policy containing controls to update the profile with.
level (str): The level of controls to be processed. If 'all', all controls are processed.
Otherwise, only controls matching the specified level are processed.
Returns:
None
"""
for control in policy.controls:
if level == 'all' or level in control.levels:
_process_control(profile, control)
def _process_controls(profile: ProfileSelections, control_line: str,
policies: dict) -> ProfileSelections:
"""
Process a control file inheritance to update profile selections based on the given policies.
Args:
profile (ProfileSelections): The profile object to be processed.
control_line (str): A string representing the control line, which contains a policy ID and
optionally a level, separated by colons.
policies (dict): A dictionary of policies, where each key is a policy ID and each value is
a policy object containing the controls.
Returns:
ProfileSelections: The updated profile object.
"""
policy_id, level = _parse_control_line(control_line)
policy = policies.get(policy_id)
if policy is None:
print(f"Policy {policy_id} not found")
return profile
_update_profile_with_policy(profile, policy, level)
return profile
def _process_selections(profile: ProfileSelections, profile_yaml: dict,
policies: dict) -> ProfileSelections:
"""
Processes the selections from the profile YAML and updates the profile accordingly.
Args:
profile (ProfileSelections): The profile object to be processed.
profile_yaml (dict): A dictionary containing the profile YAML data.
policies (dict): A dictionary containing policy information.
Returns:
profile: The updated profile object.
"""
selections = profile_yaml.get("selections", [])
for selected in selections:
if selected.startswith("!"):
profile.unselected_rules.append(selected[1:])
elif "=" in selected:
variable_name, variable_value = selected.split('=', 1)
profile.variables[variable_name] = variable_value
elif ":" in selected:
profile = _process_controls(profile, selected, policies)
else:
profile.rules.append(selected)
return profile
def _process_profile(profile: ProfileSelections, profile_yaml: dict, profiles_files: list,
policies: dict) -> ProfileSelections:
"""
Processes a profile by handling profile extensions, and processing selections.
Args:
profile (ProfileSelections): The profile object to be processed.
profile_yaml (dict): The YAML content of the profile.
profiles_files (list): A list of profile file paths.
policies (dict): A dictionary of policies defined by control files.
Returns:
ProfileSelections: The processed profile object.
"""
profile = _process_profile_extension(profile, profile_yaml, profiles_files, policies)
profile = _process_selections(profile, profile_yaml, policies)
return profile
def _load_controls_manager(controls_dir: str, product_yaml: dict) -> object:
"""
Loads and initializes a ControlsManager instance.
Args:
controls_dir (str): The directory containing control files.
product_yaml (dict): The product configuration in YAML format.
Returns:
object: An instance of ControlsManager with loaded controls.
"""
control_mgr = ControlsManager(controls_dir, product_yaml)
control_mgr.load()
return control_mgr
def _sort_profiles_selections(profiles: list) -> ProfileSelections:
"""
Sorts profiles selections (rules and variables) by selections ids.
Args:
profiles (list): A list of ProfileSelections objects to be sorted.
Returns:
ProfileSelections: The sorted list of ProfileSelections objects.
"""
for profile in profiles:
profile.rules = sorted(profile.rules)
profile.unselected_rules = sorted(profile.unselected_rules)
profile.variables = dict(sorted(profile.variables.items()))
return profiles
def get_profiles_from_products(content_dir: str, products: list,
sorted: bool = False) -> list_type:
"""
Retrieves profiles with respective variables from the given products.
Args:
content_dir (str): The directory containing the content.
products (list): A list of product names to retrieve profiles from.
Returns:
list: A list of ProfileVariables objects containing profile variables for each product.
"""
profiles = []
controls_dir = os.path.join(content_dir, 'controls')
for product in products:
product_yaml = _load_product_yaml(content_dir, product)
product_title = product_yaml.get("full_name")
profiles_files = get_profile_files_from_root(product_yaml, product_yaml)
controls_manager = _load_controls_manager(controls_dir, product_yaml)
for file in profiles_files:
profile_id = os.path.basename(file).split('.profile')[0]
profile_yaml = _load_yaml_profile_file(file)
profile_title = profile_yaml.get("title")
profile = ProfileSelections(profile_id, profile_title, product, product_title)
profile = _process_profile(profile, profile_yaml, profiles_files,
controls_manager.policies)
profiles.append(profile)
if sorted:
profiles = _sort_profiles_selections(profiles)
return profiles
|