Unify nextcloud/sqlite/gpg_key/ipa_diff plugins + safe fixes + tests

CHANGELOG.md

@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Security
 
+* **plugin:gpg_key**: The cleartext passphrase is no longer included in the module's failure output when key generation fails.
 * **role:repo_\***: HTTP basic auth credentials are now only written to the repository config files when a custom mirror URL is set. Previously, setting `lfops__repo_basic_auth_login` without `lfops__repo_mirror_url` wrote the credentials into repo files that still pointed at the public vendor mirrors, causing the package manager to send them to servers that do not use basic auth. The Icinga repo is intentionally unchanged, since its subscription URL legitimately requires basic auth.
 * **ci**: Scope `GITHUB_TOKEN` permissions in the dependabot-auto-merge workflow to the job level, with top-level now `read-all`. Matches the pattern used by the other LFOps workflows and addresses the OpenSSF Scorecard `Token-Permissions` finding.
 
@@ -42,6 +43,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+* **plugin:sqlite_query**: A `REGEXP` query against a column that contains NULL values no longer fails; a NULL value simply does not match.
 * **plugin:uptimerobot_\***: The modules no longer crash when the UptimeRobot API returns a non-list response for a list endpoint; the response is passed through instead.
 * **plugin:nextcloud_occ_app_config, plugin:nextcloud_occ_system_config, plugin:uptimerobot_monitor, plugin:uptimerobot_psp**: Fixed their documentation so `ansible-doc` renders them again. A unit-test guard now catches this class of error for every in-house plugin.
 * **plugin:bitwarden_item**: Fixed the lookup's documentation so `ansible-doc` renders it again.

plugins/module_utils/ipa_diff.py

@@ -1,12 +1,20 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8; py-indent-offset: 4 -*-
+#
+# Author:  Linuxfabrik GmbH, Zurich, Switzerland
+# Contact: info (at) linuxfabrik (dot) ch
+#          https://www.linuxfabrik.ch/
+# License: The Unlicense, see LICENSE file.
+
 # Temporary diff helpers for ansible-freeipa modules.
 # Remove once https://github.com/freeipa/ansible-freeipa/pull/1415
 # is merged and released.
 
-from __future__ import (absolute_import, division, print_function)
+from __future__ import absolute_import, division, print_function
 
 __metaclass__ = type
 
-from ansible.module_utils._text import to_text
+from ansible.module_utils.common.text.converters import to_text
 
 
 def _compare_key(arg, ipa_arg):
@@ -28,7 +36,7 @@ def _compare_key(arg, ipa_arg):
     return arg == ipa_arg
 
 
-class IPADiffTracker(object):
+class IPADiffTracker:
     """Track before/after state for Ansible --diff output."""
 
     def __init__(self):
@@ -38,17 +46,17 @@ def build_diff(self):
         """Return kwargs for exit_json (empty dict if no changes)."""
         if not self._diffs:
             return {}
-        return {"diff": self._diffs}
+        return {'diff': self._diffs}
 
     def add_entry_diff(self, name, before, after):
         """Record a diff entry for one IPA object."""
         if before == after:
             return
         self._diffs.append({
-            "before_header": name,
-            "after_header": name,
-            "before": before,
-            "after": after,
+            'before_header': name,
+            'after_header': name,
+            'before': before,
+            'after': after,
         })
 
 

plugins/modules/gpg_key.py

@@ -1,10 +1,12 @@
-#!/usr/bin/python
-# -*- coding: utf-8 -*-
+#!/usr/bin/env python3
+# -*- coding: utf-8; py-indent-offset: 4 -*-
+#
+# Author:  Linuxfabrik GmbH, Zurich, Switzerland
+# Contact: info (at) linuxfabrik (dot) ch
+#          https://www.linuxfabrik.ch/
+# License: The Unlicense, see LICENSE file.
 
-# Copyright: (c) 2022, Linuxfabrik GmbH, Zurich, Switzerland, https://www.linuxfabrik.ch
-# The Unlicense (see LICENSE or https://unlicense.org/)
-
-from __future__ import (absolute_import, division, print_function)
+from __future__ import absolute_import, division, print_function
 
 __metaclass__ = type
 
@@ -244,14 +246,14 @@
 import re
 import traceback
 
-logger = logging.getLogger('gnupg')
-logger.setLevel(logging.DEBUG)
-
 from ansible.module_utils.basic import AnsibleModule
-from ansible.module_utils._text import to_native
+from ansible.module_utils.common.text.converters import to_native
 
 from ansible_collections.linuxfabrik.lfops.plugins.module_utils.gnupg import GPG
 
