diff --git a/applications/openshift-virtualization/kubevirt-restrict-migration-tools-access/rule.yml b/applications/openshift-virtualization/kubevirt-restrict-migration-tools-access/rule.yml new file mode 100644 index 000000000000..21f999a5a135 --- /dev/null +++ b/applications/openshift-virtualization/kubevirt-restrict-migration-tools-access/rule.yml @@ -0,0 +1,32 @@ +documentation_complete: true + +title: 'Restrict Namespace Administrator Access to Migration Tools' + +description: |- + Only authorized subjects should be allowed to create + VirtualMachineInstanceMigration (vmim) and + MigrationPolicy resources. Unrestricted access to these + resources allows namespace administrators to trigger live migrations + and define migration policies, which can affect workload placement, + resource consumption, and the overall stability of the cluster. + +rationale: |- + Virtual machine live migration moves a running VM between nodes. + Granting the ability to create VirtualMachineInstanceMigration + or MigrationPolicy objects to untrusted or unnecessary + subjects increases the risk of unplanned resource contention, + denial of service through excessive migrations, and potential + exposure of workload data during the migration process. Restricting + access to these resources ensures that only approved administrators + can initiate or influence VM migration behavior. + +severity: medium + +ocil_clause: 'unauthorized subjects can create vmim or migrationpolicy resources' + +ocil: |- + Run the following commands to check which subjects can create + migration-related resources: +
$ oc adm policy who-can create vmim
+
$ oc adm policy who-can create migrationpolicy
+ Verify that only authorized subjects are listed in the output. diff --git a/build-scripts/build_cel_content.py b/build-scripts/build_cel_content.py index 74786b6b9185..5f14f2aa1e5a 100755 --- a/build-scripts/build_cel_content.py +++ b/build-scripts/build_cel_content.py @@ -61,37 +61,40 @@ def setup_logging(log_level_str): logging.basicConfig(format=MESSAGE_FORMAT, level=numeric_level) -def load_cel_rules(rules_dir): +def load_rules(rules_dir): """ - Load all rules that use the CEL checking engine. + Load all compiled rules, separating CEL rules from the rest. + + Scans the directory once, returning CEL rules (with expression and + inputs) and all remaining rules keyed by ID. Args: rules_dir: Directory containing resolved rule JSON files Returns: - dict: Dictionary of rule_id -> rule object for rules with CEL checks + tuple: (cel_rules, all_rules) where cel_rules is a dict of + rule_id -> rule object for rules with CEL checks, and + all_rules is a dict of rule_id -> rule object for all rules Raises: ValueError: If a rule with CEL checks is missing required fields """ cel_rules = {} + all_rules = {} if not os.path.isdir(rules_dir): - return cel_rules + return cel_rules, all_rules for rule_file in os.listdir(rules_dir): rule_path = os.path.join(rules_dir, rule_file) try: rule = ssg.build_yaml.Rule.from_compiled_json(rule_path) + all_rules[rule.id_] = rule - # Check if this rule has CEL checks by looking for CEL-specific fields - # A rule uses CEL if it has both expression and inputs - # (loaded from cel/shared.yml during rule compilation) has_expression = hasattr(rule, 'expression') and rule.expression has_inputs = hasattr(rule, 'inputs') and rule.inputs if has_expression and has_inputs: - # Validate required CEL fields rule_name = rule_id_to_name(rule.id_) if not hasattr(rule, 'check_type') or not rule.check_type: @@ -102,16 +105,14 @@ def load_cel_rules(rules_dir): cel_rules[rule.id_] = rule except ssg.build_yaml.DocumentationNotComplete: - # Skip documentation-incomplete rules in non-debug builds continue except ValueError: - # Re-raise validation errors raise except Exception as e: logging.warning("Failed to load rule from %s: %s", rule_file, e) continue - return cel_rules + return cel_rules, all_rules def load_profiles(profiles_dir, cel_rule_ids): @@ -300,27 +301,42 @@ def profile_to_cel_dict(profile, cel_rule_ids): return cel_profile -def generate_cel_content(cel_rules, profiles): +def generate_cel_content(cel_rules, profiles, all_rule_ids=None, + manual_rules=None): """ Generate the complete CEL content structure. Args: cel_rules: Dictionary of rules with CEL checks profiles: List of profiles targeting the CEL checking engine + all_rule_ids: Set of all compiled rule IDs (used to distinguish + manual rules from nonexistent rules) + manual_rules: Dictionary of manual rules (rules without CEL checks + that are selected by CEL profiles) Returns: dict: Complete CEL content structure Raises: - ValueError: If duplicate rule names found or profile references unknown rules + ValueError: If duplicate rule names found or profile references + a rule that does not exist """ cel_rule_ids = set(cel_rules.keys()) + if all_rule_ids is None: + all_rule_ids = cel_rule_ids + if manual_rules is None: + manual_rules = {} + + # Combine CEL and manual rules for output + all_output_rules = dict(cel_rules) + all_output_rules.update(manual_rules) + all_output_rule_ids = set(all_output_rules.keys()) # Generate rules section and check for duplicates cel_rules_list = [] rule_names_seen = set() - for rule_id in sorted(cel_rules.keys()): - rule = cel_rules[rule_id] + for rule_id in sorted(all_output_rules.keys()): + rule = all_output_rules[rule_id] cel_rule = rule_to_cel_dict(rule) # Check for duplicate rule names @@ -334,16 +350,23 @@ def generate_cel_content(cel_rules, profiles): # Generate profiles section and validate rule references cel_profiles = [] for profile in profiles: - # Validate that all selected rules have CEL checks profile_name = rule_id_to_name(profile.id_) for rule_id in profile.selected: - if rule_id not in cel_rule_ids: + if rule_id not in all_output_rule_ids: rule_name = rule_id_to_name(rule_id) - raise ValueError( - f"profile '{profile_name}' references unknown rule '{rule_name}'" - ) + if rule_id in all_rule_ids: + logging.warning( + "profile '%s' references rule '%s' without CEL checks " + "(manual rule) - skipping from CEL content", + profile_name, rule_name, + ) + else: + raise ValueError( + f"profile '{profile_name}' references unknown rule " + f"'{rule_name}'" + ) - cel_profile = profile_to_cel_dict(profile, cel_rule_ids) + cel_profile = profile_to_cel_dict(profile, all_output_rule_ids) if cel_profile: cel_profiles.append(cel_profile) @@ -360,17 +383,30 @@ def main(): args = parse_args() setup_logging(args.log) - # Load rules with CEL checks - cel_rules = load_cel_rules(args.resolved_rules_dir) + # Load all rules in a single pass + cel_rules, all_rules = load_rules(args.resolved_rules_dir) + all_rule_ids = set(all_rules.keys()) if not cel_rules: content = {'profiles': [], 'rules': []} else: # Load profiles - profiles = load_profiles(args.profiles_dir, set(cel_rules.keys())) + cel_rule_ids = set(cel_rules.keys()) + profiles = load_profiles(args.profiles_dir, cel_rule_ids) + + # Collect manual rules: selected by CEL profiles but no CEL checks + manual_rules = {} + for profile in profiles: + for rule_id in profile.selected: + if rule_id not in cel_rule_ids and rule_id in all_rule_ids: + rule = all_rules[rule_id] + rule.check_type = 'Manual' + manual_rules[rule_id] = rule # Generate CEL content - content = generate_cel_content(cel_rules, profiles) + content = generate_cel_content( + cel_rules, profiles, all_rule_ids, manual_rules + ) # Write output YAML os.makedirs(os.path.dirname(args.output), exist_ok=True) diff --git a/docs/manual/developer/13_cel_content.md b/docs/manual/developer/13_cel_content.md index 91cbddf9c15d..d4fd977096a9 100644 --- a/docs/manual/developer/13_cel_content.md +++ b/docs/manual/developer/13_cel_content.md @@ -141,7 +141,7 @@ selections: - kubevirt-persistent-reservation-disabled ``` -**Important:** CEL profiles can only select CEL rules. If a profile includes both CEL and OVAL rules, only the CEL rules will be included in the generated CEL content file. +**Important:** CEL profiles can select both CEL rules and manual rules (rules without `cel/shared.yml`). Only CEL rules are included in the generated CEL content file; manual rules are skipped with a build warning. ## Creating a CEL Rule @@ -347,8 +347,8 @@ The build system validates CEL content automatically: **Profile Validation:** - `selected` field must contain at least one rule -- All selected rules must exist in CEL rules -- Profile cannot reference OVAL rules +- Rules without CEL checks (manual rules) are skipped with a warning +- Only CEL rules are included in the generated content **Content Validation:** - No duplicate rule names (after underscore-to-hyphen conversion) @@ -434,9 +434,10 @@ cel-spec '{"resource": {"spec": {"enabled": true}}}' 'resource.spec.enabled == t **Error: `CEL profile 'profile-name' has no rules`** - Add rules to the `selections` field in the profile -**Error: `profile 'profile-name' references unknown rule 'rule-name'`** -- Verify the rule exists and has CEL checks (has `cel/shared.yml` with `expression` and `inputs`) -- Check the rule ID matches the profile selection +**Warning: `profile 'profile-name' references rule 'rule-name' without CEL checks (manual rule)`** +- This is expected for manual rules that have no automated CEL check +- The rule will be skipped from CEL content output but remains in the profile selections +- If this is unintentional, verify the rule has `cel/shared.yml` with `expression` and `inputs` ### CEL Content Not Generated diff --git a/products/ocp4/profiles/cis-vm-extension.profile b/products/ocp4/profiles/cis-vm-extension.profile index a835ac42f886..41b6d98e738e 100644 --- a/products/ocp4/profiles/cis-vm-extension.profile +++ b/products/ocp4/profiles/cis-vm-extension.profile @@ -28,3 +28,4 @@ selections: - kubevirt-persistent-reservation-disabled - kubevirt-no-vms-overcommitting-guest-memory - kubevirt-enforce-trusted-tls-registries + - kubevirt-restrict-migration-tools-access diff --git a/tests/unit/ssg-module/test_build_cel_content.py b/tests/unit/ssg-module/test_build_cel_content.py index c0fbbe67fa85..9f44737d5af0 100644 --- a/tests/unit/ssg-module/test_build_cel_content.py +++ b/tests/unit/ssg-module/test_build_cel_content.py @@ -239,25 +239,29 @@ def test_extract_controls_from_references(): assert controls_empty == {} -def test_load_cel_rules(temp_rules_dir): +def test_load_rules(temp_rules_dir): """Test loading rules with CEL checks from directory.""" - cel_rules = build_cel_content.load_cel_rules(temp_rules_dir) + cel_rules, all_rules = build_cel_content.load_rules(temp_rules_dir) - # Should load only the rule with CEL checks (identified by presence of expression + inputs) + # Should load only the rule with CEL checks assert len(cel_rules) == 1 assert 'kubevirt_nonroot_feature_gate_is_enabled' in cel_rules + # all_rules should contain both CEL and non-CEL rules + assert len(all_rules) == 2 + assert 'some_oval_rule' in all_rules + rule = cel_rules['kubevirt_nonroot_feature_gate_is_enabled'] - # Rules with CEL checks are identified by presence of expression and inputs assert hasattr(rule, 'expression') and rule.expression assert hasattr(rule, 'inputs') and rule.inputs assert rule.title == 'Ensure NonRoot Feature Gate is Enabled' -def test_load_cel_rules_nonexistent_dir(): - """Test loading rules with CEL checks from nonexistent directory.""" - cel_rules = build_cel_content.load_cel_rules('/nonexistent/path') +def test_load_rules_nonexistent_dir(): + """Test loading rules from nonexistent directory.""" + cel_rules, all_rules = build_cel_content.load_rules('/nonexistent/path') assert cel_rules == {} + assert all_rules == {} def test_load_profiles(temp_profiles_dir): @@ -469,12 +473,13 @@ def test_load_cel_rules_missing_expression(): # Should not raise error - rule is not identified as CEL without both expression and inputs # This rule will be skipped since it doesn't have both fields - cel_rules = build_cel_content.load_cel_rules(tmpdir) - assert len(cel_rules) == 0 # Rule should be skipped + cel_rules, all_rules = build_cel_content.load_rules(tmpdir) + assert len(cel_rules) == 0 # Not a CEL rule + assert len(all_rules) == 1 # But still loaded as a rule def test_load_cel_rules_missing_inputs(): - """Test that rule without inputs is skipped.""" + """Test that rule without inputs is skipped from CEL rules.""" with tempfile.TemporaryDirectory() as tmpdir: # Create rule without inputs (but with expression - incomplete for CEL checks) rule_dict = { @@ -493,10 +498,9 @@ def test_load_cel_rules_missing_inputs(): with open(rule_path, 'w') as f: json.dump(rule_dict, f) - # Should not raise error - rule is not identified as CEL without both expression and inputs - # This rule will be skipped since it doesn't have both fields - cel_rules = build_cel_content.load_cel_rules(tmpdir) - assert len(cel_rules) == 0 # Rule should be skipped + cel_rules, all_rules = build_cel_content.load_rules(tmpdir) + assert len(cel_rules) == 0 # Not a CEL rule + assert len(all_rules) == 1 # But still loaded as a rule def test_load_profiles_no_rules(): @@ -574,6 +578,8 @@ def test_generate_cel_content_unknown_rule_reference(): 'existing_rule': rule1 } + all_rule_ids = {'existing_rule'} + # Create a profile that references a non-existent rule profile = ssg.build_yaml.Profile('test_profile') profile.id_ = 'test_profile' @@ -584,7 +590,7 @@ def test_generate_cel_content_unknown_rule_reference(): profiles = [profile] with pytest.raises(ValueError, match="references unknown rule 'nonexistent-rule'"): - build_cel_content.generate_cel_content(cel_rules, profiles) + build_cel_content.generate_cel_content(cel_rules, profiles, all_rule_ids) def test_validation_empty_expression(): @@ -608,13 +614,13 @@ def test_validation_empty_expression(): with open(rule_path, 'w') as f: json.dump(rule_dict, f) - # Empty expression means rule is not identified as CEL and is skipped - cel_rules = build_cel_content.load_cel_rules(tmpdir) + cel_rules, all_rules = build_cel_content.load_rules(tmpdir) assert len(cel_rules) == 0 + assert len(all_rules) == 1 def test_validation_empty_inputs(): - """Test that rule with empty inputs list is skipped.""" + """Test that rule with empty inputs list is skipped from CEL rules.""" with tempfile.TemporaryDirectory() as tmpdir: # Create rule with empty inputs rule_dict = { @@ -634,9 +640,9 @@ def test_validation_empty_inputs(): with open(rule_path, 'w') as f: json.dump(rule_dict, f) - # Empty inputs means rule is not identified as CEL and is skipped - cel_rules = build_cel_content.load_cel_rules(tmpdir) + cel_rules, all_rules = build_cel_content.load_rules(tmpdir) assert len(cel_rules) == 0 + assert len(all_rules) == 1 def test_validation_profile_with_empty_selections(): @@ -663,7 +669,7 @@ def test_validation_profile_with_empty_selections(): def test_validation_mixed_oval_and_cel_in_profile(): - """Test that profile with both OVAL and CEL checks only includes rules with CEL checks.""" + """Test that profile with both OVAL and CEL checks only includes CEL rules in output.""" # Create rule with CEL checks cel_rule = ssg.build_yaml.Rule('cel_rule') cel_rule.id_ = 'cel_rule' @@ -679,19 +685,22 @@ def test_validation_mixed_oval_and_cel_in_profile(): 'cel_rule': cel_rule } + # oval_rule exists as a compiled rule but has no CEL checks + all_rule_ids = {'cel_rule', 'oval_rule'} + # Create a CEL profile that references both CEL and OVAL rules - # (OVAL rules won't be in cel_rule_ids) profile = ssg.build_yaml.Profile('mixed_profile') profile.id_ = 'mixed_profile' profile.title = 'Mixed Profile' profile.description = 'Test' - profile.selected = ['cel_rule', 'oval_rule'] # oval_rule doesn't have CEL checks + profile.selected = ['cel_rule', 'oval_rule'] profiles = [profile] - # This should fail because oval_rule doesn't have CEL checks - with pytest.raises(ValueError, match="references unknown rule 'oval-rule'"): - build_cel_content.generate_cel_content(cel_rules, profiles) + # Should warn about oval_rule but not error since it exists + content = build_cel_content.generate_cel_content(cel_rules, profiles, all_rule_ids) + assert len(content['rules']) == 1 + assert content['rules'][0]['id'] == 'cel_rule' def test_validation_integration_full_flow(): @@ -732,7 +741,7 @@ def test_validation_integration_full_flow(): json.dump(profile_dict, f) # Load and validate - cel_rules = build_cel_content.load_cel_rules(rules_dir) + cel_rules, all_rules = build_cel_content.load_rules(rules_dir) assert len(cel_rules) == 1 assert 'valid_cel_rule' in cel_rules