metadata: add Oracle Cloud IMDSv2 service

cloudbaseinit/conf/factory.py

@@ -21,6 +21,7 @@
     'cloudbaseinit.conf.ec2.EC2Options',
     'cloudbaseinit.conf.maas.MAASOptions',
     'cloudbaseinit.conf.openstack.OpenStackOptions',
+    'cloudbaseinit.conf.oraclecloud.OracleCloudOptions',
     'cloudbaseinit.conf.azure.AzureOptions',
     'cloudbaseinit.conf.ovf.OvfOptions',
     'cloudbaseinit.conf.packet.PacketOptions',

cloudbaseinit/conf/oraclecloud.py

@@ -0,0 +1,55 @@
+# Copyright (c) 2026, Oracle and/or its affiliates.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""Config options available for the Oracle Cloud metadata service."""
+
+from oslo_config import cfg
+
+from cloudbaseinit.conf import base as conf_base
+
+
+class OracleCloudOptions(conf_base.Options):
+
+    """Config options available for the OpenStack metadata service."""
+
+    def __init__(self, config):
+        super(OracleCloudOptions, self).__init__(config, group="oraclecloud")
+        self._options = [
+            cfg.StrOpt(
+                "metadata_base_url", default="http://169.254.169.254/",
+                help="The base URL where the service looks for metadata",
+                deprecated_group="DEFAULT"),
+            cfg.BoolOpt(
+                "add_metadata_private_ip_route", default=True,
+                help="Add a route for the metadata ip address to the gateway",
+                deprecated_group="DEFAULT"),
+            cfg.BoolOpt(
+                "https_allow_insecure", default=False,
+                help="Whether to disable the validation of HTTPS "
+                     "certificates."),
+            cfg.StrOpt(
+                "https_ca_bundle", default=None,
+                help="The path to a CA_BUNDLE file or directory with "
+                     "certificates of trusted CAs."),
+        ]
+
+    def register(self):
+        """Register the current options to the global ConfigOpts object."""
+        group = cfg.OptGroup(self.group_name, title='OracleCloud Options')
+        self._config.register_group(group)
+        self._config.register_opts(self._options, group=group)
+
+    def list(self):
+        """Return a list which contains all the available options."""
+        return self._options

cloudbaseinit/metadata/services/oraclecloudservice.py

@@ -0,0 +1,69 @@
+# Copyright (c) 2026, Oracle and/or its affiliates.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from base64 import b64decode
+
+from oslo_log import log as oslo_logging
+
+from cloudbaseinit import conf as cloudbaseinit_conf
+from cloudbaseinit.metadata.services import base
+from cloudbaseinit.utils import network
+
+CONF = cloudbaseinit_conf.CONF
+LOG = oslo_logging.getLogger(__name__)
+
+
+class OracleCloudService(base.BaseHTTPMetadataService):
+    _metadata_version = 'v2'
+    _headers = {"Authorization": "Bearer Oracle"}
+
+    def __init__(self):
+        super(OracleCloudService, self).__init__(
+            base_url=CONF.oraclecloud.metadata_base_url,
+            https_allow_insecure=CONF.oraclecloud.https_allow_insecure,
+            https_ca_bundle=CONF.oraclecloud.https_ca_bundle)
+        self._enable_retry = True
+
+    def _http_request(self, url, data=None, headers=None, method=None):
+        headers = dict(headers or {})
+        headers.update(self._headers)
+
+        return super(OracleCloudService, self)._http_request(
+            url, data, headers, method)
+
+    def load(self):
+        super(OracleCloudService, self).load()
+        if CONF.oraclecloud.add_metadata_private_ip_route:
+            network.check_metadata_ip_route(CONF.oraclecloud.metadata_base_url)
+
+        try:
+            self.get_instance_id()
+            return True
+        except Exception as ex:
+            LOG.exception(ex)
+            LOG.debug('Metadata not found at URL \'%s\'' %
+                      CONF.oraclecloud.metadata_base_url)
+            return False
+
+    def get_instance_id(self):
+        return self._get_cache_data('opc/%s/instance/id' %
+                                    self._metadata_version, decode=True)
+
+    def get_user_data(self):
+        return b64decode(self._get_cache_data(
+            'opc/%s/instance/metadata/user_data' % self._metadata_version))
+
+    def get_host_name(self):
+        return self._get_cache_data('opc/%s/instance/hostname' %
+                                    self._metadata_version, decode=True)

