mgr/cephadm: support tls creds via uri in service spec

Support populating the cert data sent to the cephadm binary using
special `URI:` prefixed strings instead of putting the cert data itself
in the smb service spec. This avoids having an extra copy of the cert
floating around but still matches the behavior of other services where
cephadm writes the certs into files.

In the future we may be able to avoid even putting the data in here as
sambacc can use rados apis - but for simplicity and matching other
services we will send the data this way for now.

Signed-off-by: John Mulligan <[email protected]>

Author: phlogistonjohn

src/pybind/mgr/cephadm/services/smb.py

@@ -163,23 +163,37 @@ def generate_config(
             _add_cfg(
                 files,
                 'remote_control.ssl.crt',
-                smb_spec.remote_control_ssl_cert,
+                self._cert_or_uri(smb_spec.remote_control_ssl_cert),
             )
             _add_cfg(
                 files,
                 'remote_control.ssl.key',
-                smb_spec.remote_control_ssl_key,
+                self._cert_or_uri(smb_spec.remote_control_ssl_key),
             )
             _add_cfg(
                 files,
                 'remote_control.ca.crt',
-                smb_spec.remote_control_ca_cert,
+                self._cert_or_uri(smb_spec.remote_control_ca_cert),
             )
 
         logger.debug('smb generate_config: %r', config_blobs)
         self._configure_cluster_meta(smb_spec, daemon_spec)
         return config_blobs, []
 
+    def _cert_or_uri(self, data: Optional[str]) -> Optional[str]:
+        if data is None:
+            return None
+        if not data.startswith("URI:"):
+            return data
+        uri = data[4:]
+        if not uri.startswith('rados:mon-config-key'):
+            raise ValueError('unhandled URI scheme')
+
+        from smb.mon_store import MonKeyConfigStore
+        store = MonKeyConfigStore(self.mgr)
+        entry = store.lookup_uri(uri)
+        return entry.get_data()
+
     def config_dashboard(
         self, daemon_descrs: List[DaemonDescription]
     ) -> None: