Add external auth helper support (OIDC/OAuth2, cookies, etc.)

Nagstamon/config.py

@@ -120,7 +120,9 @@
                   'autologin_key',
                   'custom_cert_ca_file',
                   'idp_ecp_endpoint',
-                  'monitor_site'
+                  'monitor_site',
+                  'auth_helper_command',
+                  'auth_helper_extra_args'
                   ]
 
 
@@ -694,35 +696,24 @@ def save_multiple_config(self, settingsdir, setting):
                                 value = ''
                             elif self.keyring_available and self.use_system_keyring:
                                 if self.__dict__[settingsdir][s].password != '':
-                                    # provoke crash if password saving does not work - this is the case
-                                    # on newer Ubuntu releases
-                                    try:
-                                        keyring.set_password('Nagstamon',
-                                                             '@'.join((self.__dict__[settingsdir][s].username,
-                                                                       self.__dict__[settingsdir][s].monitor_url)),
-                                                             self.__dict__[settingsdir][s].password)
-                                    except Exception:
-                                        import traceback
-                                        traceback.print_exc(file=sys.stdout)
-                                        sys.exit(1)
-                                value = ''
+                                    keyring_account = '@'.join((self.__dict__[settingsdir][s].username,
+                                                                self.__dict__[settingsdir][s].monitor_url))
+                                    if self._keyring_set('Nagstamon', keyring_account,
+                                                         self.__dict__[settingsdir][s].password):
+                                        value = ''
+                                else:
+                                    value = ''
                         if option == 'proxy_password':
                             if self.keyring_available and self.use_system_keyring:
                                 if self.__dict__[settingsdir][s].proxy_password != '':
-                                    # provoke crash if password saving does not work - this is the case
-                                    # on newer Ubuntu releases
-                                    try:
-                                        keyring.set_password('Nagstamon',
-                                                             '@'.join(('proxy',
-                                                                       self.__dict__[settingsdir][s].proxy_username,
-                                                                       self.__dict__[settingsdir][s].proxy_address)),
-                                                             self.__dict__[settingsdir][s].proxy_password)
-                                    except Exception:
-                                        import traceback
-                                        traceback.print_exc(file=sys.stdout)
-                                        sys.exit(1)
-
-                                value = ''
+                                    keyring_account = '@'.join(('proxy',
+                                                                self.__dict__[settingsdir][s].proxy_username,
+                                                                self.__dict__[settingsdir][s].proxy_address))
+                                    if self._keyring_set('Nagstamon', keyring_account,
+                                                         self.__dict__[settingsdir][s].proxy_password):
+                                        value = ''
+                                else:
+                                    value = ''
                         config.set(setting + '_' + s, option, str(value))
                     else:
                         config.set(setting + '_' + s, option, str(self.__dict__[settingsdir][s].__dict__[option]))
@@ -749,6 +740,40 @@ def save_multiple_config(self, settingsdir, setting):
             # ##        if not f.split(setting + "_")[1].split(".conf")[0] in self.__dict__[settingsdir]:
             # ##            os.unlink(self.configdir + os.sep + settingsdir + os.sep + f)
 
+    def _keyring_set(self, service, account, password):
+        """
+        Store a password in the system keyring.
+
+        On macOS, handles errSecDuplicateItem (-25299) by deleting the old
+        entry and retrying once.
+
+        Crashes the application if keyring storage ultimately fails, to prevent
+        insecure fallback to obfuscated config file storage (e.g. on newer
+        Ubuntu releases where keyring backends may be broken).
+        """
+        import keyring
+        try:
+            keyring.set_password(service, account, password)
+            return True
+        except Exception:
+            # On macOS, Keychain may return errSecDuplicateItem (-25299).
+            # Delete the existing entry and retry once.
+            if OS == OS_MACOS:
+                try:
+                    keyring.delete_password(service, account)
+                except Exception:
+                    pass
+                try:
+                    keyring.set_password(service, account, password)
+                    return True
+                except Exception:
+                    pass
+            # provoke crash if password saving does not work, this is the case
+            # on newer Ubuntu releases
+            import traceback
+            traceback.print_exc(file=sys.stdout)
+            sys.exit(1)
+
     def is_keyring_available(self):
         """
         Determines if the keyring module and a suitable backend are available for secure password storage.
@@ -1131,6 +1156,11 @@ def __init__(self):
         # LibreNMS
         self.treat_services_as_alerts = False
 
+        # Auth helper - external command for authentication (OIDC, cookies, etc.)
+        self.use_auth_helper = False
+        self.auth_helper_command = ''
+        self.auth_helper_extra_args = ''
+
 
 class Action:
     """