cloudbaseinit/tests/metadata/services/test_oraclecloudservice.py

@@ -0,0 +1,104 @@
+# Copyright (c) 2026, Oracle and/or its affiliates.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import unittest
+
+try:
+    import unittest.mock as mock
+except ImportError:
+    import mock
+
+from cloudbaseinit import conf as cloudbaseinit_conf
+from cloudbaseinit.metadata.services import oraclecloudservice
+from cloudbaseinit.tests import testutils
+
+CONF = cloudbaseinit_conf.CONF
+
+
+class OracleCloudServiceTest(unittest.TestCase):
+
+    def setUp(self):
+        self._service = oraclecloudservice.OracleCloudService()
+
+    @mock.patch('cloudbaseinit.utils.network.check_metadata_ip_route')
+    @mock.patch('cloudbaseinit.metadata.services.oraclecloudservice.'
+                'OracleCloudService.get_instance_id')
+    def _test_load(self, mock_get_instance_id, mock_check_metadata_ip_route,
+                   side_effect):
+        mock_get_instance_id.side_effect = [side_effect]
+        with testutils.LogSnatcher('cloudbaseinit.metadata.services.'
+                                   'oraclecloudservice'):
+            response = self._service.load()
+
+        mock_check_metadata_ip_route.assert_called_once_with(
+            CONF.oraclecloud.metadata_base_url)
+        mock_get_instance_id.assert_called_once_with()
+        if side_effect is Exception:
+            self.assertFalse(response)
+        else:
+            self.assertTrue(response)
+
+    def test_load(self):
+        self._test_load(side_effect=None)
+
+    def test_load_exception(self):
+        self._test_load(side_effect=Exception)
+
+    @mock.patch('cloudbaseinit.metadata.services.base.'
+                'BaseHTTPMetadataService._http_request')
+    def test_http_request(self, mock_http_request):
+        headers = {'Test-Header': 'test-value'}
+        expected_headers = dict(headers)
+        expected_headers.update(self._service._headers)
+
+        response = self._service._http_request(
+            mock.sentinel.url, data=mock.sentinel.data, headers=headers,
+            method=mock.sentinel.method)
+
+        mock_http_request.assert_called_once_with(
+            mock.sentinel.url, mock.sentinel.data, expected_headers,
+            mock.sentinel.method)
+        self.assertEqual(mock_http_request.return_value, response)
+        self.assertEqual({'Test-Header': 'test-value'}, headers)
+
+    @mock.patch('cloudbaseinit.metadata.services.oraclecloudservice.'
+                'OracleCloudService._get_cache_data')
+    def test_get_instance_id(self, mock_get_cache_data):
+        mock_get_cache_data.return_value = 'test-instance-id'
+        response = self._service.get_instance_id()
+        mock_get_cache_data.assert_called_once_with(
+            'opc/%s/instance/id' % self._service._metadata_version,
+            decode=True)
+        self.assertEqual(mock_get_cache_data.return_value, response)
+
+    @mock.patch('cloudbaseinit.metadata.services.oraclecloudservice.'
+                'OracleCloudService._get_cache_data')
+    def test_get_user_data(self, mock_get_cache_data):
+        mock_get_cache_data.return_value = (
+            'VGVzdGluZyBvdXIgY29kZSBpcyBnb29kCg==')
+        response = self._service.get_user_data()
+        mock_get_cache_data.assert_called_once_with(
+            'opc/%s/instance/metadata/user_data' %
+            self._service._metadata_version)
+        self.assertEqual(response.decode(), "Testing our code is good\n")
+
+    @mock.patch('cloudbaseinit.metadata.services.oraclecloudservice.'
+                'OracleCloudService._get_cache_data')
+    def test_get_host_name(self, mock_get_cache_data):
+        mock_get_cache_data.return_value = 'test-hostname'
+        response = self._service.get_host_name()
+        mock_get_cache_data.assert_called_once_with(
+            'opc/%s/instance/hostname' % self._service._metadata_version,
+            decode=True)
+        self.assertEqual(mock_get_cache_data.return_value, response)