diff --git a/cloudbaseinit/conf/factory.py b/cloudbaseinit/conf/factory.py index 7cdd1fdc..b29a1546 100644 --- a/cloudbaseinit/conf/factory.py +++ b/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', diff --git a/cloudbaseinit/conf/oraclecloud.py b/cloudbaseinit/conf/oraclecloud.py new file mode 100644 index 00000000..80afbb73 --- /dev/null +++ b/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 diff --git a/cloudbaseinit/metadata/services/oraclecloudservice.py b/cloudbaseinit/metadata/services/oraclecloudservice.py new file mode 100644 index 00000000..f126a0ad --- /dev/null +++ b/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) diff --git a/cloudbaseinit/tests/metadata/services/test_oraclecloudservice.py b/cloudbaseinit/tests/metadata/services/test_oraclecloudservice.py new file mode 100644 index 00000000..6bda66a5 --- /dev/null +++ b/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)