Fix/rsyslog forwarding

docs/changelog.md

@@ -2,6 +2,10 @@
 
 ## Latest
 
+## 2026.1.14.0
+
+- fix: base/remote_syslog/forwarder.py - improved idempotency and bugfix
+
 ## 2026.1.13.0
 
 - fix: base/ssl_certificates/personal_certificate.py - fails to import if the keystore does not contain any personal certificates yet

ibmsecurity/isam/base/remote_syslog/forwarder.py

@@ -56,7 +56,7 @@ def get(isamAppliance, server=None, port=None, protocol=None, id=None, check_mod
         return_obj = isamAppliance.create_return_object()
         return_obj['data'], i = _find_forwarder(ret_obj, server, port, protocol)
         warnings = []
-        if return_obj['data'] == None:
+        if return_obj['data'] is None:
             warnings.append(f"No entry found for server {server} port {port} and protocol {protocol}.")
             return_obj['warnings'] = warnings
 
@@ -117,6 +117,10 @@ def set(isamAppliance, server=None, port=None, protocol='udp', id=None, debug=Fa
     Update or create a specific remote syslog forwarder
     """
     warnings = []
+
+    if port and isinstance(port, basestring):
+        port = int(port)
+
     json_data = {
         'server': server,
         'port': port,
@@ -127,9 +131,6 @@ def set(isamAppliance, server=None, port=None, protocol='udp', id=None, debug=Fa
     if id is None:
         ret_obj = get_all(isamAppliance, check_mode, force)
 
-        if isinstance(port, basestring):
-            port = int(port)
-
         existing_forwarder, i = _find_forwarder(ret_obj, server, port, protocol)
         if existing_forwarder is not None and sources == [] and existing_forwarder['sources'] != sources:
             sources = existing_forwarder['sources']
@@ -154,23 +155,19 @@ def set(isamAppliance, server=None, port=None, protocol='udp', id=None, debug=Fa
             json_to_post = ret_obj['data']
             json_to_post.append(json_data)
             update_required = True
-            warnings.append("existing_forwarder is None")
+            warnings.append("This is a new forwarder")
         else:
             if format is not None:
                 if ibmsecurity.utilities.tools.version_compare(isamAppliance.facts['version'], "10.0.2.0") < 0:
                     warnings.append(f"Appliance at version: {isamAppliance.facts['version']}, format requires 10.0.2.0")
                 else:
                     json_data["format"] = format
-            elif 'format' in ret_obj['data'][i]:
-                del ret_obj['data'][i]['format']
-            sorted_json_data = tools.json_sort(json_data)
-            logger.debug(f"Sorted input: {sorted_json_data}")
-            sorted_ret_obj = tools.json_sort(existing_forwarder)
-            logger.debug(f"Sorted existing data: {sorted_ret_obj}")
-            if sorted_ret_obj != sorted_json_data:
-                logger.info("Changes detected, update needed.")
-                ret_obj['data'][i] = json_data
-                json_to_post = ret_obj['data']
+            else:
+                ret_obj['data'][i].pop('format', None)
+            existing_forwarder.pop('id', None)
+            if tools.json_equals(existing_forwarder, json_data, ignore_keys_not_in_new=True, skipkeys=True, sort_keys=True):
+                update_required = False
+            else:
                 update_required = True
 
         if update_required:

pyproject.toml

@@ -1,13 +1,13 @@
 [build-system]
 requires = [
-    "setuptools>=61",
+    "setuptools>=65.3.0",
     "wheel"
 ]
 build-backend = "setuptools.build_meta"
 
 [project]
 name = "ibmsecurity"
-version = "2026.1.13.0"
+version = "2026.1.14.0"
 authors = [
   { name="IBM", email="secorch@wwpdl.vnet.ibm.com" },
 ]

setup.py

@@ -5,7 +5,7 @@
     packages=find_packages(exclude=["test.*","test"]),
     # Date of release used for version - please be sure to use YYYY.MM.DD.seq#, MM and DD should be two digits e.g. 2017.02.05.0
     # seq# will be zero unless there are multiple release on a given day - then increment by one for additional release for that date
-    version="2026.1.13.0",
+    version="2026.1.14.0",
     description="Idempotent functions for IBM Verify Appliance REST APIs",
     author="IBM",
     author_email="secorch@wwpdl.vnet.ibm.com",

test/test_0_base_X_rsyslog_forwarding.py

@@ -0,0 +1,56 @@
+import logging
+import pytest
+
+import ibmsecurity.isam.base.remote_syslog.forwarder
+import ibmsecurity.isam.appliance
+
+def getTestData():
+    testdata = [
+        {
+          "port": "514",
+          "protocol": "tcp",
+          "server": "rsyslog",
+          "sources": [
+           {"facility": "local0",
+            "name": "WebSEAL:default:msg__webseald-default.log",
+            "severity": "info",
+            "tag": "webseal"},
+           {"facility": "local0",
+            "name": "WebSEAL:default:request.log",
+            "severity": "info",
+            "tag": "request"},
+            ]
+        },
+        {
+            "port": 6514,
+            "protocol": "udp",
+            "server": "rsyslog",
+            "sources": [
+                {"facility": "local0",
+                 "name": "WebSEAL:default:msg__webseald-default.log",
+                 "severity": "info",
+                 "tag": "webseal2"},
+                {"facility": "local0",
+                 "name": "WebSEAL:default:request.log",
+                 "severity": "info",
+                 "tag": "request2"},
+            ]
+        },
+    ]
+    return testdata
+
+
+@pytest.mark.parametrize("items", getTestData())
+def test_set_remote_syslog_forwarder(iviaServer, caplog, items) -> None:
+    """Get tracing protection"""
+    caplog.set_level(logging.DEBUG)
+    arg = {}
+    for k, v in items.items():
+        arg[k] = v
+
+    returnValue = ibmsecurity.isam.base.remote_syslog.forwarder.set(iviaServer,
+                                                     **arg
+                                                    )
+    logging.log(logging.INFO, returnValue)
+
+    assert not returnValue.failed()