Add Ruff to the project and CI

.github/workflows/ruff.yaml

@@ -0,0 +1,23 @@
+name: Ruff Gate
+on:
+    merge_group:
+        branches: [ 'master' ]
+    push:
+        branches: [ '*', '!stabilization*', '!stable*', '!master' ]
+    pull_request:
+        branches: [ 'master', 'stabilization*', 'oscal-update-*' ]
+
+concurrency:
+    group: ${{ github.workflow }}-${{ github.event.number || github.run_id }}
+    cancel-in-progress: true
+jobs:
+    ruff:
+        name: Run ruff
+        runs-on: ubuntu-latest
+        steps:
+            -   name: Checkout
+                uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 #v5.0.0
+            -   name: Install ruff
+                run: python3 -m pip install ruff
+            -   name: Run ruff check
+                run: ruff check

build-scripts/compose_ds.py

@@ -55,7 +55,7 @@ def embed_sce_checks_in_datastream(datastreamtree, checklists, checks, build_dir
 
         # Create a component reference to map the checklist to the extended component
         component_ref_id = "scap_{}_cref_{}".format(ID_NS, mangled_path)
-        component_ref = ET.SubElement(
+        component_ref = ET.SubElement(  # noqa: F841
             checks, '{%s}component-ref' % datastream_namespace,
             attrib={
                 'id': component_ref_id,
@@ -66,7 +66,7 @@ def embed_sce_checks_in_datastream(datastreamtree, checklists, checks, build_dir
         checklists_component_ref = checklists.find(
             "{%s}component-ref" % datastream_namespace)
         catalog = checklists_component_ref.find('{%s}catalog' % cat_namespace)
-        uri = ET.SubElement(
+        uri = ET.SubElement(  # noqa: F841
             catalog, '{%s}uri' % cat_namespace,
             attrib={
                 'name': file,

ruff.toml

@@ -0,0 +1,16 @@
+line-length = 99
+indent-width = 4
+target-version = "py39"
+
+[lint]
+select = [
+    "F",    # Pyflakes
+    "E",    # pycodestyle
+    "W",    # pycodestyle
+]
+
+ignore = [
+    "F401",     # unused-import
+    "E402",     # module-import-not-at-top-of-file
+    "E501",         # line-too-long
+]

shared/templates/file_groupowner/template.py

@@ -15,7 +15,7 @@ def preprocess(data, lang):
     except ValueError:
         data["group_represented_with_gid"] = False
 
-    if data["group_represented_with_gid"] == False:
+    if not data["group_represented_with_gid"]:
         groups = data["gid_or_name"].split("|")
         if any(element.isnumeric() for element in groups):
             raise ValueError("gid_or_name list cannot contain gids when there are multiple groups")

shared/templates/file_owner/template.py

@@ -15,11 +15,11 @@ def preprocess(data, lang):
     except ValueError:
         data["owner_represented_with_uid"] = False
 
-    if data["owner_represented_with_uid"] == False:
+    if not data["owner_represented_with_uid"]:
         owners = data["uid_or_name"].split("|")
         if any(element.isnumeric() for element in owners):
             raise ValueError("uid_or_name list cannot contain uids when there are multiple owners")
-    
+
     if lang == "oval":
         data["fileid"] = data["_rule_id"].replace("file_owner", "")
     return data

shared/templates/pam_account_password_faillock/template.py

@@ -8,7 +8,7 @@ def preprocess(data, lang):
         data[var] = data.get(var, None)
         if data.get(var) is not None and \
            data.get(var) != "use_ext_variable" and \
-           type(data.get(var)) != int:
+            not isinstance(data.get(var), int):
             errmsg = ("The template instance of the rule {0} requires the "
                       "parameter {1} is either 'use_ext_variable' or "
                       "a number or undefined".format(data["_rule_id"], data["var"]))

ssg/build_cpe.py

@@ -45,7 +45,7 @@ def load_product_cpes(self, env_yaml):
         try:
             product_cpes_list = env_yaml["cpes"]
             self.load_product_cpes_from_list(product_cpes_list)
-        except KeyError as exc:
+        except KeyError:
             raise Exception("Product %s does not define 'cpes'" % (env_yaml["product"]))
 
     def load_product_cpes_from_list(self, product_cpes_list):

ssg/build_yaml.py

@@ -3159,7 +3159,7 @@ def load_benchmark(self, directory):
         for gid in benchmark_first_groups:
             try:
                 self.benchmark.add_group(self.groups[gid], self.env_yaml, self.product_cpes)
-            except KeyError as exc:
+            except KeyError:
                 # Add only the groups we have compiled and loaded
                 pass
         self.benchmark.drop_rules_not_included_in_a_profile()

ssg/content_diff.py

@@ -362,7 +362,7 @@ def check_for_new_rules(self, rules_in_new_benchmark, old_rule_mapping):
             new_stig_id = self._get_stig_id(new_rule)
             new_sv_rule_id = self.get_stig_rule_SV(new_rule.get_attr("id"))
             try:
-                old_sv_rule_id = old_rule_mapping[new_sv_rule_id]
+                old_sv_rule_id = old_rule_mapping[new_sv_rule_id]  # noqa: F841
             except KeyError:
                 print("%s was added in new data stream." % (new_stig_id))
 

ssg/controls.py

@@ -300,7 +300,7 @@ def add_references(self, reference_type, rules):
                 continue
             try:
                 rule.add_control_reference(reference_type, self.id)
-            except ValueError as exc:
+            except ValueError:
                 msg = (
                     "Please remove any duplicate listing of rule '%s' in "
                     "control '%s'." % (
@@ -403,7 +403,7 @@ def represent_as_dict(self):
         data["source"] = self.source
         data["definition_location"] = self.filepath
         data["controls"] = [c.represent_as_dict() for c in self.controls]
-        data["levels"] = [l.represent_as_dict() for l in self.levels]
+        data["levels"] = [level.represent_as_dict() for level in self.levels]
         return data
 
     @property
@@ -726,9 +726,9 @@ def get_level_with_ancestors_sequence(self, level_id):
         levels[level] = ""
         if level.inherits_from:
             for lv in level.inherits_from:
-                eligible_levels = [l for l in self.get_level_with_ancestors_sequence(lv) if l not in levels.keys()]
-                for l in eligible_levels:
-                    levels[l] = ""
+                eligible_levels = [le for le in self.get_level_with_ancestors_sequence(lv) if le not in levels.keys()]
+                for le in eligible_levels:
+                    levels[le] = ""
         return list(levels.keys())
 
     def _check_conflict_in_rules(self, rules):

ssg/entities/common.py

@@ -291,7 +291,7 @@ def from_yaml(cls, yaml_file, env_yaml=None, product_cpes=None):
 
         try:
             data_dict = cls.parse_yaml_into_processed_dict(yaml_file, local_env_yaml, product_cpes)
-        except DocumentationNotComplete as exc:
+        except DocumentationNotComplete:
             raise
         except Exception as exc:
             msg = (

ssg/playbook_builder.py

@@ -258,7 +258,7 @@ def build(self, profile_id=None, rule_id=None):
             profile_path = os.path.join(self.profiles_dir, profile_file)
             try:
                 profile = self.open_profile(profile_path)
-            except ssg.yaml.DocumentationNotComplete as e:
+            except ssg.yaml.DocumentationNotComplete:
                 msg = "Skipping incomplete profile {0}. To include incomplete " + \
                     "profiles, build in debug mode.\n"
                 sys.stderr.write(msg.format(profile_path))

ssg/yaml.py

@@ -190,7 +190,7 @@ def open_and_expand(yaml_file, substitutions_dict=None):
     expanded_template = process_file(yaml_file, substitutions_dict)
     try:
         yaml_contents = _open_yaml(expanded_template, yaml_file, substitutions_dict)
-    except yaml.scanner.ScannerError as e:
+    except yaml.scanner.ScannerError:
         print("A Jinja template expansion can mess up the indentation.")
         print("Please, check if the contents below are correctly expanded:")
         print("Source yaml: {}".format(yaml_file))

tests/ssg_test_suite/combined.py

@@ -31,8 +31,8 @@ class CombinedChecker(rule.RuleChecker):
     If a rule doesn't have any test scenario, it is skipped.
     Skipped rules are reported at the end.
     """
-    def __init__(self, test_env):
-        super(CombinedChecker, self).__init__(test_env)
+    def __init__(self, local_test_env):
+        super(CombinedChecker, self).__init__(local_test_env)
 
         self.rules_not_tested_yet = set()
         self.results = list()

tests/ssg_test_suite/oscap.py

@@ -44,7 +44,7 @@
 def analysis_to_serializable(analysis):
     result = dict(analysis)
     for key, value in analysis.items():
-        if type(value) == set:
+        if isinstance(value, set):
             result[key] = tuple(value)
     return result
 

tests/ssg_test_suite/rule.py

@@ -454,7 +454,7 @@ def _test_target(self):
         rules_to_test = self._get_rules_to_test()
         source = self.rule_spec
         if not self.rule_spec:
-            source = self.template_spec
+            source = self.template_spec  # noqa: F841
         if not rules_to_test:
             return
 

tests/ssg_test_suite/test_env.py

@@ -28,7 +28,7 @@ def map_on_top(self, function, args_list):
             current_running_state = self.environment.reset_state_to(
                 self.name, "running_%d" % idx)
             function(* args)
-        current_running_state = self.environment.reset_state_to(
+        current_running_state = self.environment.reset_state_to(  # noqa: F841
             self.name, "running_last")
 
     @classmethod
@@ -153,7 +153,7 @@ def scp_transfer_file(self, source, destination, log_file, error_msg=None):
                 "Failed to copy {source} to {destination}"
                 .format(source=source, destination=destination))
         try:
-            result = common.run_with_stdout_logging(
+            result = common.run_with_stdout_logging(  # noqa: F841
                 "scp", tuple(self.ssh_additional_options) + (source, destination), log_file)
         except Exception as exc:
             error_msg = error_msg + ": " + str(exc)
@@ -171,7 +171,7 @@ def reset_state_to(self, state_name, new_running_state_name):
 
     def save_state(self, state_name):
         self.running_state_base = common.get_prefixed_name(state_name)
-        running_state = self.running_state
+        running_state = self.running_state  # noqa: F841
         return self._save_state(state_name)
 
     def _delete_saved_state(self, state_name):
@@ -362,7 +362,7 @@ def get_ssh_port(self):
         if self.domain_ip == 'localhost':
             try:
                 ports = self._get_container_ports(self.current_container)
-            except Exception as exc:
+            except Exception:
                 msg = (
                     "Unable to extract SSH ports from the container. "
                     "This usually means that the container backend reported its configuration "

tests/test_machine_only_rules.py

@@ -14,7 +14,7 @@
     r'^.*\[ ! -f /.dockerenv \] && \[ ! -f /run/.containerenv \].*$', re.M)
 ANSIBLE_MACHINE_CONDITIONAL = re.compile(
     r'ansible_virtualization_type not in \["docker",\s+"lxc",\s+"openvz",\s+"podman",\s+' +
-    '"container"\]',
+    r'"container"\]',
     re.M)
 MACHINE_PLATFORM_ONE_LINE = re.compile(
     r'^\s*platform:\s+machine\s*$', re.M)

tests/test_product_stability.py

@@ -71,7 +71,6 @@ def corresponding_product_built(build_dir, product_id):
 
 
 def get_matching_compiled_product_filename(build_dir, product_id):
-    ref_path_components = reference_fname.split(os.path.sep)
     matching_filename = os.path.join(build_dir, product_id, "product.yml")
     if os.path.isfile(matching_filename):
         return matching_filename

tests/unit/ssg-module/test_build_yaml.py

@@ -578,4 +578,4 @@ def test_profile_with_version(profile_with_version):
 def test_profile_ospp_with_invalid_status():
     value_file = os.path.join(DATADIR, "ospp_invalid_status.profile")
     with pytest.raises(Exception):
-        profile = ssg.build_yaml.Profile.from_yaml(value_file)
+        profile = ssg.build_yaml.Profile.from_yaml(value_file)  # noqa: F841

tests/unit/ssg-module/test_controls.py

@@ -409,7 +409,7 @@ def test_policy_parse_from_nested(minimal_empty_controls, one_simple_subcontrol)
     assert len(controls) == 2
     control, subcontrol = order_by_attribute(controls, "id", ("c", "s"))
     assert control.title == "control"
-    assert control.selections == ["a"]
+    assert control.selections == ["a", "b"]
     assert subcontrol.title == "subcontrol"
     assert subcontrol.selections == ["b"]
 
@@ -445,7 +445,7 @@ def test_manager_removes_rules():
     assert len(control.selections) == 0
 
 
-def test_policy_parse_from_nested():
+def test_policy_parse_from_nested2():
     top_control_dict = dict(id="top", controls=["nested-1"])
     first_nested_dict = dict(id="nested-1", controls=["nested-2"], rules="Y")
     second_nested_dict = dict(id="nested-2", rules=["X"])
@@ -456,7 +456,7 @@ def test_policy_parse_from_nested():
     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])
+    controls = policy.save_controls_tree([top_control_dict, second_nested_dict, first_nested_dict])  # noqa: F841
     controls_manager.resolve_controls()
     control = policy.get_control("top")
     assert "Y" in control.selections

utils/add_kubernetes_rule.py

@@ -220,7 +220,7 @@ def createNodeRuleFunc(args):
     template['name'] = args.template
 
     template['vars'] = dict()
-    template_vars = set_template_vars(args.template_vars, template['vars'])
+    template_vars = set_template_vars(args.template_vars, template['vars'])  # noqa: F841
 
     save_rule(rule_yaml_path, rule_yaml)
     return 0

utils/ansible_shell_diff.py

@@ -53,7 +53,7 @@ def get_shell_tasks(tasks):
 
     diff = set(new_shell_modules) - set(old_shell_modules)
     if diff:
-        print(f"Changes in Ansible shell module have been found:")
+        print("Changes in Ansible shell module have been found:")
         print("\n".join(diff))
 
     Path.unlink('old.yml')

utils/build_control_from_reference.py

@@ -35,7 +35,7 @@ def _parse_args() -> argparse.Namespace:
                         help=f"YAML file with information about the build configuration. "
                              f"Defaults to {BUILD_CONFIG}")
     parser.add_argument("-o", "--output", type=str, required=True,
-                        help=f"Path to output the control file.")
+                        help="Path to output the control file.")
     return parser.parse_args()
 
 

utils/compare_versions.py

@@ -137,16 +137,16 @@ def _print_profile_comparison(
             rules_added: set, rules_removed: set,
             values_added: set, values_removed: set) -> None:
         if rules_added:
-            print(f"The following rules were added:")
+            print("The following rules were added:")
             print_set(rules_added)
         if rules_removed:
-            print(f"The following rules were removed:")
+            print("The following rules were removed:")
             print_set(rules_removed)
         if values_added:
-            print(f"The following values were added:")
+            print("The following values were added:")
             print_set(values_added)
         if values_removed:
-            print(f"The following values were removed:")
+            print("The following values were removed:")
             print_set(values_removed)
         print()
 

utils/create-stig-overlay.py

@@ -42,7 +42,7 @@ def element_value(element, element_obj):
         elem = elem.text
     try:
         return elem
-    except UnboundLocalError as e:
+    except UnboundLocalError:
         return ""
 
 
@@ -85,7 +85,7 @@ def new_stig_overlay(xccdftree, ssgtree, outfile, quiet):
     for group in xccdftree.findall("./{%s}Group" % XCCDF11_NS):
         vkey = group.get("id").strip('V-')
         for title in group.findall("./{%s}title" % XCCDF11_NS):
-            srg = title.text
+            srg = title.text  # noqa: F841
         for rule in group.findall("./{%s}Rule" % XCCDF11_NS):
             svkey_raw = rule.get("id")
             svkey = svkey_raw.strip()[3:9]
@@ -99,13 +99,13 @@ def new_stig_overlay(xccdftree, ssgtree, outfile, quiet):
         else:
             try:
                 mapped_id = ''.join(ssg_mapping[version].keys())
-            except KeyError as e:
+            except KeyError:
                 mapped_id = "XXXX"
 
         overlay = ET.SubElement(new_stig_overlay, "overlay", owner=owner,
                                 ruleid=mapped_id, ownerid=version,
                                 severity=severity)
-        vmsinfo = ET.SubElement(overlay, "VMSinfo", VKey=vkey,
+        vmsinfo = ET.SubElement(overlay, "VMSinfo", VKey=vkey,  # noqa: F841
                                 SVKey=svkey, VRelease=release)
         title = ET.SubElement(overlay, "title", text=rule_title)
 

utils/create_srg_export.py

@@ -455,7 +455,7 @@ def main() -> None:
     check_product_value_path(args.root, args.product)
 
     srgs = ssg.build_stig.parse_srgs(args.manual)
-    product_dir = os.path.join(args.root, "products", args.product)
+    product_dir = os.path.join(args.root, "products", args.product)  # noqa: F841
     env_yaml = get_env_yaml(
             args.root, args.product, args.build_config_yaml)
     policy = get_policy(args, env_yaml)