Adding option --force-server to specify a server to ipa-certupdate tool.

This can be used for disaster recovery or situation with replication
issues, when only one server is fixed/working and is a source of
truth about CA certs for all the other replicas and clients.

Fixes: https://pagure.io/freeipa/issue/9839
Signed-off-by: Aleksandr Sharov (asharov@redhat.com)
Reviewed-By: Rob Crittenden <rcritten@redhat.com>
Reviewed-By: Florence Blanc-Renaud <flo@redhat.com>
Reviewed-By: Alexander Bokovoy <abokovoy@redhat.com>

client/man/ipa-certupdate.1

@@ -16,7 +16,7 @@
 .\"
 .\" Author: Jan Cholasta <jcholast@redhat.com>
 .\"
-.TH "ipa-certupdate" "1" "Jul 2 2014" "IPA" "IPA Manual Pages"
+.TH "ipa-certupdate" "1" "Dec 27 2025" "IPA" "IPA Manual Pages"
 .SH "NAME"
 ipa\-certupdate \- Update local IPA certificate databases with certificates from the server
 .SH "SYNOPSIS"
@@ -25,6 +25,9 @@ ipa\-certupdate \- Update local IPA certificate databases with certificates from
 \fBipa\-certupdate\fR can be used to update local IPA certificate databases with certificates from the server.
 .SH "OPTIONS"
 .TP
+\fB\-\-force\-server\fR=\fIipa.server.fqdn\fR
+Force the use of the specified server for the update. Use only fully-qualified domain name of the server, and verify that target server has a correct set of CA certificates, local certificates will be overwritten.
+.TP
 \fB\-v\fR, \fB\-\-verbose\fR
 Print debugging information.
 .TP

ipaclient/install/ipa_certupdate.py

@@ -45,8 +45,20 @@ class CertUpdate(admintool.AdminTool):
     description = ("Update local IPA certificate databases with certificates "
                    "from the server.")
 
-    def validate_options(self):
-        super(CertUpdate, self).validate_options(needs_root=True)
+    @classmethod
+    def add_options(cls, parser, debug_option=False):
+        super(CertUpdate, cls).add_options(parser)
+        parser.add_option(
+            "--force-server",
+            dest="forced_server",
+            type=str,
+            default=None,
+            metavar="ipa.server.fqdn",
+            help="Force the use of the specified server for the update. ",
+        )
+
+    def validate_options(self, needs_root=True):
+        super().validate_options(needs_root)
 
     def run(self):
         check_client_configuration()
@@ -56,11 +68,20 @@ class CertUpdate(admintool.AdminTool):
         os.environ['KRB5CCNAME'] = "MEMORY:"
 
         try:
-            api.bootstrap(context='cli_installer', confdir=paths.ETC_IPA)
+            # set bootstrap parameters
+            bootstrap_kw = {
+                'context': 'cli_installer',
+                'confdir': paths.ETC_IPA,
+            }
+            # pass forced server to bootstrap if specified
+            if self.options.forced_server is not None:
+                bootstrap_kw['forced_server'] = self.options.forced_server
+
+            api.bootstrap(**bootstrap_kw)
             api.finalize()
 
             api.Backend.rpcclient.connect()
-            run_with_args(api)
+            run_with_args(api, self.options)
             api.Backend.rpcclient.disconnect()
         except errors.CCacheError:
             logger.error(
@@ -75,7 +96,7 @@ class CertUpdate(admintool.AdminTool):
                 os.environ['KRB5CCNAME'] = old_krb5ccname
 
 
-def run_with_args(api):
+def run_with_args(api, options=None):
     """
     Run the certupdate procedure with the given API object.
 
@@ -83,7 +104,15 @@ def run_with_args(api):
                 (such that Commands can be invoked)
 
     """
-    server = urlsplit(api.env.jsonrpc_uri).hostname
+    if options is None or getattr(options, "forced_server", None) is None:
+        # Use the server from the API environment
+        server = urlsplit(api.env.jsonrpc_uri).hostname
+    else:
+        # Use the forced server
+        server = options.forced_server
+
+    logger.info("Updating certificates from server %s", server)
+
     ldap = ipaldap.LDAPClient.from_hostname_secure(server)
 
     try:

ipatests/test_integration/test_commands.py

@@ -1500,6 +1500,51 @@ class TestIPACommand(IntegrationTest):
         # Run it again for good measure
         self.master.run_command(["ipa-certupdate"])
 
+    def test_certupdate_force_server(self):
+        """Test that certupdate works with a forced server.
+
+           This is useful for running certupdate against a specific master,
+           especially if replicas have different CA certs in a disaster
+           recovery scenario.
+        """
+        tasks.kdestroy_all(self.master)
+
+        # try with a correct server
+        server = self.replicas[0].hostname
+        result = self.master.run_command(
+            ["ipa-certupdate", "--force-server", server]
+        )
+        stdstring = "Updating certificates from server %s" % server
+        assert stdstring in result.stderr_text
+        assert "The ipa-certupdate command was successful" \
+            in result.stderr_text
+
+        # try with a non-existent server
+        non_server = 'non-' + server
+        result = self.master.run_command(
+            ["ipa-certupdate", "--force-server", non_server],
+            raiseonerr=False
+        )
+        assert result.returncode != 0
+        assert "cannot connect to 'ldap://%s:389" % non_server \
+            in result.stderr_text
+
+        # try with a turned off/inaccessible server
+        self.replicas[0].run_command(["ipactl", "stop"])
+        result = self.master.run_command(
+            ["ipa-certupdate", "--force-server", server],
+            raiseonerr=False
+        )
+        assert result.returncode != 0
+        # Error may be either LDAP URI - Transport endpoint is not connected
+        # or an HTTP URI - [Errno 111] Connection refused
+        pattern = rf"cannot connect to .*{re.escape(server)}"
+        error = r'\b(?:Transport endpoint is not connected|\
+            Connection refused)\b'
+        assert re.search(pattern,result.stderr_text)
+        assert re.search(error, result.stderr_text)
+        self.replicas[0].run_command(["ipactl", "start"])
+
     def test_proxycommand_invalid_shell(self):
         """Test that ssh works with a user with an invalid shell.