diff options
| -rw-r--r-- | cloudinit/sources/DataSourceHetzner.py | 36 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_hetzner.py | 19 |
2 files changed, 41 insertions, 14 deletions
diff --git a/cloudinit/sources/DataSourceHetzner.py b/cloudinit/sources/DataSourceHetzner.py index 79353882..1d965bf7 100644 --- a/cloudinit/sources/DataSourceHetzner.py +++ b/cloudinit/sources/DataSourceHetzner.py @@ -3,15 +3,18 @@ # # This file is part of cloud-init. See LICENSE file for license information. # -"""Hetzner Cloud API Documentation. +"""Hetzner Cloud API Documentation https://docs.hetzner.cloud/""" +from cloudinit import log as logging from cloudinit import net as cloudnet from cloudinit import sources from cloudinit import util import cloudinit.sources.helpers.hetzner as hc_helper +LOG = logging.getLogger(__name__) + BASE_URL_V1 = 'http://169.254.169.254/hetzner/v1' BUILTIN_DS_CONFIG = { @@ -43,9 +46,12 @@ class DataSourceHetzner(sources.DataSource): self._network_config = None self.dsmode = sources.DSMODE_NETWORK - def get_data(self): - if not on_hetzner(): + def _get_data(self): + (on_hetzner, serial) = get_hcloud_data() + + if not on_hetzner: return False + nic = cloudnet.find_fallback_nic() with cloudnet.EphemeralIPv4Network(nic, "169.254.0.1", 16, "169.254.255.255"): @@ -75,8 +81,18 @@ class DataSourceHetzner(sources.DataSource): self.metadata['public-keys'] = md.get('public-keys', None) self.vendordata_raw = md.get("vendor_data", None) + # instance-id and serial from SMBIOS should be identical + if self.metadata['instance-id'] != serial: + raise RuntimeError( + "SMBIOS serial does not match instance ID from metadata" + ) + return True + def check_instance_id(self, sys_cfg): + return sources.instance_id_matches_system_uuid( + self.get_instance_id(), 'system-serial-number') + @property def network_config(self): """Configure the networking. This needs to be done each boot, since @@ -96,8 +112,18 @@ class DataSourceHetzner(sources.DataSource): return self._network_config -def on_hetzner(): - return util.read_dmi_data('system-manufacturer') == "Hetzner" +def get_hcloud_data(): + vendor_name = util.read_dmi_data('system-manufacturer') + if vendor_name != "Hetzner": + return (False, None) + + serial = util.read_dmi_data("system-serial-number") + if serial: + LOG.debug("Running on Hetzner Cloud: serial=%s", serial) + else: + raise RuntimeError("Hetzner Cloud detected, but no serial found") + + return (True, serial) # Used to match classes to dependencies diff --git a/tests/unittests/test_datasource/test_hetzner.py b/tests/unittests/test_datasource/test_hetzner.py index d0879545..fc3649a4 100644 --- a/tests/unittests/test_datasource/test_hetzner.py +++ b/tests/unittests/test_datasource/test_hetzner.py @@ -77,10 +77,10 @@ class TestDataSourceHetzner(CiTestCase): @mock.patch('cloudinit.net.find_fallback_nic') @mock.patch('cloudinit.sources.helpers.hetzner.read_metadata') @mock.patch('cloudinit.sources.helpers.hetzner.read_userdata') - @mock.patch('cloudinit.sources.DataSourceHetzner.on_hetzner') - def test_read_data(self, m_on_hetzner, m_usermd, m_readmd, m_fallback_nic, - m_net): - m_on_hetzner.return_value = True + @mock.patch('cloudinit.sources.DataSourceHetzner.get_hcloud_data') + def test_read_data(self, m_get_hcloud_data, m_usermd, m_readmd, + m_fallback_nic, m_net): + m_get_hcloud_data.return_value = (True, METADATA.get('instance-id')) m_readmd.return_value = METADATA.copy() m_usermd.return_value = USERDATA m_fallback_nic.return_value = 'eth0' @@ -107,11 +107,12 @@ class TestDataSourceHetzner(CiTestCase): @mock.patch('cloudinit.sources.helpers.hetzner.read_metadata') @mock.patch('cloudinit.net.find_fallback_nic') - @mock.patch('cloudinit.sources.DataSourceHetzner.on_hetzner') - def test_not_on_hetzner_returns_false(self, m_on_hetzner, m_find_fallback, - m_read_md): - """If helper 'on_hetzner' returns False, return False from get_data.""" - m_on_hetzner.return_value = False + @mock.patch('cloudinit.sources.DataSourceHetzner.get_hcloud_data') + def test_not_on_hetzner_returns_false(self, m_get_hcloud_data, + m_find_fallback, m_read_md): + """If helper 'get_hcloud_data' returns False, + return False from get_data.""" + m_get_hcloud_data.return_value = (False, None) ds = self.get_ds() ret = ds.get_data() |