Nagstamon/qui/dialogs/server.py

@@ -79,7 +79,11 @@ def __init__(self):
             self.window.input_checkbox_show_options: [self.window.groupbox_options],
             self.window.input_checkbox_custom_cert_use: [self.window.label_custom_ca_file,
                                                          self.window.input_lineedit_custom_cert_ca_file,
-                                                         self.window.button_choose_custom_cert_ca_file]}
+                                                         self.window.button_choose_custom_cert_ca_file],
+            self.window.input_checkbox_use_auth_helper: [self.window.label_auth_helper_command,
+                                                         self.window.input_lineedit_auth_helper_command,
+                                                         self.window.label_auth_helper_extra_args,
+                                                         self.window.input_lineedit_auth_helper_extra_args]}
 
         self.TOGGLE_DEPS_INVERTED = [self.window.input_checkbox_use_proxy_from_os]
 
@@ -185,6 +189,9 @@ def __init__(self):
         # when authentication is changed to Kerberos then disable username/password as they are now useless
         self.window.input_combobox_authentication.activated.connect(self.toggle_authentication)
 
+        # when auth helper is toggled, hide/show username/password accordingly
+        self.window.input_checkbox_use_auth_helper.toggled.connect(self.toggle_authentication)
+
         # reset Checkmk views
         self.window.button_checkmk_view_hosts_reset.clicked.connect(self.checkmk_view_hosts_reset)
         self.window.button_checkmk_view_services_reset.clicked.connect(self.checkmk_view_services_reset)
@@ -243,6 +250,16 @@ def toggle_authentication(self):
                 widget.show()
             self.window.button_delete_web_cookies.hide()
 
+        # no need for username + password or authentication type when using auth helper
+        if self.window.input_checkbox_use_auth_helper.isChecked():
+            for widget in self.AUTHENTICATION_WIDGETS:
+                widget.hide()
+            self.window.label_auth_type.hide()
+            self.window.input_combobox_authentication.hide()
+        else:
+            self.window.label_auth_type.show()
+            self.window.input_combobox_authentication.show()
+
         # after hiding authentication widgets dialog might shrink
         self.window.adjustSize()
 

Nagstamon/qui/widgets/statuswindow.py

@@ -595,6 +595,7 @@ def create_server_vbox(self, name):
             # display authentication dialog if password is not known
             if not conf.servers[server.name].save_password and \
                     not conf.servers[server.name].use_autologin and \
+                    not getattr(conf.servers[server.name], 'use_auth_helper', False) and \
                     conf.servers[server.name].password == '' and \
                     not conf.servers[server.name].authentication == 'kerberos' and \
                     not conf.servers[server.name].authentication == 'web':

Nagstamon/resources/qui/settings_server.ui

@@ -788,6 +788,53 @@
         </property>
        </widget>
       </item>