+logger = logging.getLogger('gnupg')
+logger.setLevel(logging.DEBUG)
+
 # taken from https://www.iana.org/assignments/pgp-parameters/pgp-parameters.xhtml#pgp-parameters-12
 algo_ids = {
     1: 'RSA',
@@ -362,12 +364,6 @@ def run_module():
         supports_check_mode=True
     )
 
-    # if debug
-    # console_logger = logging.StreamHandler()
-    # console_logger.setLevel(logging.DEBUG)
-    # console_logger.setFormatter(logging.Formatter('%(levelname)s: %(message)s'))
-    # logger.addHandler(console_logger)
-
     gnupghome = module.params['gnupghome']
     if gnupghome and not os.path.isdir(gnupghome):
 
@@ -384,7 +380,7 @@ def run_module():
             gnupghome=gnupghome,
         )
     except (OSError, ValueError) as e:
-        module.fail_json(msg='There was an error executing gpg: {}'.format(to_native(e)), exception=traceback.format_exc(), **result)
+        module.fail_json(msg=f'There was an error executing gpg: {to_native(e)}', exception=traceback.format_exc(), **result)
 
     # use whatever logic you need to determine whether or not this module
     # made any modifications to your target
@@ -425,11 +421,10 @@ def run_module():
         params['no_protection'] = True
 
     input_data = gpg.gen_key_input(**params)
-    # print(params)
-    # print(input_data)
     new_key = gpg.gen_key(input_data)
     if not new_key:
-        module.fail_json(msg='Failed to generate a new key.', rc=new_key.returncode, stdout=new_key.data, stderr=new_key.stderr, input_data=input_data, **result)
+        # do not echo input_data here: it contains the cleartext passphrase
+        module.fail_json(msg='Failed to generate a new key.', rc=new_key.returncode, stdout=new_key.data, stderr=new_key.stderr, **result)
 
     # list the keys again, as we only got the fingerprint from gen_key()
     keys = gpg.list_keys(secret=True)

plugins/modules/nextcloud_occ_app.py

@@ -1,8 +1,10 @@
-#!/usr/bin/python
-# -*- coding: utf-8 -*-
-
-# Copyright: (c) 2026, Linuxfabrik GmbH, Zurich, Switzerland, https://www.linuxfabrik.ch
-# The Unlicense (see LICENSE or https://unlicense.org/)
+#!/usr/bin/env python3
+# -*- coding: utf-8; py-indent-offset: 4 -*-
+#
+# Author:  Linuxfabrik GmbH, Zurich, Switzerland
+# Contact: info (at) linuxfabrik (dot) ch
+#          https://www.linuxfabrik.ch/
+# License: The Unlicense, see LICENSE file.
 
 from __future__ import absolute_import, division, print_function
 

plugins/modules/nextcloud_occ_app_config.py

@@ -1,8 +1,10 @@
-#!/usr/bin/python
-# -*- coding: utf-8 -*-
-
-# Copyright: (c) 2026, Linuxfabrik GmbH, Zurich, Switzerland, https://www.linuxfabrik.ch
-# The Unlicense (see LICENSE or https://unlicense.org/)
+#!/usr/bin/env python3
+# -*- coding: utf-8; py-indent-offset: 4 -*-
+#
+# Author:  Linuxfabrik GmbH, Zurich, Switzerland
+# Contact: info (at) linuxfabrik (dot) ch
+#          https://www.linuxfabrik.ch/
+# License: The Unlicense, see LICENSE file.
 
 from __future__ import absolute_import, division, print_function
 
@@ -128,10 +130,6 @@ def main():
             installed_config_json=dict(type='raw'),
     )
 
-    # the AnsibleModule object will be our abstraction working with Ansible
-    # this includes instantiation, a couple of common attr would be the
-    # args/params passed to the execution, as well as if this module
-    # supports check mode
     module = AnsibleModule(
         argument_spec=module_args,
         supports_check_mode=True,
@@ -169,7 +167,7 @@ def main():
             try:
                 installed_config_json = json.loads(installed_config_json)
             except (json.JSONDecodeError, ValueError):
-                module.fail_json(msg=f'Failed to parse installed_config_json')
+                module.fail_json(msg='Failed to parse installed_config_json')
 
         app_configs = installed_config_json.get('apps', {}).get(app, {})
         key_exists = name in app_configs

plugins/modules/nextcloud_occ_system_config.py

@@ -1,8 +1,10 @@
-#!/usr/bin/python
-# -*- coding: utf-8 -*-
-
-# Copyright: (c) 2026, Linuxfabrik GmbH, Zurich, Switzerland, https://www.linuxfabrik.ch
-# The Unlicense (see LICENSE or https://unlicense.org/)
+#!/usr/bin/env python3
+# -*- coding: utf-8; py-indent-offset: 4 -*-
+#
+# Author:  Linuxfabrik GmbH, Zurich, Switzerland
+# Contact: info (at) linuxfabrik (dot) ch
+#          https://www.linuxfabrik.ch/
+# License: The Unlicense, see LICENSE file.
 
 from __future__ import absolute_import, division, print_function
 
@@ -119,10 +121,6 @@ def main():
             installed_config_json=dict(type='raw'),
     )
 
