feat(Model): support credential-get hook tool in both model and harness

ops/__init__.py

@@ -129,6 +129,8 @@
     'BindingMapping',
     'BlockedStatus',
     'CheckInfoMapping',
+    'CloudCredential',
+    'CloudSpec',
     'ConfigData',
     'Container',
     'ContainerMapping',
@@ -270,6 +272,8 @@
     BindingMapping,
     BlockedStatus,
     CheckInfoMapping,
+    CloudCredential,
+    CloudSpec,
     ConfigData,
     Container,
     ContainerMapping,

ops/model.py

@@ -281,6 +281,20 @@ def get_secret(self, *, id: Optional[str] = None, label: Optional[str] = None) -
         content = self._backend.secret_get(id=id, label=label)
         return Secret(self._backend, id=id, label=label, content=content)
 
+    def get_cloud_spec(self) -> 'CloudSpec':
+        """Get details of the cloud in which the model is deployed.
+
+        Returns a specification for the cloud in which the model is deployed,
+        including access credential information.
+
+        Note: This information is only available for machine charms,
+        not Kubernetes sidecar charms.
+
+        Raises:
+            :class:`ModelError`: if called in a "CaaS" model.
+        """
+        return self._backend.credential_get()
+
 
 if typing.TYPE_CHECKING:
     # (entity type, name): instance.
@@ -3507,6 +3521,14 @@ def reboot(self, now: bool = False):
         else:
             self._run("juju-reboot")
 
+    def credential_get(self) -> 'CloudSpec':
+        """Access cloud credentials by running the credential-get hook tool.
+
+        Returns the cloud specification used by the model.
+        """
+        result = self._run('credential-get', return_output=True, use_json=True)
+        return CloudSpec.from_dict(typing.cast(Dict[str, Any], result))
+
 
 class _ModelBackendValidator:
     """Provides facilities for validating inputs and formatting them for model backends."""
@@ -3596,3 +3618,56 @@ def _ensure_loaded(self):
         self._notice = self._container.get_notice(self.id)
         assert self._notice.type == self.type
         assert self._notice.key == self.key
+
+
+@dataclasses.dataclass(frozen=True)
+class CloudCredential:
+    """CloudCredential contains a cloud credential possibly with secrets redacted.
+
+    Used as the type of attribute `credential` in `CloudSpec`, see below.
+    """
+
+    auth_type: str
+    attributes: Dict[str, Any]
+    redacted: List[str]
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> 'CloudCredential':
+        """Create a new CloudCredential object from a dictionary."""
+        return cls(
+            auth_type=d.get('auth-type', ''),
+            attributes=d.get('attrs', {}),
+            redacted=d.get('redacted', []),
+        )
+
+
+@dataclasses.dataclass(frozen=True)
+class CloudSpec:
+    """Cloud specification information (metadata) including credentials."""
+
+    type: str
+    name: str
+    region: Optional[str]
+    endpoint: Optional[str]
+    is_controller_cloud: Optional[str]
+    credential: Optional[CloudCredential]
+    identity_endpoint: Optional[str]
+    storage_endpoint: Optional[str]
+    ca_certificates: Optional[List[str]]
+    skip_tls_verify: Optional[bool]
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> 'CloudSpec':
+        """Create a new CloudSpec object from a dict parsed from JSON."""
+        return cls(
+            type=typing.cast(str, d.get('type')),
+            name=typing.cast(str, d.get('name')),
+            region=d.get('region'),
+            endpoint=d.get('endpoint'),
+            is_controller_cloud=d.get('isControllerCloud'),
+            credential=CloudCredential.from_dict(d.get('credential', {})),
+            identity_endpoint=d.get('identityEndpoint'),
+            storage_endpoint=d.get('storageEndpoint'),
+            ca_certificates=d.get('caACertificates'),
+            skip_tls_verify=d.get('skipTLSVerify'),
+        )

ops/testing.py

@@ -1911,6 +1911,39 @@ def run_action(self, action_name: str,
                 output=action_under_test.output)
         return action_under_test.output
 
+    def set_cloud_spec(self, spec: 'model.CloudSpec'):
+        """Set cloud specification (metadata) including credentials.
+
+        Call this method before trying to call :meth:`ops.Model.get_cloud_spec`.
+
+        Example usage::
+
+            class MyVMCharm(ops.CharmBase):
+                ...
+                def _on_start(self, event: ops.StartEvent):
+                    spec = self.model.get_cloud_spec()
+                    ....
+
+
+            class TestCharm(unittest.TestCase):
+                def setUp(self):
+                    self.harness = ops.testing.Harness(MyVMCharm)
+                    self.addCleanup(self.harness.cleanup)
+
+                def test_start(self):
+                    cloud_spec_dict = {
+                        "name": "localhost",
+                        "type": "lxd",
+                        "endpoint": "https://127.0.0.1:8443"
+                    }
+                    self.harness.set_cloud_spec(ops.model.CloudSpec.from_dict(cloud_spec_dict))
+                    self.harness.begin_with_initial_hooks()
+                    cloud_spec = self.harness.model.get_cloud_spec()
+                    ...
+
+        """
+        self._backend._cloud_spec = spec
+
 
 def _get_app_or_unit_name(app_or_unit: AppUnitOrName) -> str:
     """Return name of given application or unit (return strings directly)."""
@@ -2126,6 +2159,8 @@ def __init__(self, unit_name: str, meta: charm.CharmMeta, config: '_RawConfig'):
         self._networks: Dict[Tuple[Optional[str], Optional[int]], _NetworkDict] = {}
         self._reboot_count = 0
         self._running_action: Optional[_RunningAction] = None