+      <item row="38" column="1" colspan="4">
+       <widget class="QCheckBox" name="input_checkbox_use_auth_helper">
+        <property name="text">
+         <string>Use auth helper for authentication</string>
+        </property>
+       </widget>
+      </item>
+      <item row="39" column="1">
+       <widget class="QLabel" name="label_auth_helper_command">
+        <property name="sizePolicy">
+         <sizepolicy hsizetype="Minimum" vsizetype="Preferred">
+          <horstretch>0</horstretch>
+          <verstretch>0</verstretch>
+         </sizepolicy>
+        </property>
+        <property name="text">
+         <string>Helper command:</string>
+        </property>
+       </widget>
+      </item>
+      <item row="39" column="2" colspan="3">
+       <widget class="QLineEdit" name="input_lineedit_auth_helper_command">
+        <property name="placeholderText">
+         <string>e.g., uvx nagstamon-auth-helper</string>
+        </property>
+       </widget>
+      </item>
+      <item row="40" column="1">
+       <widget class="QLabel" name="label_auth_helper_extra_args">
+        <property name="sizePolicy">
+         <sizepolicy hsizetype="Minimum" vsizetype="Preferred">
+          <horstretch>0</horstretch>
+          <verstretch>0</verstretch>
+         </sizepolicy>
+        </property>
+        <property name="text">
+         <string>Extra arguments:</string>
+        </property>
+       </widget>
+      </item>
+      <item row="40" column="2" colspan="3">
+       <widget class="QLineEdit" name="input_lineedit_auth_helper_extra_args">
+        <property name="placeholderText">
+         <string>e.g., --client-id nagstamon --redirect-port 12345</string>
+        </property>
+       </widget>
+      </item>
      </layout>
     </widget>
    </item>
@@ -875,6 +922,9 @@
   <tabstop>button_checkmk_view_services_reset</tabstop>
   <tabstop>input_lineedit_idp_ecp_endpoint</tabstop>
   <tabstop>input_lineedit_disabled_backends</tabstop>
+  <tabstop>input_checkbox_use_auth_helper</tabstop>
+  <tabstop>input_lineedit_auth_helper_command</tabstop>
+  <tabstop>input_lineedit_auth_helper_extra_args</tabstop>
  </tabstops>
  <resources/>
  <connections/>

Nagstamon/servers/Generic.py

@@ -22,7 +22,9 @@
 import json
 from pathlib import Path
 import platform
+import shlex
 import socket
+import subprocess
 import sys
 import traceback
 import urllib.parse
@@ -357,6 +359,10 @@ def create_session(self):
             cookies = load_cookies()
             session.cookies = cookie_data_to_jar(self.name, cookies)
 
+        # Auth helper provides its own credentials, skip built-in auth
+        if getattr(self, 'use_auth_helper', False) and getattr(self, 'auth_helper_command', ''):
+            session.auth = None
+
         # default to check TLS validity
         if self.ignore_cert:
             session.verify = False
@@ -413,6 +419,124 @@ def reset_http(self):
         if self.authentication != 'web':
             self.session = None
 