-    # the AnsibleModule object will be our abstraction working with Ansible
-    # this includes instantiation, a couple of common attr would be the
-    # args/params passed to the execution, as well as if this module
-    # supports check mode
     module = AnsibleModule(
         argument_spec=module_args,
         supports_check_mode=True,
@@ -158,7 +156,7 @@ def main():
             try:
                 installed_config_json = json.loads(installed_config_json)
             except (json.JSONDecodeError, ValueError):
-                module.fail_json(msg=f'Failed to parse installed_config_json')
+                module.fail_json(msg='Failed to parse installed_config_json')
 
         # navigate nested config by path parts (e.g. "trusted_domains 0")
         current = installed_config_json.get('system', {})

plugins/modules/sqlite_query.py

@@ -1,8 +1,10 @@
-#!/usr/bin/python
-# -*- coding: utf-8 -*-
-
-# Copyright: (c) 2026, Linuxfabrik GmbH, Zurich, Switzerland, https://www.linuxfabrik.ch
-# The Unlicense (see LICENSE or https://unlicense.org/)
+#!/usr/bin/env python3
+# -*- coding: utf-8; py-indent-offset: 4 -*-
+#
+# Author:  Linuxfabrik GmbH, Zurich, Switzerland
+# Contact: info (at) linuxfabrik (dot) ch
+#          https://www.linuxfabrik.ch/
+# License: The Unlicense, see LICENSE file.
 
 from __future__ import absolute_import, division, print_function
 
@@ -107,14 +109,14 @@
 '''
 
 
-from ansible.module_utils.basic import AnsibleModule
+import os
+import re
+import sqlite3
 
+from ansible.module_utils.basic import AnsibleModule
 
-# all sqlite functions taken from
+# the close / connect / select / regexp helpers below are taken from
 # https://git.linuxfabrik.ch/linuxfabrik/lib/-/blob/master/db_mysql3.py
-import os
-import sqlite3
-import re
 
 
 def close(conn):
@@ -124,7 +126,7 @@ def close(conn):
     """
     try:
         conn.close()
-    except:
+    except Exception:
         pass
     return True
 
@@ -143,16 +145,18 @@ def connect(path='', filename=''):
         conn.text_factory = str
         conn.create_function("REGEXP", 2, regexp)
     except Exception as e:
-        return(False, 'Connecting to DB {} failed, Error: {}, CWD: {}'.format(db, e, os.getcwd()))
+        return (False, f'Connecting to DB {db} failed, Error: {e}, CWD: {os.getcwd()}')
     return (True, conn)
 
 
-def select(conn, sql, data={}, fetchone=False, as_dict=True):
+def select(conn, sql, data=None, fetchone=False, as_dict=True):
     """The SELECT statement is used to query the database. The result of a
     SELECT is zero or more rows of data where each row has a fixed number
     of columns. A SELECT statement does not make any changes to the
     database.
     """
+    if data is None:
+        data = {}
     c = conn.cursor()
     try:
         if data:
@@ -171,7 +175,7 @@ def select(conn, sql, data={}, fetchone=False, as_dict=True):
             return (True,  c.fetchone())
         return (True, c.fetchall())
     except Exception as e:
-        return(False, 'Query failed: {}, Error: {}, Data: {}'.format(sql, e, data))
+        return (False, f'Query failed: {sql}, Error: {e}, Data: {data}')
 
 
 def regexp(expr, item):
@@ -180,6 +184,9 @@ def regexp(expr, item):
     For Python, you have to implement REGEXP using a Python function at runtime.
     https://stackoverflow.com/questions/5365451/problem-with-regexp-python-and-sqlite/5365533#5365533
     """
+    if item is None:
+        # a NULL column value cannot match a regex (and re.search(None) raises)
+        return False
     reg = re.compile(expr)
     return reg.search(item) is not None
 
@@ -215,7 +222,7 @@ def main():
 
     success, conn = connect(path=path, filename=db)
     if not success:
-        module.fail_json(msg='Unable to connect to database: {}'.format(conn))
+        module.fail_json(msg=f'Unable to connect to database: {conn}')
 
     query_result = []
     if query_type == 'select':