autotailor: Apply review feedback for data stream validation

- Move XML namespace dict to module-level DS_NAMESPACES constant
- Split _parse_datastream into _extract_profiles/values/rules/groups
- Add selector validation for -V/--var-select option
- Fix terminology: "datastream" -> "data stream" in all user-facing text
- Add --no-validate option to the man page

Author: ggbecker

tests/utils/autotailor_integration_test.sh

@@ -112,6 +112,17 @@ assert_exists 1 '/Benchmark/TestResult/rule-result[@idref="xccdf_com.example.www
 assert_exists 1 '/Benchmark/TestResult/rule-result[@idref="xccdf_com.example.www_rule_R3"]/result[text()="notselected"]'
 assert_exists 1 '/Benchmark/TestResult/rule-result[@idref="xccdf_com.example.www_rule_R4"]/result[text()="notselected"]'
 
+# invalid selector for V1 should fail with a descriptive error
+! python3 $autotailor --id-namespace "com.example.www" --var-select V1=invalid $ds $original_profile 2>$stdout
+grep "Selector 'invalid' does not exist" $stdout
+
+# invalid selector for V2 should fail with available selectors listed
+! python3 $autotailor --id-namespace "com.example.www" --var-select V2=invalid $ds $original_profile 2>$stdout
+grep "Available selectors" $stdout
+
+# --no-validate bypasses selector validation
+python3 $autotailor --id-namespace "com.example.www" --no-validate --var-select V1=invalid $ds $original_profile > $tailoring
+
 # use JSON tailoring (P1)
 python3 $autotailor $ds --id-namespace "com.example.www" --json-tailoring $json_tailoring > $tailoring
 $OSCAP xccdf eval --profile JSON_P1 --progress --tailoring-file $tailoring --results $result $ds

tests/utils/test_autotailor.py

@@ -229,3 +229,51 @@ def test_validator_suggestions():
     error_msg = str(e.value)
     assert "Did you mean one of these?" in error_msg
     assert "xccdf_com.example.www_rule_R1" in error_msg
+
+
+def test_validate_selector():
+    """Test that validate_selector rejects selectors not present in the data stream."""
+    ds_path = pathlib.Path(__file__).parent.joinpath("data_stream.xml")
+    validator = autotailor.DataStreamValidator(str(ds_path))
+
+    # V1 has selector "thirty"; V2 has "some" and "other"
+    validator.validate_selector("xccdf_com.example.www_value_V1", "thirty")
+    validator.validate_selector("xccdf_com.example.www_value_V2", "some")
+    validator.validate_selector("xccdf_com.example.www_value_V2", "other")
+
+    # Invalid selector for V1
+    with pytest.raises(ValueError) as e:
+        validator.validate_selector("xccdf_com.example.www_value_V1", "invalid")
+    error_msg = str(e.value)
+    assert "Selector 'invalid' does not exist for Value 'xccdf_com.example.www_value_V1'" in error_msg
+    assert "thirty" in error_msg
+
+    # Invalid selector for V2 with a close-enough typo triggers a suggestion
+    with pytest.raises(ValueError) as e:
+        validator.validate_selector("xccdf_com.example.www_value_V2", "ther")
+    error_msg = str(e.value)
+    assert "Selector 'ther' does not exist for Value 'xccdf_com.example.www_value_V2'" in error_msg
+    assert "other" in error_msg
+
+
+def test_profile_selector_validation():
+    """Test that Profile validates selectors via -V/--var-select through refine_value."""
+    ds_path = pathlib.Path(__file__).parent.joinpath("data_stream.xml")
+    validator = autotailor.DataStreamValidator(str(ds_path))
+
+    p = autotailor.Profile(validator=validator)
+    p.reverse_dns = "com.example.www"
+
+    # Valid selector passes
+    p.change_selectors(["V1=thirty"])
+    p.change_selectors(["V2=some"])
+
+    # Invalid selector raises
+    with pytest.raises(ValueError) as e:
+        p.change_selectors(["V1=invalid"])
+    assert "Selector 'invalid' does not exist for Value 'xccdf_com.example.www_value_V1'" in str(e.value)
+
+    # Invalid value ID still raises before reaching selector check
+    with pytest.raises(ValueError) as e:
+        p.change_selectors(["NONEXISTENT=thirty"])
+    assert "Value ID 'xccdf_com.example.www_value_NONEXISTENT' does not exist" in str(e.value)

utils/autotailor

@@ -38,6 +38,11 @@ ROLES = ["full", "unscored", "unchecked"]
 SEVERITIES = ["unknown", "info", "low", "medium", "high"]
 ATTRIBUTES = ["role", "severity"]
 
+DS_NAMESPACES = {
+    'ds': DS_NS,
+    'xccdf': NS
+}
+
 
 def quote(string):
     return "\"" + string + "\""
