Add per product control files

build-scripts/compile_all.py

@@ -213,11 +213,15 @@ def main():
     load_benchmark_source_data_from_directory_tree(loader, env_yaml, product_yaml)
 
     controls_dir = os.path.join(project_root_abspath, "controls")
+    product_controls_dir = os.path.join(product_yaml['product_dir'], 'controls')
+    controls_dirs = [controls_dir]
+    if os.path.exists(product_controls_dir):
+        controls_dirs.append(product_controls_dir)
 
     existing_rules = find_existing_rules(project_root_abspath)
 
     controls_manager = ssg.controls.ControlsManager(
-        controls_dir, env_yaml, existing_rules)
+        controls_dirs, env_yaml, existing_rules)
     controls_manager.load()
     controls_manager.remove_selections_not_known(loader.all_rules)
     controls_manager.add_references(loader.all_rules)

docs/adr/0003_per_product_controls.md

@@ -0,0 +1,19 @@
+---
+content-version: 0.1.79
+title: ADR-0003 - Per Product Controls
+status: proposed
+---
+
+## Context
+As of late October 2025  there was over 50 control files in the `controls` folder.
+Many of these control files are product specific.
+The goal of this ADR is to help keep product specific information separate from the global standard like ANSSI.
+
+
+## Decision
+We will allow the creation of `controls` directory under each product.
+All product specific control files will be moved to the product specific control file.
+The product specific control files in `products/example/controls` can override the controls in `controls`.
+
+## Consequences
+This will create more places to look for control files, but it will help keep product specific information with the product.

docs/manual/developer/03_creating_content.md

