Feature/ivia11 features (#442)

* refactor: add pylintrc

* refactor: update pylintrc to disable some checks for now

* fix: update container repo code (user and secret are NOT required values, although it is stated like that in the documentation)

* fix: add empty version (test)

* fix: update import submodules call (try)

* fix: make idempotent

* fix: small correction

* fix: update admin.py for new parameters in IVIA11

* fix: code improvement updates flake8

* feature: add label for v11

* fix: tiny update

* documentation: update versions and changelog

* documentation: update license related to PEP 639

Author: tombosmansibm

changelog.md

@@ -1,5 +1,16 @@
 # Manual change log
 
+## Latest
+
+
+## 2025.3.28.0
+
+- fix: base/extensions.py - improve idempotency #441
+- fix: base/container_ext/repo.py - user and secret are not required (although documentation states they are)
+- feature: base/admin.py - improve idempotency, support for new parameters in IVIA 11
+- feature: base/ssl_certificates/personal_certificate.py - support label parameter in IVIA 11
+- PEP 639: https://packaging.python.org/en/latest/guides/writing-pyproject-toml/#license
+
 ## 2025.3.14.0
 
 - build: add pylint configuration

ibmsecurity/isam/base/admin.py

@@ -1,4 +1,5 @@
 import logging
+from ibmsecurity.utilities import tools
 
 logger = logging.getLogger(__name__)
 
@@ -37,17 +38,17 @@ def get(isamAppliance, registry_id, check_mode=False, force=False):
 def add(
     isamAppliance,
     registry,
-    user,
-    secret,
+    user=None,
+    secret=None,
     **kwargs,
 ):
     """
     Add a credential for a user and container registry
 
     :param isamAppliance:
     :param registry: hostname of the registry, eg. icr.io
-    :param user: user
-    :param secret: the secret for the user
+    :param user: user (optional)
+    :param secret: the secret for the user (optional)
     :param proxy_host (optional)
     :param proxy_port (optional)
     :param proxy_user (optional)
@@ -99,12 +100,13 @@ def add(
 def update(
     isamAppliance,
     registry,
-    user,
-    secret,
+    user=None,
+    secret=None,
     **kwargs,
 ):
     """
     Update a credential for a user and container registry
+    (user and secret are NOT required values, although it is stated like that in the documentation)
 
     :param isamAppliance:
     :param registry: hostname of the registry, eg. icr.io
@@ -134,13 +136,26 @@ def update(
         if check_mode:
             return isamAppliance.create_return_object(changed=True, warnings=warnings)
         else:
-            put_data = {
-                "host": registry,
-                "user": user,
-                "secret": secret,
-            }
-
-            # put_data.update(input_args)
+            if tools.version_compare(isamAppliance.facts["version"], "10.0.9.0") < 0:
+                warnings.append(
+                    "Appliance is at version: {0}. Using the previous format for container repo.".format(
+                        isamAppliance.facts["version"]
+                    )
+                )
+                put_data = {
+                    "host": registry,
+                    "user": user,
+                }
+
+            else:
+                put_data = {
+                    "host": registry,
+                    "user": user,
+                }
+                # append other fields from input parameters
+                put_data.update(input_args)
+            if secret is not None:
+                put_data['secret'] = secret
 
             return isamAppliance.invoke_put(
                 "Update registry",
@@ -157,17 +172,18 @@ def update(
 def set(
     isamAppliance,
     registry,
-    user,
-    secret,
+    user=None,
+    secret=None,
     **kwargs,
 ):
     """
     Set - we can't make this function idempotent if a secret is set.
+    (user and secret are NOT required values, although it is stated like that in the documentation)
     TODO: execute every time ?
     """
     if _check(isamAppliance, registry, user):
         logger.debug(f"\nUpdating {registry} with {user}")
-
+        # TODO: make this idempotent
         return update(
             isamAppliance,
             registry,
@@ -185,7 +201,7 @@ def set(
             **kwargs)
 
 
-def search(isamAppliance, host, user, check_mode=False, force=False):
+def search(isamAppliance, host, user=None, check_mode=False, force=False):
     """
     Return the id of the repository
 
@@ -200,10 +216,16 @@ def search(isamAppliance, host, user, check_mode=False, force=False):
     return_obj = isamAppliance.create_return_object()
     return_obj["data"] = None
     for obj in ret_obj["data"]:
-        if obj.get("host", "") == host and obj.get("user", "") == user:
-            return_obj["data"] = obj["id"]
-            return_obj["rc"] = 0
-            break
+        if user is None:
+            if obj.get("host", "") == host:
+                return_obj["data"] = obj["id"]
+                return_obj["rc"] = 0
+                break
+        else:
+            if obj.get("host", "") == host and obj.get("user", "") == user:
+                return_obj["data"] = obj["id"]
+                return_obj["rc"] = 0
+                break
 
     return return_obj
 
@@ -223,10 +245,9 @@ def _check(isamAppliance, registry, user):
     return False
 
 
-def delete(isamAppliance, registry, user, check_mode=False, force=False):
+def delete(isamAppliance, registry, user=None, check_mode=False, force=False):
     """
-    Delete a repo by host and
-    ./testisam_cmd.py --hostname=${ISAM_LMI} --method=ibmsecurity.isam.base.container_ext.repo.delete --method_options="name=icr.io/isva/verify-access-oidc-provider:23.03"
+    Delete a repo by host (and user)
     """
     ret_obj = search(isamAppliance, registry, user)
     registry_id = ret_obj.get("data", None)

ibmsecurity/isam/base/container_ext/repo.py

@@ -28,17 +28,26 @@ def get_all(isamAppliance, check_mode=False, force=False):
     )
 
 
-def get(isamAppliance, extId, check_mode=False, force=False):
+def get(isamAppliance, extId, detailed=False, check_mode=False, force=False):
     """
     Retrieve installed extension by id
     """
-    ret_obj = isamAppliance.create_return_object()
-    extensions = get_all(isamAppliance)
-    for obj in extensions["data"]:
-        if obj["id"] == extId:
-            ret_obj["data"] = obj
-            break
-    return ret_obj
+    if detailed:
+        ret_obj = isamAppliance.invoke_get(
+            "Retrieve details about installed extension",
+            f"{uri}/{extId}",
+            requires_modules=requires_modules,
+            requires_version=requires_version,
+        )
+        return ret_obj
+    else:
+        ret_obj = isamAppliance.create_return_object()
+        extensions = get_all(isamAppliance)
+        for obj in extensions["data"]:
+            if obj["id"] == extId:
+                ret_obj["data"] = obj
+                break
+        return ret_obj
 
 
 def add(
@@ -137,6 +146,20 @@ def update(
             return isamAppliance.create_return_object(changed=True)
         else:
             config_str = _get_config_data(extId, config_data)
+            logger.debug("\nCONFIGDATA from input\n" + config_str)
+            if not force:
+                # Compare .  It's not possible to compare the extension (gui_json) itself, since we don't expect the extension here
+                currentConfigData = get(isamAppliance, extId=extId, detailed=True)
+                currentConfigData = currentConfigData.get('data', {'config_data': ''})
+                currentConfigData = currentConfigData.get('config_data', '')
+
+                currentConfigDataStr = json.dumps(currentConfigData, skipkeys=True, sort_keys=True)
+
+                logger.debug("\nCURRENT CONFIGDATA\n" + currentConfigDataStr)
+
+                if config_str == currentConfigDataStr:
+                    return isamAppliance.create_return_object(changed=False)
+
             files = {}
             files["config_data"] = (None, config_str)
 
@@ -183,14 +206,15 @@ def set(
 ):
 
     # MUST have an extId to do an update.
+    # Also, some extensions do not like updates, they require delete/add
     if extId and search(isamAppliance, extId):
         return update(
             isamAppliance=isamAppliance,
             extId=extId,
             config_data=config_data,
             third_party_package=third_party_package,
             check_mode=check_mode,
-            force=True,
+            force=force,
         )
     else:
         return add(
@@ -275,7 +299,7 @@ def _get_config_data(extId, config_data):
         return '{"extId": "' + extId + '",' + config_data + "}"
     else:
         config_data["extId"] = extId
-        return json.dumps(config_data)
+        return json.dumps(config_data, skipkeys=True, sort_keys=True)
 
 
 def search(isamAppliance, extId, check_mode=False, force=False):

ibmsecurity/isam/base/extensions.py

@@ -4,22 +4,18 @@
 requires_modules = ["mga", "federation"]
 requires_model = "Appliance"
 
+
 def get(isamAppliance, check_mode=False, force=False):
     """
     Get runtime tuning settings
     """
     return isamAppliance.invoke_get("Retrieving runtime tuning parameters",
                                     "/mga/runtime_tuning/v1",
-                                    requires_modules=requires_modules,requires_model=requires_model)
+                                    requires_modules=requires_modules,
+                                    requires_model=requires_model)
 
 
 def set(isamAppliance, option, value, check_mode=False, force=False):
-
-#    ret_obj = get(isamAppliance)
-#    for key, val in ret_obj.items():
-#        if key == 'warnings' and val != []:
-#            if "Docker" in val[0]:
-#                return isamAppliance.create_return_object(warnings=ret_obj['warnings'])
     """
     Set a runtime tuning parameter
     """
@@ -33,14 +29,14 @@ def set(isamAppliance, option, value, check_mode=False, force=False):
 
     if force is True or matches is False:
         if check_mode is True:
-            return isamAppliance.create_return_object(changed=True,warnings=warnings)
+            return isamAppliance.create_return_object(changed=True, warnings=warnings)
         else:
             return isamAppliance.invoke_put(
                 "Setting a runtime tuning parameter",
                 "/mga/runtime_tuning/{0}/v1".format(option),
                 {
                     'value': value
-                }, requires_modules=requires_modules,requires_model=requires_model)
+                }, requires_modules=requires_modules, requires_model=requires_model)
 
     return isamAppliance.create_return_object(warnings=warnings)
 
@@ -62,7 +58,7 @@ def _check(isamAppliance, option, value):
             logger.info(
                 "Tuning parameter {0} value does not match {1} != {2}.".format(option, ret_obj['data'][option], value))
         exists = True
-    except:
+    except Exception:
         logger.info("Runtime tuning parameter does not exist")
 
     return matches, exists, warnings
@@ -78,14 +74,14 @@ def reset(isamAppliance, option, check_mode=False, force=False):
     if force is False:
         matches, exists, warnings = _check(isamAppliance, option, None)
 
-    if force is True or exists is True:
+    if force is True or exists:
         if check_mode is True:
-            return isamAppliance.create_return_object(changed=True,warnings=warnings)
+            return isamAppliance.create_return_object(changed=True, warnings=warnings)
         else:
             return isamAppliance.invoke_delete(
                 "Reset a runtime tuning parameter to default value",
                 "/mga/runtime_tuning/{0}/v1".format(option),
-                requires_modules=requires_modules,requires_model=requires_model)
+                requires_modules=requires_modules, requires_model=requires_model)
     elif exists is False and warnings == []:
         warnings.append("Tuning Parameter {0} was not found. Skipping reset() request.".format(option))
 

ibmsecurity/isam/base/runtime/tuning_parameters.py

@@ -107,8 +107,8 @@ def set(isamAppliance, kdb_id, cert_id, default='no', check_mode=False, force=Fa
             f"Appliance is at version: {isamAppliance.facts['version']}. Setting certificates as default is no longer supported."
         )
     else:
-        if force is True or _check_default(isamAppliance, kdb_id, cert_id, default):
-            if check_mode is True:
+        if force or _check_default(isamAppliance, kdb_id, cert_id, default):
+            if check_mode:
                 return isamAppliance.create_return_object(changed=True)
             else:
                 return isamAppliance.invoke_put(
@@ -151,13 +151,13 @@ def delete(isamAppliance, kdb_id, cert_id, check_mode=False, force=False):
     """
     Deleting a personal certificate from a certificate database
     """
-    if force is True or _check(isamAppliance, kdb_id, cert_id) is True:
-        if check_mode is True:
+    if force or _check(isamAppliance, kdb_id, cert_id):
+        if check_mode:
             return isamAppliance.create_return_object(changed=True)
         else:
             return isamAppliance.invoke_delete(
                 "Deleting a personal certificate from a certificate database",
-                "/isam/ssl_certificates/{0}/personal_cert/{1}".format(kdb_id, cert_id))
+                f"/isam/ssl_certificates/{kdb_id}/personal_cert/{cert_id}")
 
     return isamAppliance.create_return_object()
 
@@ -179,26 +179,39 @@ def export_cert(isamAppliance, kdb_id, cert_id, filename, check_mode=False, forc
 def import_cert(isamAppliance, kdb_id, cert, label=None, password=None, check_mode=False, force=False):
     """
     Importing a personal certificate into a certificate database
-    Remark: you can add a label, but it's only used for a half-hearted check if the certificate already exists.
+    Remark: you can add a label, but it's only used for a half-hearted check if the certificate already exists for versions < 11
     """
-    if force is True or not _check(isamAppliance, kdb_id, label):
-        if check_mode is True:
+    warnings = []
+    if force or not _check(isamAppliance, kdb_id, label):
+        if check_mode:
             return isamAppliance.create_return_object(changed=True)
         else:
+            iviaVersion = isamAppliance.facts["version"]
+            post_data = {
+                'file_formfield': 'cert',
+                'filename': cert,
+                'mimetype': 'application/octet-stream'
+            }
+            json_data = {
+                'operation': 'import'
+            }
+            if password is not None:
+                json_data['password'] = password
+            if label is not None:
+                if ibmsecurity.utilities.tools.version_compare(iviaVersion, "10.0.9.0") < 0:
+                    warnings.append(f"Appliance at version: {iviaVersion}, label: {label} is not supported. Needs 10.0.9 or higher. Ignoring.")
+                else:
+                    logger.debug(f"Adding label: {label}")
+                    json_data['label'] = label
+            logger.debug(f"Posting json data: {json_data}")
             return isamAppliance.invoke_post_files(
                 "Importing a personal certificate into a certificate database",
-                "/isam/ssl_certificates/{0}/personal_cert".format(kdb_id),
+                f"/isam/ssl_certificates/{kdb_id}/personal_cert",
                 [
-                    {
-                        'file_formfield': 'cert',
-                        'filename': cert,
-                        'mimetype': 'application/octet-stream'
-                    }
+                    post_data
                 ],
-                {
-                    'password': password,
-                    'operation': 'import'
-                }
+                json_data,
+                warnings=warnings
             )
 
     return isamAppliance.create_return_object()