# This file is part of cloud-init. See LICENSE file for license information.
"""Datasource for Oracle (OCI/Oracle Cloud Infrastructure)
Notes:
* This datasource does not support OCI Classic. OCI Classic provides an EC2
lookalike metadata service.
* The UUID provided in DMI data is not the same as the meta-data provided
instance-id, but has an equivalent lifespan.
* We do need to support upgrade from an instance that cloud-init
identified as OpenStack.
* Bare metal instances use iSCSI root, virtual machine instances do not.
* Both bare metal and virtual machine instances provide a chassis-asset-tag of
OracleCloud.com.
"""
import base64
import ipaddress
import logging
from collections import namedtuple
from typing import Optional, Tuple
from cloudinit import atomic_helper, dmi, net, sources, util
from cloudinit.distros.networking import NetworkConfig
from cloudinit.net import (
cmdline,
ephemeral,
get_interfaces_by_mac,
is_netfail_master,
)
from cloudinit.url_helper import UrlError, readurl
LOG = logging.getLogger(__name__)
BUILTIN_DS_CONFIG = {
# Don't use IMDS to configure secondary NICs by default
"configure_secondary_nics": False,
}
CHASSIS_ASSET_TAG = "OracleCloud.com"
METADATA_ROOT = "http://169.254.169.254/opc/v{version}/"
METADATA_PATTERN = METADATA_ROOT + "{path}/"
# https://docs.cloud.oracle.com/iaas/Content/Network/Troubleshoot/connectionhang.htm#Overview,
# indicates that an MTU of 9000 is used within OCI
MTU = 9000
V2_HEADERS = {"Authorization": "Bearer Oracle"}
OpcMetadata = namedtuple("OpcMetadata", "version instance_data vnics_data")
class KlibcOracleNetworkConfigSource(cmdline.KlibcNetworkConfigSource):
"""Override super class to lower the applicability conditions.
If any `/run/net-*.cfg` files exist, then it is applicable. Even if
`/run/initramfs/open-iscsi.interface` does not exist.
"""
def is_applicable(self) -> bool:
"""Override is_applicable"""
return bool(self._files)
def _ensure_netfailover_safe(network_config: NetworkConfig) -> None:
"""
Search network config physical interfaces to see if any of them are
a netfailover master. If found, we prevent matching by MAC as the other
failover devices have the same MAC but need to be ignored.
Note: we rely on cloudinit.net changes which prevent netfailover devices
from being present in the provided network config. For more details about
netfailover devices, refer to cloudinit.net module.
:param network_config
A v1 or v2 network config dict with the primary NIC, and possibly
secondary nic configured. This dict will be mutated.
"""
# ignore anything that's not an actual network-config
if "version" not in network_config:
return
if network_config["version"] not in [1, 2]:
LOG.debug(
"Ignoring unknown network config version: %s",
network_config["version"],
)
return
mac_to_name = get_interfaces_by_mac()
if network_config["version"] == 1:
for cfg in [c for c in network_config["config"] if "type" in c]:
if cfg["type"] == "physical":
if "mac_address" in cfg:
mac = cfg["mac_address"]
cur_name = mac_to_name.get(mac)
if not cur_name:
continue
elif is_netfail_master(cur_name):
del cfg["mac_address"]
elif network_config["version"] == 2:
for _, cfg in network_config.get("ethernets", {}).items():
if "match" in cfg:
macaddr = cfg.get("match", {}).get("macaddress")
if macaddr:
cur_name = mac_to_name.get(macaddr)
if not cur_name:
continue
elif is_netfail_master(cur_name):
del cfg["match"]["macaddress"]
del cfg["set-name"]
cfg["match"]["name"] = cur_name
class DataSourceOracle(sources.DataSource):
dsname = "Oracle"
system_uuid = None
vendordata_pure = None
network_config_sources: Tuple[sources.NetworkConfigSource, ...] = (
sources.NetworkConfigSource.CMD_LINE,
sources.NetworkConfigSource.SYSTEM_CFG,
sources.NetworkConfigSource.DS,
sources.NetworkConfigSource.INITRAMFS,
)
_network_config: dict = {"config": [], "version": 1}
def __init__(self, sys_cfg, *args, **kwargs):
super(DataSourceOracle, self).__init__(sys_cfg, *args, **kwargs)
self._vnics_data = None
self.ds_cfg = util.mergemanydict(
[
util.get_cfg_by_path(sys_cfg, ["datasource", self.dsname], {}),
BUILTIN_DS_CONFIG,
]
)
self._network_config_source = KlibcOracleNetworkConfigSource()
def _has_network_config(self) -> bool:
return bool(self._network_config.get("config", []))
@staticmethod
def ds_detect() -> bool:
"""Check platform environment to report if this datasource may run."""
return _is_platform_viable()
def _get_data(self):
self.system_uuid = _read_system_uuid()
network_context = ephemeral.EphemeralDHCPv4(
self.distro,
iface=net.find_fallback_nic(),
connectivity_url_data={
"url": METADATA_PATTERN.format(version=2, path="instance"),
"headers": V2_HEADERS,
},
)
fetch_primary_nic = not self._is_iscsi_root()
fetch_secondary_nics = self.ds_cfg.get(
"configure_secondary_nics",
BUILTIN_DS_CONFIG["configure_secondary_nics"],
)
with network_context:
fetched_metadata = read_opc_metadata(
fetch_vnics_data=fetch_primary_nic or fetch_secondary_nics
)
data = self._crawled_metadata = fetched_metadata.instance_data
self.metadata_address = METADATA_ROOT.format(
version=fetched_metadata.version
)
self._vnics_data = fetched_metadata.vnics_data
self.metadata = {
"availability-zone": data["ociAdName"],
"instance-id": data["id"],
"launch-index": 0,
"local-hostname": data["hostname"],
"name": data["displayName"],
}
if "metadata" in data:
user_data = data["metadata"].get("user_data")
if user_data:
self.userdata_raw = base64.b64decode(user_data)
self.metadata["public_keys"] = data["metadata"].get(
"ssh_authorized_keys"
)
return True
def check_instance_id(self, sys_cfg) -> bool:
"""quickly check (local only) if self.instance_id is still valid
On Oracle, the dmi-provided system uuid differs from the instance-id
but has the same life-span."""
return sources.instance_id_matches_system_uuid(self.system_uuid)
def get_public_ssh_keys(self):
return sources.normalize_pubkey_data(self.metadata.get("public_keys"))
def _is_iscsi_root(self) -> bool:
"""Return whether we are on a iscsi machine."""
return self._network_config_source.is_applicable()
def _get_iscsi_config(self) -> dict:
return self._network_config_source.render_config()
@property
def network_config(self):
"""Network config is read from initramfs provided files
Priority for primary network_config selection:
- iscsi
- imds
If none is present, then we fall back to fallback configuration.
"""
if self._has_network_config():
return self._network_config
set_primary = False
# this is v1
if self._is_iscsi_root():
self._network_config = self._get_iscsi_config()
if not self._has_network_config():
LOG.warning(
"Could not obtain network configuration from initramfs. "
"Falling back to IMDS."
)
set_primary = True
set_secondary = self.ds_cfg.get(
"configure_secondary_nics",
BUILTIN_DS_CONFIG["configure_secondary_nics"],
)
if set_primary or set_secondary:
try:
# Mutate self._network_config to include primary and/or
# secondary VNICs
self._add_network_config_from_opc_imds(set_primary)
except Exception:
util.logexc(
LOG,
"Failed to parse IMDS network configuration!",
)
# we need to verify that the nic selected is not a netfail over
# device and, if it is a netfail master, then we need to avoid
# emitting any match by mac
_ensure_netfailover_safe(self._network_config)
return self._network_config
def _add_network_config_from_opc_imds(self, set_primary: bool = False):
"""Generate primary and/or secondary NIC config from IMDS and merge it.
It will mutate the network config to include the secondary VNICs.
:param set_primary: If True set primary interface.
:raises:
Exceptions are not handled within this function. Likely
exceptions are KeyError/IndexError
(if the IMDS returns valid JSON with unexpected contents).
"""
if self._vnics_data is None:
LOG.warning("NIC data is UNSET but should not be")
return
if not set_primary and ("nicIndex" in self._vnics_data[0]):
# TODO: Once configure_secondary_nics defaults to True, lower the
# level of this log message. (Currently, if we're running this
# code at all, someone has explicitly opted-in to secondary
# VNIC configuration, so we should warn them that it didn't
# happen. Once it's default, this would be emitted on every Bare
# Metal Machine launch, which means INFO or DEBUG would be more
# appropriate.)
LOG.warning(
"VNIC metadata indicates this is a bare metal machine; "
"skipping secondary VNIC configuration."
)
return
interfaces_by_mac = get_interfaces_by_mac()
vnics_data = self._vnics_data if set_primary else self._vnics_data[1:]
for index, vnic_dict in enumerate(vnics_data):
is_primary = set_primary and index == 0
mac_address = vnic_dict["macAddr"].lower()
if mac_address not in interfaces_by_mac:
LOG.warning(
"Interface with MAC %s not found; skipping",
mac_address,
)
continue
name = interfaces_by_mac[mac_address]
network = ipaddress.ip_network(vnic_dict["subnetCidrBlock"])
if self._network_config["version"] == 1:
if is_primary:
subnet = {"type": "dhcp"}
else:
subnet = {
"type": "static",
"address": (
f"{vnic_dict['privateIp']}/{network.prefixlen}"
),
}
interface_config = {
"name": name,
"type": "physical",
"mac_address": mac_address,
"mtu": MTU,
"subnets": [subnet],
}
self._network_config["config"].append(interface_config)
elif self._network_config["version"] == 2:
# Why does this elif exist???
# Are there plans to switch to v2?
interface_config = {
"mtu": MTU,
"match": {"macaddress": mac_address},
"dhcp6": False,
"dhcp4": is_primary,
}
if not is_primary:
interface_config["addresses"] = [
f"{vnic_dict['privateIp']}/{network.prefixlen}"
]
self._network_config["ethernets"][name] = interface_config
def _read_system_uuid() -> Optional[str]:
sys_uuid = dmi.read_dmi_data("system-uuid")
return None if sys_uuid is None else sys_uuid.lower()
def _is_platform_viable() -> bool:
asset_tag = dmi.read_dmi_data("chassis-asset-tag")
return asset_tag == CHASSIS_ASSET_TAG
def _fetch(metadata_version: int, path: str, retries: int = 2) -> dict:
return readurl(
url=METADATA_PATTERN.format(version=metadata_version, path=path),
headers=V2_HEADERS if metadata_version > 1 else None,
retries=retries,
)._response.json()
def read_opc_metadata(*, fetch_vnics_data: bool = False) -> OpcMetadata:
"""Fetch metadata from the /opc/ routes.
:return:
A namedtuple containing:
The metadata version as an integer
The JSON-decoded value of the instance data endpoint on the IMDS
The JSON-decoded value of the vnics data endpoint if
`fetch_vnics_data` is True, else None
"""
# Per Oracle, there are short windows (measured in milliseconds) throughout
# an instance's lifetime where the IMDS is being updated and may 404 as a
# result. To work around these windows, we retry a couple of times.
metadata_version = 2
try:
instance_data = _fetch(metadata_version, path="instance")
except UrlError:
metadata_version = 1
instance_data = _fetch(metadata_version, path="instance")
vnics_data = None
if fetch_vnics_data:
try:
vnics_data = _fetch(metadata_version, path="vnics")
except UrlError:
util.logexc(LOG, "Failed to fetch IMDS network configuration!")
return OpcMetadata(metadata_version, instance_data, vnics_data)
# Used to match classes to dependencies
datasources = [
(DataSourceOracle, (sources.DEP_FILESYSTEM,)),
]
# Return a list of data sources that match this set of dependencies
def get_datasource_list(depends):
return sources.list_from_depends(depends, datasources)
if __name__ == "__main__":
import argparse
description = """
Query Oracle Cloud metadata and emit a JSON object with two keys:
`read_opc_metadata` and `_is_platform_viable`. The values of each are
the return values of the corresponding functions defined in
DataSourceOracle.py."""
parser = argparse.ArgumentParser(description=description)
parser.parse_args()
print(
atomic_helper.json_dumps(
{
"read_opc_metadata": read_opc_metadata(),
"_is_platform_viable": _is_platform_viable(),
}
)
)