@@ -837,6 +837,8 @@ controls:
        - systemd_target_multi_user
 ```
 
+Control files that apply to multiple products should be stored in `controls` folder in the root of the project.
+If the control file is only applicable to one product it should be store in the `controls` directory under the products folder.
 
 ### Defining levels
 

docs/manual/developer/04_style_guide.md

@@ -317,7 +317,8 @@ Benchmark sections must be in the following order, if they are present.
 
 ### Controls
 
-These rules apply to the files in `controls/`.
+These rules apply to the files in `controls/` and `products/*/controls`.
+Product specific controls should be stored under the respective controls directory.
 All the above [YAML](#yaml) rules apply.
 
 #### Control Sections

ssg/controls.py

@@ -5,7 +5,9 @@
 import collections
 import os
 import copy
+import sys
 from glob import glob
+from typing import List, Dict
 
 import ssg.entities.common
 import ssg.yaml
@@ -369,6 +371,7 @@ class Policy(ssg.entities.common.XCCDFEntity):
         product (list): A list of products associated with the policy.
     """
     def __init__(self, filepath, env_yaml=None):
+        self.controls_dirs = []
         self.id = None
         self.env_yaml = env_yaml
         self.filepath = filepath
@@ -637,6 +640,7 @@ def load(self):
         controls_dir = yaml_contents.get("controls_dir")
         if controls_dir:
             self.controls_dir = os.path.join(os.path.dirname(self.filepath), controls_dir)
+            self.controls_dirs = [self.controls_dir]
         self.id = ssg.utils.required_key(yaml_contents, "id")
         self.policy = ssg.utils.required_key(yaml_contents, "policy")
         self.title = ssg.utils.required_key(yaml_contents, "title")
@@ -786,17 +790,17 @@ def add_references(self, rules):
             control.add_references(self.reference_type, rules)
 
 
-class ControlsManager():
+class ControlsManager:
     """
     Manages the loading, processing, and saving of control policies.
 
     Attributes:
-        controls_dir (str): The directory where control policy files are located.
+        controls_dirs (List[str]): The directories where control policy files are located.
         env_yaml (str, optional): The environment YAML file.
         existing_rules (dict, optional): Existing rules to check against.
         policies (dict): A dictionary of loaded policies.
     """
-    def __init__(self, controls_dir, env_yaml=None, existing_rules=None):
+    def __init__(self, controls_dirs: List[str], env_yaml=None, existing_rules=None):
         """
         Initializes the Controls class.
 
@@ -805,19 +809,25 @@ def __init__(self, controls_dir, env_yaml=None, existing_rules=None):
             env_yaml (str, optional): Path to the environment YAML file. Defaults to None.
             existing_rules (dict, optional): Dictionary of existing rules. Defaults to None.
         """
-        self.controls_dir = os.path.abspath(controls_dir)
+        self.controls_dirs = [os.path.abspath(controls_dir) for controls_dir in controls_dirs]
         self.env_yaml = env_yaml
         self.existing_rules = existing_rules
         self.policies = {}
 
     def _load(self, format):
-        if not os.path.exists(self.controls_dir):
-            return
-        for filename in sorted(glob(os.path.join(self.controls_dir, "*." + format))):
-            filepath = os.path.join(self.controls_dir, filename)
-            policy = Policy(filepath, self.env_yaml)
-            policy.load()
-            self.policies[policy.id] = policy
+        for controls_dir in self.controls_dirs:
+            if not os.path.isdir(controls_dir):
+                continue
+            for filepath in sorted(glob(os.path.join(controls_dir, "*." + format))):
+                policy = Policy(filepath, self.env_yaml)
+                policy.load()
+                if policy.id in self.policies:
+                    print(f"Policy {policy.id} was defined first at "
+                                     f"{self.policies[policy.id].filepath} and now another policy "
+                                     f"with the same ID is being loaded from {policy.filepath}."
+                                     f"Overriding with later.",
+                                     file=sys.stderr)
+                self.policies[policy.id] = policy
         self.check_all_rules_exist()
         self.resolve_controls()
 
@@ -948,7 +958,7 @@ def get_control(self, policy_id, control_id):
         control = policy.get_control(control_id)
         return control
 
-    def get_all_controls_dict(self, policy_id):
+    def get_all_controls_dict(self, policy_id: str) -> Dict[str, Control]:
         """
         Retrieve all controls for a given policy as a dictionary.
 
@@ -959,7 +969,6 @@ def get_all_controls_dict(self, policy_id):
             Dict[str, list]: A dictionary where the keys are control IDs and the values are lists
                              of controls.
         """
-        # type: (str) -> typing.Dict[str, list]
         policy = self._get_policy(policy_id)
         return policy.controls_by_id
 

ssg/profiles.py

@@ -294,16 +294,21 @@ def _load_controls_manager(controls_dir: str, product_yaml: dict) -> object:
 
     Args:
         controls_dir (str): The directory containing control files.
+        product_yaml (dict): The YAML content of the product.
 
     Returns:
         object: An instance of ControlsManager with loaded controls.
     """
-    control_mgr = ControlsManager(controls_dir, product_yaml)
+    product_controls_dir = os.path.join(product_yaml['product_dir'], 'controls')
+    control_dirs = [controls_dir]
+    if os.path.exists(product_controls_dir):
+        control_dirs.append(product_controls_dir)
+    control_mgr = ControlsManager(control_dirs, product_yaml)
     control_mgr.load()
     return control_mgr
 
 
-def _sort_profiles_selections(profiles: list) -> ProfileSelections:
+def _sort_profiles_selections(profiles: list) -> list_type[ProfileSelections]:
     """
     Sorts profiles selections (rules and variables) by selections ids.
 

tests/unit/ssg-module/data/content_dir/products/rhel8/controls/wxyz-levels.yml

@@ -0,0 +1,18 @@
+---
+policy: WXYZ Benchmark for securing Linux systems with levels
+title: WXYZ Benchmark for securing Linux systems with levels
+id: wxyz-levels
+version: 1.2.3
+source: https://www.wxyz.com/linux.pdf
+levels:
+  - id: low
+  - id: medium
+    inherits_from:
+      - low
+  - id: high
+    inherits_from:
+      - medium
+
+controls:
+  - id: S1
+    title: Package sudo must be installed

tests/unit/ssg-module/test_controls.py

@@ -8,7 +8,7 @@
 
 ssg_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
 data_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "data"))
-controls_dir = os.path.join(data_dir, "controls_dir")
+controls_dir = [os.path.join(data_dir, "controls_dir")]
 profiles_dir = os.path.join(data_dir, "profiles_dir")
 
 
@@ -63,6 +63,13 @@ def env_yaml():
     env_yaml = open_environment(build_config_yaml, product_yaml)
     return env_yaml
 
+@pytest.fixture
+def rhel8_controls_manager(env_yaml):
+    rhel_controls_dir =  controls_dir.copy()
+    rhel_controls_dir.append(os.path.join(data_dir, "content_dir", "products", "rhel8", "controls"))
+    controls_manager = ssg.controls.ControlsManager(rhel_controls_dir, env_yaml)
+    return controls_manager
+
 
 @pytest.fixture
 def controls_manager(env_yaml):
@@ -84,7 +91,7 @@ def compiled_controls_dir_py3(tmp_path):
 @pytest.fixture
 def compiled_controls_manager(env_yaml, controls_manager,compiled_controls_dir_py2):
     controls_manager.save_everything(compiled_controls_dir_py2)
-    controls_manager = ssg.controls.ControlsManager(compiled_controls_dir_py2, env_yaml)
+    controls_manager = ssg.controls.ControlsManager([compiled_controls_dir_py2], env_yaml)
     controls_manager.load_compiled()
     return controls_manager
 
@@ -413,7 +420,7 @@ def test_policy_parse_from_nested(minimal_empty_controls, one_simple_subcontrol)
     assert subcontrol.title == "subcontrol"
     assert subcontrol.selections == ["b"]
 
-    controls_manager = ssg.controls.ControlsManager("", dict())
+    controls_manager = ssg.controls.ControlsManager([""], dict())
     controls_manager.policies[policy.id] = policy
 
     controls_manager.resolve_controls()
@@ -428,7 +435,7 @@ def test_manager_removes_rules():
     policy.save_controls_tree([control_dict])
     policy.id = "P"
 
-    controls_manager = ssg.controls.ControlsManager("", dict())
+    controls_manager = ssg.controls.ControlsManager([], dict())
     controls_manager.policies[policy.id] = policy
 
     control = controls_manager.get_control("P", "top")
@@ -453,7 +460,7 @@ def test_policy_parse_from_nested2():
     policy = ssg.controls.Policy("")
     policy.id = "P"
 
-    controls_manager = ssg.controls.ControlsManager("", dict())
+    controls_manager = ssg.controls.ControlsManager([], dict())
     controls_manager.policies[policy.id] = policy
 
     controls = policy.save_controls_tree([top_control_dict, second_nested_dict, first_nested_dict])  # noqa: F841
@@ -476,7 +483,7 @@ def test_policy_parse_from_ours_and_foreign():
     foreign_policy.id = "foreign"
     foreign_policy.save_controls_tree([foreign_control_dict])
 
-    controls_manager = ssg.controls.ControlsManager("", dict())
+    controls_manager = ssg.controls.ControlsManager([], dict())
     controls_manager.policies[main_policy.id] = main_policy
     controls_manager.policies[foreign_policy.id] = foreign_policy
 
@@ -506,7 +513,7 @@ def test_policy_parse_foreign_with_all():
     foreign_policy.levels_by_id = {"level_1": level1, "level_2": level2}
     foreign_policy.save_controls_tree(foreign_control_dicts)
 
-    controls_manager = ssg.controls.ControlsManager("", dict())
+    controls_manager = ssg.controls.ControlsManager([], dict())
     controls_manager.policies[main_policy.id] = main_policy
     controls_manager.policies[foreign_policy.id] = foreign_policy
 
@@ -593,3 +600,14 @@ def test_references_from_controls(controls_manager, rules_for_test_references_fr
     assert len(rules["compiled_references_test_rule_2"].references) == 2
     assert rules["compiled_references_test_rule_2"].references["cis"] == ["R2"]
     assert rules["compiled_references_test_rule_2"].references["stig"] == ["17"]
+
+
+def test_product_controls(rhel8_controls_manager: ssg.controls.ControlsManager):
+    rhel8_controls_manager.load()
+    s1 = rhel8_controls_manager.get_control("wxyz-levels", "S1")
+    assert s1.title == "Package sudo must be installed"
+
+
+def test_not_overriding_controls(rhel8_controls_manager: ssg.controls.ControlsManager):
+    rhel8_controls_manager.load()
+    assert_control_confirms_to_standard(rhel8_controls_manager, 'abcd')

utils/controleval.py

@@ -138,7 +138,9 @@ def load_product_yaml(product: str) -> yaml:
 
 def load_controls_manager(controls_dir: str, product: str) -> object:
     product_yaml = load_product_yaml(product)
-    ctrls_mgr = controls.ControlsManager(controls_dir, dict(product_yaml))
+    product_controls_dir = os.path.join(product_yaml['product_dir'], 'controls')
+    ctrls_mgr = controls.ControlsManager([controls_dir, product_controls_dir],
+                                         dict(product_yaml))
     ctrls_mgr.load()
     return ctrls_mgr
 

utils/controlrefcheck.py

@@ -74,7 +74,11 @@ def get_controls_env(args):
     product_yaml = os.path.join(product_base, "product.yml")
     env_yaml = ssg.environment.open_environment(
         args.build_config_yaml, product_yaml, os.path.join(SSG_ROOT, "product_properties"))
-    controls_manager = ssg.controls.ControlsManager(args.controls, env_yaml)
+    controls_dir = [os.path.join(SSG_ROOT, "controls")]
+    product_controls_dir = os.path.join(product_base, "controls")
+    if os.path.exists(product_controls_dir):
+        controls_dir.append(product_controls_dir)
+    controls_manager = ssg.controls.ControlsManager(controls_dir, env_yaml)
     controls_manager.load()
     return controls_manager, env_yaml
 

utils/import_disa_stig.py

@@ -48,7 +48,7 @@ def _get_env_yaml(ssg_root: str, product: str, build_config_yaml: str) -> dict:
 
 
 def _get_controls(control_name, ssg_root, env_yaml):
-    control_manager = ssg.controls.ControlsManager(os.path.join(ssg_root, "controls"), env_yaml)
+    control_manager = ssg.controls.ControlsManager([os.path.join(ssg_root, "controls")], env_yaml)
     control_manager.load()
     controls = control_manager.get_all_controls_dict(control_name)
     return controls

utils/oscal/control_selector.py

@@ -37,7 +37,7 @@ def __init__(
             filter_by_level: Optional level to filter by.
         """
         controls_dir = os.path.join(ssg_root, "controls")
-        controls_manager = ControlsManager(controls_dir=controls_dir, env_yaml=env_yaml)
+        controls_manager = ControlsManager(controls_dirs=[controls_dir], env_yaml=env_yaml)
         controls_manager.load()
         if control not in controls_manager.policies:
             raise ValueError(f"Policy {control} not found in controls")

utils/refchecker.py

@@ -139,7 +139,7 @@ def main():
 
     controls_manager = None
     if os.path.exists(args.controls):
-        controls_manager = ssg.controls.ControlsManager(args.controls, env_yaml)
+        controls_manager = ssg.controls.ControlsManager([args.controls], env_yaml)
         controls_manager.load()
 
     profiles_root = os.path.join(product_base, "profiles")