mgr/cephadm: adding new comands to remvoe keys/certificates

Signed-off-by: Redouane Kachach <[email protected]>

Author: rkachach

doc/cephadm/certmgr.rst

@@ -229,6 +229,32 @@ To update or set a new private key:
 
 This command allows administrators to provide new private keys for services.
 
+Removing a Certificate
+======================
+
+To remove an existing certificate:
+
+.. prompt:: bash #
+
+   ceph orch certmgr cert rm <certificate_name> [--service_name <value>] [--hostname <value>]
+
+**Note:** For certificates with host or service scope, use the `--service-name` or `--hostname` option to specify the target.
+
+``<certificate_name>`` must be a valid certificate name. Use ``ceph orch certmgr cert ls`` to list supported certificates.
+
+Removing a Private Key
+======================
+
+To remove an existing private key:
+
+.. prompt:: bash #
+
+   ceph orch certmgr key rm <key-name> [--service_name <value>] [--hostname <value>]
+
+**Note:** For keys with host or service scope, use the `--service-name` or `--hostname` option to specify the target.
+
+``<key_name>`` must be a valid key name. Use ``ceph orch certmgr key ls`` to list supported keys.
+
 Generating Certificates
 =======================
 

src/pybind/mgr/cephadm/module.py

@@ -17,7 +17,7 @@
 
 from ceph.deployment.service_spec import PrometheusSpec
 from cephadm.cert_mgr import CertMgr
-from cephadm.tlsobject_store import TLSObjectScope
+from cephadm.tlsobject_store import TLSObjectScope, TLSObjectException
 
 import string
 from typing import List, Dict, Optional, Callable, Tuple, TypeVar, \
@@ -3317,6 +3317,36 @@ def cert_store_set_key(
         self.cert_mgr.save_key(key_name, key, service_name, hostname, True)
         return f'Key for {key_name} set correctly'
 
+    @handle_orch_error
+    def cert_store_rm_cert(
+        self,
+        cert_name: str,
+        service_name: Optional[str] = None,
+        hostname: Optional[str] = None,
+    ) -> str:
+
+        try:
+            self.cert_mgr.rm_cert(cert_name, service_name, hostname)
+            return f'Certificate for {cert_name} removed correctly'
+        except TLSObjectException:
+            raise OrchestratorError("Cannot delete the certificate. Please use 'ceph orch certmgr cert ls' to list available certificates. \n"
+                                    "Note: for certificates with host/service scope use --service-name or --hostname to specify the target.")
+
+    @handle_orch_error
+    def cert_store_rm_key(
+        self,
+        key_name: str,
+        service_name: Optional[str] = None,
+        hostname: Optional[str] = None,
+    ) -> str:
+
+        try:
+            self.cert_mgr.rm_key(key_name, service_name, hostname)
+            return f'Key for {key_name} removed correctly'
+        except TLSObjectException:
+            raise OrchestratorError("Cannot delete the key. Please use 'ceph orch certmgr key ls' to list available keys. \n"
+                                    "Note: for keys with host/service scope use --service-name or --hostname to specify the target.")
+
     @handle_orch_error
     def apply_mon(self, spec: ServiceSpec) -> str:
         return self._apply(spec)

src/pybind/mgr/orchestrator/_interface.py

@@ -623,6 +623,22 @@ def cert_store_set_key(
     ) -> OrchResult[str]:
         raise NotImplementedError()
 
+    def cert_store_rm_cert(
+        self,
+        cert_name: str,
+        service_name: Optional[str] = None,
+        hostname: Optional[str] = None,
+    ) -> OrchResult[str]:
+        raise NotImplementedError()
+
+    def cert_store_rm_key(
+        self,
+        key_name: str,
+        service_name: Optional[str] = None,
+        hostname: Optional[str] = None,
+    ) -> OrchResult[str]:
+        raise NotImplementedError()
+
     @handle_orch_error
     def apply(
         self,

src/pybind/mgr/orchestrator/module.py

@@ -1318,6 +1318,42 @@ def _cert_store_set_key(
         output = raise_if_exception(completion)
         return HandleCommandResult(stdout=output)
 
+    @_cli_write_command('orch certmgr cert rm')
+    def _cert_store_rm_cert(
+        self,
+        cert_name: str,
+        _end_positional_: int = 0,
+        service_name: Optional[str] = None,
+        hostname: Optional[str] = None,
+        inbuf: Optional[str] = None
+    ) -> HandleCommandResult:
+
+        completion = self.cert_store_rm_cert(
+            cert_name,
+            service_name,
+            hostname,
+        )
+        output = raise_if_exception(completion)
+        return HandleCommandResult(stdout=output)
+
+    @_cli_write_command('orch certmgr key rm')
+    def _cert_store_rm_key(
+        self,
+        key_name: str,
+        _end_positional_: int = 0,
+        service_name: Optional[str] = None,
+        hostname: Optional[str] = None,
+        inbuf: Optional[str] = None
+    ) -> HandleCommandResult:
+
+        completion = self.cert_store_rm_key(
+            key_name,
+            service_name,
+            hostname,
+        )
+        output = raise_if_exception(completion)
+        return HandleCommandResult(stdout=output)
+
     def _get_credentials(self, username: Optional[str] = None, password: Optional[str] = None, inbuf: Optional[str] = None) -> Tuple[str, str]:
 
         _username = username