@@ -56,60 +61,62 @@ def is_valid_xccdf_id(string):
 
 
 class DataStreamValidator:
-    """Validates IDs against the SCAP datastream."""
+    """Validates IDs against the SCAP data stream."""
 
     def __init__(self, datastream_path):
         self.datastream_path = datastream_path
         self.profile_ids = set()
         self.value_ids = set()
+        self.value_selectors = {}
         self.rule_ids = set()
         self.group_ids = set()
         self._parse_datastream()
 
+    def _extract_profiles(self, benchmark):
+        for profile in benchmark.findall('.//xccdf:Profile', DS_NAMESPACES):
+            profile_id = profile.get('id')
+            if profile_id:
+                self.profile_ids.add(profile_id)
+
+    def _extract_values(self, benchmark):
+        for value in benchmark.findall('.//xccdf:Value', DS_NAMESPACES):
+            value_id = value.get('id')
+            if not value_id:
+                continue
+            self.value_ids.add(value_id)
+            selectors = set()
+            for val_elem in value.findall('xccdf:value', DS_NAMESPACES):
+                selector = val_elem.get('selector')
+                if selector:
+                    selectors.add(selector)
+            self.value_selectors[value_id] = selectors
+
+    def _extract_rules(self, benchmark):
+        for rule in benchmark.findall('.//xccdf:Rule', DS_NAMESPACES):
+            rule_id = rule.get('id')
+            if rule_id:
+                self.rule_ids.add(rule_id)
+
+    def _extract_groups(self, benchmark):
+        for group in benchmark.findall('.//xccdf:Group', DS_NAMESPACES):
+            group_id = group.get('id')
+            if group_id:
+                self.group_ids.add(group_id)
+
     def _parse_datastream(self):
-        """Parse the datastream to extract all valid IDs."""
         try:
             tree = ET.parse(self.datastream_path)
             root = tree.getroot()
-
-            # Register namespaces
-            namespaces = {
-                'ds': DS_NS,
-                'xccdf': NS
-            }
-
-            # Find all Benchmark elements (may be in data-stream-collection or standalone)
-            benchmarks = root.findall('.//xccdf:Benchmark', namespaces)
-
+            benchmarks = root.findall('.//xccdf:Benchmark', DS_NAMESPACES)
             for benchmark in benchmarks:
-                # Extract Profile IDs
-                for profile in benchmark.findall('.//xccdf:Profile', namespaces):
-                    profile_id = profile.get('id')
-                    if profile_id:
-                        self.profile_ids.add(profile_id)
-
-                # Extract Value IDs
-                for value in benchmark.findall('.//xccdf:Value', namespaces):
-                    value_id = value.get('id')
-                    if value_id:
-                        self.value_ids.add(value_id)
-
-                # Extract Rule IDs
-                for rule in benchmark.findall('.//xccdf:Rule', namespaces):
-                    rule_id = rule.get('id')
-                    if rule_id:
-                        self.rule_ids.add(rule_id)
-
-                # Extract Group IDs
-                for group in benchmark.findall('.//xccdf:Group', namespaces):
-                    group_id = group.get('id')
-                    if group_id:
-                        self.group_ids.add(group_id)
-
+                self._extract_profiles(benchmark)
+                self._extract_values(benchmark)
+                self._extract_rules(benchmark)
+                self._extract_groups(benchmark)
         except ET.ParseError as e:
-            raise ValueError(f"Failed to parse datastream '{self.datastream_path}': {e}")
+            raise ValueError(f"Failed to parse data stream '{self.datastream_path}': {e}")
         except FileNotFoundError:
-            raise ValueError(f"Datastream file not found: '{self.datastream_path}'")
+            raise ValueError(f"Data stream file not found: '{self.datastream_path}'")
 
     def _suggest_similar(self, invalid_id, valid_ids, n=3):
         """Suggest similar valid IDs using fuzzy matching."""
@@ -120,8 +127,7 @@ class DataStreamValidator:
         return matches
 
     def _create_validation_error(self, id_type, invalid_id, valid_ids):
-        """Create a detailed error message with suggestions."""
-        msg = f"{id_type} ID '{invalid_id}' does not exist in the datastream."
+        msg = f"{id_type} ID '{invalid_id}' does not exist in the data stream."
 
         suggestions = self._suggest_similar(invalid_id, valid_ids)
         if suggestions:
@@ -130,29 +136,40 @@ class DataStreamValidator:
                 msg += f"\n  - {suggestion}"
 
         if not valid_ids:
-            msg += f"\n\nNo {id_type.lower()}s found in the datastream."
+            msg += f"\n\nNo {id_type.lower()}s found in the data stream."
         else:
