-
Notifications
You must be signed in to change notification settings - Fork 816
CMP-4435: Add manual rule kubevirt-restrict-migration-tools-access for CIS OCP-Virt 1.10 #14920
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from all commits
d269715
99cff34
72672a7
f8a6ce2
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| documentation_complete: true | ||
|
|
||
| title: 'Restrict Namespace Administrator Access to Migration Tools' | ||
|
|
||
| description: |- | ||
| Only authorized subjects should be allowed to create | ||
| <tt>VirtualMachineInstanceMigration</tt> (vmim) and | ||
| <tt>MigrationPolicy</tt> 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 <tt>VirtualMachineInstanceMigration</tt> | ||
| or <tt>MigrationPolicy</tt> 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: | ||
| <pre>$ oc adm policy who-can create vmim</pre> | ||
| <pre>$ oc adm policy who-can create migrationpolicy</pre> | ||
| Verify that only authorized subjects are listed in the output. |
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -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) | ||||||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why are these comments removed? |
||||||
| 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 | ||||||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It is not clear to me why we need |
||||||
| 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", | ||||||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The rule is not skipped, it is added without the CEL expression and inputs.
Suggested change
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We already identify that a rule is Manual here. Could we process profiles before rules? |
||||||
| 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: | ||||||
|
Comment on lines
+399
to
+400
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||||||
| if rule_id not in cel_rule_ids and rule_id in all_rule_ids: | ||||||
| rule = all_rules[rule_id] | ||||||
| rule.check_type = 'Manual' | ||||||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I don't agree with the The check type is about what is being checked, whether the kubernetes resource in the A Manual rule is just missing the automated check.
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We don't need to filter out manual CEL rules here, they can be identified directly inside |
||||||
| 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) | ||||||
|
|
||||||
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -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. | ||||||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Suggested change
|
||||||
|
|
||||||
| ## 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 | ||||||
|
|
||||||
|
|
||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.