Bug/resource extract auditing

pyproject.toml

@@ -4,7 +4,7 @@
 
 [tool.poetry]
     name="pyracf"
-    version="1.0b3"
+    version="1.0b4"
     description="Python interface to RACF using IRRSMO00 RACF Callable Service."
     license = "Apache-2.0"
     authors = [

pyracf/common/security_admin.py

@@ -465,27 +465,9 @@ def __format_data_set_generic_profile_data(
             and messages[i + 1] is not None
             and ("-" in messages[i + 1])
         ):
-            field = " ".join(
-                [
-                    txt.lower().strip()
-                    for txt in list(filter(None, messages[i].split(" ")))
-                ]
+            self.__format_tabular_data(
+                messages, profile, current_segment, i, list_fields
             )
-            field = self._profile_field_to_camel_case(current_segment, field)
-            value = messages[i + 2]
-            if "(" in value:
-                value_tokens = value.split("(")
-                subfield = self._profile_field_to_camel_case(
-                    current_segment, value_tokens[0].lower()
-                )
-                profile[current_segment][field] = {
-                    subfield: self._clean_and_separate(value_tokens[-1].rstrip(")"))
-                }
-            elif field in list_fields:
-                profile[current_segment][field] = []
-                profile[current_segment][field].append(self._clean_and_separate(value))
-            else:
-                profile[current_segment][field] = self._clean_and_separate(value)
             i += 1
         elif "NO INSTALLATION DATA" in messages[i]:
             profile[current_segment]["installationData"] = None
@@ -668,6 +650,44 @@ def __format_semi_tabular_data(
                 messages[i + 2][indexes[j] : ind_e1]
             )
 
+    def __format_tabular_data(
+        self,
+        messages: List[str],
+        profile: dict,
+        current_segment: str,
+        i: int,
+        list_fields: List[str] = ["volumes"],
+    ) -> None:
+        field = " ".join(
+            [txt.lower().strip() for txt in list(filter(None, messages[i].split(" ")))]
+        )
+        field = self._profile_field_to_camel_case(current_segment, field)
+        values = (
+            [messages[i + 2]]
+            if "," not in messages[i + 2]
+            else messages[i + 2].split(",")
+        )
+        for value in values:
+            if "(" in value:
+                value_tokens = value.split("(")
+                subfield = self._profile_field_to_camel_case(
+                    current_segment, value_tokens[0].lower()
+                )
+                if field not in profile[current_segment]:
+                    profile[current_segment][field] = {
+                        subfield: self._clean_and_separate(value_tokens[-1].rstrip(")"))
+                    }
+                else:
+                    profile[current_segment][field][
+                        subfield
+                    ] = self._clean_and_separate(value_tokens[-1].rstrip(")"))
+            elif field in list_fields:
+                profile[current_segment][field] = []
+                profile[current_segment][field].append(self._clean_and_separate(value))
+            else:
+                profile[current_segment][field] = self._clean_and_separate(value)
+        return
+
     def __add_key_value_pairs_to_segment(
         self,
         segment_name: str,

pyracf/resource/resource_admin.py

@@ -1,5 +1,6 @@
 """General Resource Profile Administration."""
 
+from collections import Counter
 from typing import List, Union
 
 from pyracf.common.add_operation_error import AddOperationError
@@ -25,7 +26,7 @@ def __init__(
         self._valid_segment_traits = {
             "base": {
                 "base:application_data": "racf:appldata",
-                "base:audit_alter:": "racf:audaltr",
+                "base:audit_alter": "racf:audaltr",
                 "base:audit_control": "racf:audcntl",
                 "base:audit_none": "racf:audnone",
                 "base:audit_read": "racf:audread",
@@ -285,6 +286,135 @@ def get_user_access(
         self.set_running_userid(original_userid)
         return self._get_field(profile, "base", "yourAccess")
 
+    # ============================================================================
+    # Auditing Rules
+    # ============================================================================
+    def get_audit_rules(
+        self, resource: str, class_name: str
+    ) -> Union[dict, bytes, None]:
+        """Get the auditing rules associated with this general resource profile."""
+        profile = self.extract(resource, class_name, profile_only=True)
+        return self._get_field(profile, "base", "auditing")
+
+    def overwrite_audit_by_attempt(
+        self,
+        resource: str,
+        class_name: str,
+        success: Union[str, None] = None,
+        failure: Union[str, None] = None,
+        all: Union[str, None] = None,
+    ) -> Union[dict, bytes]:
+        """
+        Overwrites the auditing rules for this general resource profile with new
+        rules to audit based on specified access attempts.
+        """
+        self.__validate_access_levels(success, failure, all)
+        traits = {}
+        if success is not None:
+            traits[f"base:audit_{success}"] = "success"
+        if failure is not None:
+            traits[f"base:audit_{failure}"] = "failure"
+        if all is not None:
+            traits[f"base:audit_{all}"] = "all"
+        result = self.alter(resource, class_name, traits=traits)
+        return self._to_steps(result)
+
+    def alter_audit_by_attempt(
+        self,
+        resource: str,
+        class_name: str,
+        success: Union[str, None] = None,
+        failure: Union[str, None] = None,
+        all: Union[str, None] = None,
+    ) -> Union[dict, bytes]:
+        """
+        Alters the auditing rules for this general resource profile with new rules
+        to audit by access level, preserving existing non-conflicting rules.
+        """
+        self.__validate_access_levels(success, failure, all)
+        profile = self.extract(resource, class_name, profile_only=True)
+        audit_rules = self._get_field(profile, "base", "auditing")
+        traits = {}
+        if "success" in audit_rules:
+            traits[f"base:audit_{audit_rules['success']}"] = "success"
+        if "failures" in audit_rules:
+            traits[f"base:audit_{audit_rules['failures']}"] = "failure"
+        if "all" in audit_rules:
+            traits[f"base:audit_{audit_rules['all']}"] = "all"
+        if success is not None:
+            traits[f"base:audit_{success}"] = "success"
+        if failure is not None:
+            traits[f"base:audit_{failure}"] = "failure"
+        if all is not None:
+            traits[f"base:audit_{all}"] = "all"
+        result = self.alter(resource, class_name, traits=traits)
+        return self._to_steps(result)
+
+    def overwrite_audit_by_access_level(
+        self,
+        resource: str,
+        class_name: str,
+        alter: Union[str, None] = None,
+        control: Union[str, None] = None,
+        read: Union[str, None] = None,
+        update: Union[str, None] = None,
+    ) -> Union[dict, bytes]:
+        """
+        Overwrites the auditing rules for this general resource profile with new
+        rules to audit based on specified access levels.
+        """
+        traits = {}
+        if alter is not None:
+            traits["base:audit_alter"] = alter
+        if control is not None:
+            traits["base:audit_control"] = control
+        if read is not None:
+            traits["base:audit_read"] = read
+        if update is not None:
+            traits["base:audit_update"] = update
+        result = self.alter(resource, class_name, traits=traits)
+        return self._to_steps(result)
+
+    def alter_audit_by_access_level(
+        self,
+        resource: str,
+        class_name: str,
+        alter: Union[str, None] = None,
+        control: Union[str, None] = None,
+        read: Union[str, None] = None,
+        update: Union[str, None] = None,
+    ) -> Union[dict, bytes]:
+        """
+        Alters the auditing rules for this general resource profile with a new
+        rule to audit alter access, preserving existing non-conflicting rules.
+        """
+        profile = self.extract(resource, class_name, profile_only=True)
+        audit_rules = self._get_field(profile, "base", "auditing")
+        traits = {}
+        if "success" in audit_rules:
+            traits[f"base:audit_{audit_rules['success']}"] = "success"
+        if "failures" in audit_rules:
+            traits[f"base:audit_{audit_rules['failures']}"] = "failure"
+        if "all" in audit_rules:
+            traits[f"base:audit_{audit_rules['all']}"] = "all"
+        if alter is not None:
+            traits["base:audit_alter"] = alter
+        if control is not None:
+            traits["base:audit_control"] = control
+        if read is not None:
+            traits["base:audit_read"] = read
+        if update is not None:
+            traits["base:audit_update"] = update
+        result = self.alter(resource, class_name, traits=traits)
+        return self._to_steps(result)
+
+    def remove_all_audit_rules(
+        self, resource: str, class_name: str
+    ) -> Union[dict, bytes]:
+        """Clears the auditing rules completely."""
+        result = self.alter(resource, class_name, {"base:audit_none": True})
+        return self._to_steps(result)
+
     # ============================================================================
     # Class Administration
     # ============================================================================
@@ -589,3 +719,67 @@ def _format_profile(self, result: dict) -> None:
 
         del result["securityResult"]["resource"]["commands"][0]["messages"]
         result["securityResult"]["resource"]["commands"][0]["profiles"] = profiles
+
+    def __validate_access_levels(
+        self,
+        success: Union[str, None] = None,
+        failure: Union[str, None] = None,
+        all: Union[str, None] = None,
+    ):
+        valid_access_levels = ("alter", "control", "read", "update")
+        value_error_text = (
+            "Valid access levels include 'alter', 'control', 'read', and 'update'."
+        )
+        bad_access_levels = []
+        for attempt_argument in (success, failure, all):
+            if (
+                attempt_argument is not None
+                and str(attempt_argument).lower() not in valid_access_levels
+            ):
+                bad_access_levels.append(attempt_argument)
+        match len(bad_access_levels):
+            case 0:
+                self.__check_for_duplicates(
+                    [success, failure, all], "attempt types", "access level"
+                )
+                return
+            case 1:
+                value_error_text = (
+                    f"'{bad_access_levels[0]}' is not a valid access level. "
+                    + f"{value_error_text}"
+                )
+            case 2:
+                value_error_text = (
+                    f"'{bad_access_levels[0]}' and '{bad_access_levels[1]}' are not valid "
+                    + f"access levels. {value_error_text}"
+                )
+            case _:
+                bad_access_levels = [
+                    f"'{bad_access_level}'" for bad_access_level in bad_access_levels
+                ]
+                bad_access_levels[-1] = f"and {bad_access_levels[-1]} "
+                value_error_text = (
+                    f"{', '.join(bad_access_levels)}are not valid access levels. "
+                    + f"{value_error_text}"
+                )
+        raise ValueError(value_error_text)
+
+    def __check_for_duplicates(
+        self, input_list: list, argument: str, value: str
+    ) -> None:
+        duplicates = [
+            key
+            for (key, value) in Counter(input_list).items()
+            if (value > 1 and not (key is None))
+        ]
+        if duplicates == []:
+            return
+        value_error_text = []
+        for duplicate in duplicates:
+            value_error_text.append(
+                f"You entered '{duplicate}' for multiple parameters which is not supported."
+            )
+        value_error_text.append(
+            f"Please use multiple function calls to set multiple {argument} to the same {value}."
+        )
+        raise ValueError("\n".join(value_error_text))

tests/resource/resource_log_samples/alter_resource_error.log

@@ -48,7 +48,7 @@
       <message> </message>
       <message>AUDITING</message>
       <message>--------</message>
-      <message>FAILURES(READ)</message>
+      <message>SUCCESS(UPDATE),FAILURES(READ)</message>
       <message> </message>
       <message>NOTIFY</message>
       <message>------</message>
@@ -97,7 +97,7 @@
             " ",
             "AUDITING",
             "--------",
-            "FAILURES(READ)",
+            "SUCCESS(UPDATE),FAILURES(READ)",
             " ",
             "NOTIFY",
             "------",
@@ -144,6 +144,7 @@
                 "installationData": null,
                 "applicationData": null,
                 "auditing": {
+                  "success": "update",
                   "failures": "read"
                 },
                 "notify": null,

tests/resource/resource_log_samples/alter_resource_success.log

@@ -48,7 +48,7 @@
       <message> </message>
       <message>AUDITING</message>
       <message>--------</message>
-      <message>FAILURES(READ)</message>
+      <message>SUCCESS(UPDATE),FAILURES(READ)</message>
       <message> </message>
       <message>NOTIFY</message>
       <message>------</message>
@@ -97,7 +97,7 @@
             " ",
             "AUDITING",
             "--------",
-            "FAILURES(READ)",
+            "SUCCESS(UPDATE),FAILURES(READ)",
             " ",
             "NOTIFY",
             "------",
@@ -144,6 +144,7 @@
                 "installationData": null,
                 "applicationData": null,
                 "auditing": {
+                  "success": "update",
                   "failures": "read"
                 },
                 "notify": null,

tests/resource/resource_log_samples/extract_resource_base_success.log

@@ -48,7 +48,7 @@
       <message> </message>
       <message>AUDITING</message>
       <message>--------</message>
-      <message>FAILURES(READ)</message>
+      <message>SUCCESS(UPDATE),FAILURES(READ)</message>
       <message> </message>
       <message>NOTIFY</message>
       <message>------</message>
@@ -97,7 +97,7 @@
             " ",
             "AUDITING",
             "--------",
-            "FAILURES(READ)",
+            "SUCCESS(UPDATE),FAILURES(READ)",
             " ",
             "NOTIFY",
             "------",
@@ -144,6 +144,7 @@
                 "installationData": null,
                 "applicationData": null,
                 "auditing": {
+                  "success": "update",
                   "failures": "read"
                 },
                 "notify": null,

tests/resource/resource_request_samples/resource_alter_audit_by_access_level.xml

@@ -0,0 +1,9 @@
+<securityrequest xmlns="http://www.ibm.com/systems/zos/saf" xmlns:racf="http://www.ibm.com/systems/zos/racf">
+  <resource name="TESTING" class="ELIJTEST" operation="set" requestid="ResourceRequest">
+    <base>
+      <racf:audupdt operation="set">success</racf:audupdt>
+      <racf:audread operation="set">failure</racf:audread>
+      <racf:audaltr operation="set">success</racf:audaltr>
+    </base>
+  </resource>
+</securityrequest>