+    def _run_auth_helper(self, args):
+        """
+        Run the auth helper command with given arguments.
+        Returns (stdout, stderr, returncode) tuple.
+        """
+        cmd = shlex.split(self.auth_helper_command) + args
+        try:
+            result = subprocess.run(cmd, capture_output=True, text=True, timeout=180)
+            if conf.debug_mode and result.stderr:
+                self.debug(server=self.get_name(), debug=f'[Auth helper stderr] {result.stderr.strip()}')
+            return result.stdout, result.stderr, result.returncode
+        except FileNotFoundError:
+            return '', f'Auth helper command not found: {self.auth_helper_command}', -1
+        except subprocess.TimeoutExpired:
+            return '', 'Auth helper command timed out', -1
+        except Exception as e:
+            return '', f'Auth helper error: {e}', -1
+
+    def _get_auth_helper_credentials(self):
+        """
+        Call the external auth helper to get HTTP credentials (headers and/or cookies).
+        Returns a dict with 'headers' and optionally 'cookies' on success,
+        or None on failure (sets status_description).
+        Automatically triggers re-authentication if the helper signals exit code 1.
+        """
+        stdout, stderr, rc = self._run_auth_helper([
+            'get-headers', '--server-name', self.name
+        ])
+
+        if rc == 0:
+            try:
+                data = json.loads(stdout)
+                return self._normalize_auth_helper_response(data)
+            except (json.JSONDecodeError, ValueError) as e:
+                self.status_description = f'Auth helper returned invalid JSON: {e}'
+                return None
+
+        # Exit code 1 means re-authentication required, run 'authenticate' automatically
+        if rc == 1:
+            if conf.debug_mode:
+                self.debug(server=self.get_name(),
+                           debug='[Auth helper] get-headers requires re-authentication, launching authenticate...')
+
+            auth_args = [
+                'authenticate',
+                '--server-name', self.name,
+                '--monitor-url', self.monitor_url,
+            ]
+            # Append user-configured extra arguments (e.g. --client-id, --redirect-port)
+            extra = getattr(self, 'auth_helper_extra_args', '')
+            if extra:
+                auth_args.extend(shlex.split(extra))
+            if getattr(self, 'ignore_cert', False):
+                auth_args.append('--insecure')
+
+            auth_stdout, auth_stderr, auth_rc = self._run_auth_helper(auth_args)
+
+            if auth_rc == 0:
+                # Retry get-headers after successful authentication
+                stdout, stderr, rc = self._run_auth_helper([
+                    'get-headers', '--server-name', self.name
+                ])
+                if rc == 0:
+                    try:
+                        data = json.loads(stdout)
+                        return self._normalize_auth_helper_response(data)
+                    except (json.JSONDecodeError, ValueError) as e:
+                        self.status_description = f'Auth helper returned invalid JSON: {e}'
+                        return None
+
+            # Authentication failed or retry failed
+            error_detail = ''
+            try:
+                error_data = json.loads(auth_stdout or stdout)
+                error_detail = error_data.get('error', '')
+            except (json.JSONDecodeError, ValueError):
+                error_detail = auth_stderr or stderr
+            self.status_description = f'Auth helper authentication failed: {error_detail}'
+            return None
+
+        # Any other exit code
+        error_detail = ''
+        try:
+            error_data = json.loads(stdout)
+            error_detail = error_data.get('error', stdout)
+        except (json.JSONDecodeError, ValueError):
+            error_detail = stderr or stdout
+        self.status_description = f'Auth helper error (exit {rc}): {error_detail}'
+        return None
+
+    @staticmethod
+    def _normalize_auth_helper_response(data):
+        """
+        Normalize the auth helper JSON response into {'headers': {...}, 'cookies': {...}}.
+        Accepts either the structured format or a flat dict (treated as headers-only).
+        """
+        if 'headers' in data:
+            return {'headers': data['headers'], 'cookies': data.get('cookies', {})}
+        # Flat dict, treat entire response as headers
+        return {'headers': data, 'cookies': {}}
+
+    def _apply_auth_helper_credentials(self):
+        """
+        Apply auth helper credentials (headers and cookies) to the session.
+        Returns None on success, or a Result on failure (caller should return it).
+        """
+        credentials = self._get_auth_helper_credentials()
+        if credentials is None:
+            return Result(result='ERROR',
+                          error=self.status_description,
+                          status_code=-1)
+        if self.session:
+            if credentials.get('headers'):
+                self.session.headers.update(credentials['headers'])
+            if credentials.get('cookies'):
+                self.session.cookies.update(credentials['cookies'])
+        return None
+
     def get_name(self):
         """
         return stringified name
@@ -950,6 +1074,13 @@ def get_status(self, output=None):
         # initialize HTTP first
         self.init_http()
 
+        # apply auth helper credentials if configured
+        if getattr(self, 'use_auth_helper', False) and getattr(self, 'auth_helper_command', ''):
+            auth_result = self._apply_auth_helper_credentials()
+            if auth_result is not None:
+                self.isChecking = False
+                return auth_result
+
         # get all trouble hosts/services from server specific _get_status()
         status = self._get_status()
 
@@ -986,6 +1117,12 @@ def get_status(self, output=None):
                'login failed' in self.status_description.lower() or \
                self.status_code in self.STATUS_CODES_NO_AUTH:
                 if conf.servers[self.name].enabled is True:
+                    # Auth helper handles its own re-authentication, don't prompt for credentials
+                    if getattr(self, 'use_auth_helper', False) and getattr(self, 'auth_helper_command', ''):
+                        self.isChecking = False
+                        return Result(result='ERROR',
+                                      error=self.status_description,
+                                      status_code=self.status_code)
                     # needed to get valid credentials
                     self.refresh_authentication = True
                     # clean existent authentication