+        # For `Model.get_cloud_spec`, initialised to None.
+        self._cloud_spec: Optional[model.CloudSpec] = None
 
     def _validate_relation_access(self, relation_name: str, relations: List[model.Relation]):
         """Ensures that the named relation exists/has been added.
@@ -2709,6 +2744,12 @@ def reboot(self, now: bool = False):
         # to handle everything after the exit.
         raise SystemExit()
 
+    def credential_get(self) -> model.CloudSpec:
+        if not self._cloud_spec:
+            raise model.ModelError(
+                'ERROR cloud spec is empty, set it with `Harness.set_cloud_spec()` first')
+        return self._cloud_spec
+
 
 @_copy_docstrings(pebble.ExecProcess)
 class _TestingExecProcess:

test/test_model.py

@@ -3791,5 +3791,82 @@ def test_repr(self):
         )
 
 
+class TestCloudCredential(unittest.TestCase):
+    def setUp(self) -> None:
+        self.cloud_credential_dict = {
+            'auth-type': 'certificate',
+            'attrs': {
+                'client-cert': 'foo',
+                'client-key': 'bar',
+                'server-cert': 'baz'
+            },
+        }
+
+    def test_from_dict(self):
+        cloud_cred = ops.CloudCredential.from_dict(self.cloud_credential_dict)
+        self.assertEqual(cloud_cred.auth_type, 'certificate')
+        self.assertEqual(cloud_cred.attributes, self.cloud_credential_dict.get('attrs'))
+
+
+class TestCloudSpec(unittest.TestCase):
+    def setUp(self) -> None:
+        self.cloud_credential_dict = {
+            'auth-type': 'certificate',
+            'attrs': {
+                'client-cert': 'foo',
+                'client-key': 'bar',
+                'server-cert': 'baz'
+            },
+        }
+
+    def test_from_dict(self):
+        cloud_spec = ops.CloudSpec.from_dict(
+            {
+                'type': 'lxd',
+                'name': 'localhost',
+                'region': 'localhost',
+                'endpoint': 'https://10.76.251.1:8443',
+                'isControllerCloud': None,
+                'credential': self.cloud_credential_dict,
+                'identityEndpoint': None,
+                'storageEndpoint': None,
+                'caACertificates': None,
+                'skipTLSVerify': None
+            }
+        )
+        self.assertEqual(cloud_spec.type, 'lxd')
+        self.assertEqual(cloud_spec.name, 'localhost')
+        self.assertEqual(cloud_spec.region, 'localhost')
+        self.assertEqual(cloud_spec.endpoint, 'https://10.76.251.1:8443')
+        self.assertEqual(cloud_spec.is_controller_cloud, None)
+        self.assertEqual(cloud_spec.identity_endpoint, None)
+        self.assertEqual(cloud_spec.storage_endpoint, None)
+        self.assertEqual(cloud_spec.ca_certificates, None)
+        self.assertEqual(cloud_spec.skip_tls_verify, None)
+        self.assertEqual(
+            cloud_spec.credential,
+            ops.CloudCredential.from_dict(
+                self.cloud_credential_dict))
+
+
+class TestGetCloudSpec(unittest.TestCase):
+    def setUp(self):
+        self.model = ops.Model(ops.CharmMeta(), _ModelBackend('myapp/0'))
+
+    def test_get_cloud_spec(self):
+        fake_script(self, 'credential-get', """echo '{"type": "lxd", "name": "localhost"}'""")
+        cloud_spec = self.model.get_cloud_spec()
+        self.assertEqual(cloud_spec.type, 'lxd')
+        self.assertEqual(cloud_spec.name, 'localhost')
+        self.assertEqual(fake_script_calls(self, clear=True),
+                         [['credential-get', '--format=json']])
+
+    @patch("ops.model._ModelBackend.credential_get")
+    def test_get_cloud_spec_error(self, credential_get: MagicMock):
+        credential_get.side_effect = ops.ModelError
+        with self.assertRaises(ops.ModelError):
+            self.model.get_cloud_spec()
+
+
 if __name__ == "__main__":
     unittest.main()

test/test_testing.py

@@ -5854,3 +5854,34 @@ def test_get_notices(self):
 class TestNotices(unittest.TestCase, _TestingPebbleClientMixin, PebbleNoticesMixin):
     def setUp(self):
         self.client = self.get_testing_client()
+
+
+class TestCloudSpec(unittest.TestCase):
+    def test_get_set_cloud_spec(self):
+        harness = ops.testing.Harness(EventRecorder, meta='name: myapp')
+        self.addCleanup(harness.cleanup)
+        cloud_spec_dict = {
+            'name': 'localhost',
+            'type': 'lxd',
+            'endpoint': 'https://127.0.0.1:8443',
+            'credential': {
+                'authtype': 'certificate',
+                'attrs': {
+                    'client-cert': 'foo',
+                    'client-key': 'bar',
+                    'server-cert': 'baz'
+                },
+            },
+        }
+        harness.set_cloud_spec(ops.model.CloudSpec.from_dict(cloud_spec_dict))
+        harness.begin()
+        result = harness.model.get_cloud_spec()
+        expected = ops.model.CloudSpec.from_dict(cloud_spec_dict)
+        self.assertEqual(repr(result), repr(expected))
+
+    def test_get_cloud_spec_without_set_error(self):
+        harness = ops.testing.Harness(EventRecorder, meta='name: myapp')
+        self.addCleanup(harness.cleanup)
+        harness.begin()
+        with self.assertRaises(ops.ModelError):
+            harness.model.get_cloud_spec()