-            msg += f"\n\nAvailable {id_type.lower()}s can be listed by examining the datastream file."
+            msg += f"\n\nAvailable {id_type.lower()}s can be listed by examining the data stream file."
 
         return msg
 
     def validate_profile(self, profile_id):
-        """Validate a profile ID exists in the datastream."""
         if profile_id not in self.profile_ids:
             raise ValueError(self._create_validation_error("Profile", profile_id, self.profile_ids))
 
     def validate_value(self, value_id):
-        """Validate a value ID exists in the datastream."""
         if value_id not in self.value_ids:
             raise ValueError(self._create_validation_error("Value", value_id, self.value_ids))
 
+    def validate_selector(self, value_id, selector):
+        selectors = self.value_selectors.get(value_id, set())
+        if selector not in selectors:
+            msg = f"Selector '{selector}' does not exist for Value '{value_id}' in the data stream."
+            if selectors:
+                suggestions = self._suggest_similar(selector, selectors)
+                if suggestions:
+                    msg += "\n\nDid you mean one of these?"
+                    for suggestion in suggestions:
+                        msg += f"\n  - {suggestion}"
+                msg += f"\n\nAvailable selectors: {', '.join(sorted(selectors))}"
+            else:
+                msg += "\n\nNo selectors found for this value in the data stream."
+            raise ValueError(msg)
+
     def validate_rule(self, rule_id):
-        """Validate a rule ID exists in the datastream."""
         if rule_id not in self.rule_ids:
             raise ValueError(self._create_validation_error("Rule", rule_id, self.rule_ids))
 
     def validate_group(self, group_id):
-        """Validate a group ID exists in the datastream."""
         if group_id not in self.group_ids:
             raise ValueError(self._create_validation_error("Group", group_id, self.group_ids))
 
@@ -251,6 +268,8 @@ class Profile:
         Profile._validate_value_refinement_params(value_id, attribute, value)
         if self.validator:
             self.validator.validate_value(value_id)
+            if attribute == 'selector':
+                self.validator.validate_selector(value_id, value)
         self._prevent_duplicate_value_refinement(attribute, value_id, value)
         self._value_refinements[value_id][attribute] = value
 
@@ -610,8 +629,8 @@ def get_parser():
         "Absolute paths are converted to basename, relative paths are preserved.")
     parser.add_argument(
         "--no-validate", action="store_true",
-        help="Skip validation of IDs against the datastream. This significantly speeds up "
-        "execution on large datastreams but may produce invalid tailoring files if incorrect "
+        help="Skip validation of IDs against the data stream. This significantly speeds up "
+        "execution on large data streams but may produce invalid tailoring files if incorrect "
         "IDs are provided. Use with caution.")
     return parser
 
@@ -626,13 +645,12 @@ if __name__ == "__main__":
         parser.error("one of the following arguments has to be provided: "
                      "BASE_PROFILE_ID or --json-tailoring JSON_TAILORING_FILENAME")
 
-    # Create validator to check IDs against the datastream (unless --no-validate is specified)
     validator = None
     if not args.no_validate:
         try:
             validator = DataStreamValidator(args.datastream)
         except (ValueError, FileNotFoundError) as e:
-            print(f"Error loading datastream: {e}", file=sys.stderr)
+            print(f"Error loading data stream: {e}", file=sys.stderr)
             sys.exit(1)
 
     t = Tailoring(validator=validator)

utils/autotailor.8

@@ -8,8 +8,8 @@ A tailoring file adds a new profile, which is supposed to extend a profile that
 Tailoring can add, remove or refine rules, and it also can redefine contents of XCCDF variables.
 
 The tool requires data stream location and ID of the base profile as inputs.
-Note however, that the referenced data stream is not opened, and the validity of tailoring is not checked against it.
-The tool doesn't prevent you from extending non-existent profiles, selecting non-existent rules, and so on.
+By default, the tool parses the referenced data stream and validates provided IDs (profiles, rules, values, groups, and selectors) against it.
+Validation can be skipped using the \fB--no-validate\fR option, which significantly speeds up execution on large data streams at the cost of no longer catching invalid IDs.
 
 .SH SYNOPSIS
 autotailor [OPTION...] DATASTREAM_FILE [BASE_PROFILE_ID]
@@ -81,6 +81,11 @@ Use local path for the benchmark href instead of absolute file:// URI. This opti
 When this option is specified, absolute paths are converted to basename only, while relative paths are preserved as provided.
 By default, the tool uses absolute file:// URIs for backward compatibility.
 .RE
+.TP
+\fB--no-validate\fR
+.RS
+Skip validation of IDs against the data stream. This significantly speeds up execution on large data streams but may produce invalid tailoring files if incorrect IDs are provided. Use with caution.
+.RE
 
 .SH USAGE
 .SS Modify a variable value