diff --git a/README.md b/README.md index 6023ca84..ead6b549 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,7 @@ Available source types: * AWS EC2, EBS, and network interfaces, with opt-in S3, Lambda, and EKS inventory (see [AWS source](docs/source_aws.md)) * oVirt / Red Hat Virtualization (see [oVirt source](docs/source_ovirt.md)) * [bb-ricardo/check_redfish](https://github.com/bb-Ricardo/check_redfish) inventory files +* [DigitalOcean](docs/source_digitalocean.md) **IMPORTANT: READ INSTRUCTIONS CAREFULLY BEFORE RUNNING THIS PROGRAM** diff --git a/docs/source_digitalocean.md b/docs/source_digitalocean.md new file mode 100644 index 00000000..8655e7e4 --- /dev/null +++ b/docs/source_digitalocean.md @@ -0,0 +1,193 @@ +# DigitalOcean source + +The `digitalocean` source reads Droplets, attached block volumes, and managed +database clusters from the DigitalOcean API and reconciles them into NetBox. +API access is read-only; inventory changes are written to NetBox by the normal +netbox-sync process. + +## Installation and access + +The source uses the `requests` dependency already installed by netbox-sync; no +DigitalOcean SDK is required. Create a DigitalOcean API token with read access +to regions, Droplets, and volumes. If managed database ingestion is enabled, the token +must also be able to read databases. + +Keep the API token out of version control and provide it through a protected +configuration file or the application's standard source environment variables. + +## Configuration + +Add a source section to the existing settings file: + +```ini +[source/example-digitalocean] +type = digitalocean +api_token = REPLACE_WITH_READ_ONLY_API_TOKEN +region_mapping = fra1=Germany, nyc1=United States +sync_volumes = True +sync_databases = True +; Optional isolation for overlapping private addresses across teams: +;private_ip_vrf = DigitalOcean Team-A +``` + +`api_token` is required. `sync_volumes` and `sync_databases` default to `True`. +Set either to `False` when the token should not query that API resource. +Accounts without managed databases can leave `sync_databases` enabled: +both `{"databases": []}` and `{"databases": null}` are treated as empty +database collections. Missing collection fields and other malformed values +remain errors rather than being treated as empty inventory. +Generate the complete configuration with: + +```sh +python netbox-sync.py -g +``` + +`region_mapping` is optional. It accepts comma-separated, exact +`DigitalOcean region slug=NetBox Region name` pairs. Quote destination names +containing commas, for example `nyc1="New York, USA"`. Duplicate source slugs, +empty entries, and Region names longer than 100 characters are rejected. + +On NetBox 4.2+, the source reuses a Region with the destination name, then +tries its generated slug, and creates it if neither exists. Explicit mappings +take precedence over Regions matching the DigitalOcean region. Without a +mapping, the DigitalOcean region name and slug are used (for example +`Frankfurt 1` and `fra1`). Region slugs (including those returned for databases) +are resolved through `GET /v2/regions` once per run, so cluster names do not +depend on whether Droplets exist in the region. If a slug is absent from that +response, it is used as the name. Existing Regions are not renamed. +Regions previously created by this source retain their source ownership tags; +reused, unmanaged Regions are not claimed by the source. + +Multiple DigitalOcean regions can map to the same NetBox Region; their +virtualization clusters remain separate and retain their DigitalOcean names. +Existing clusters in the source's cluster group are updated in place when +their scope changes, including previously unscoped clusters. Ambiguous +duplicate clusters in that group cause an explicit error instead of choosing +one arbitrarily. + +The NetBox token needs permission to add Regions when they do not already +exist. On NetBox versions before 4.2, Region creation and mapping are not +applied; clusters remain unscoped. + +## NetBox mapping + +### Private address isolation + +`private_ip_vrf` optionally names a NetBox VRF for this source's private +network addresses (IPv4 and IPv6). The source reuses a VRF with that exact +name or creates it if missing; names must be non-empty and at most 100 +characters. The NetBox token needs permission to create VRFs if missing. +An ambiguous VRF name stops the run rather than selecting one arbitrarily. +Referenced VRFs inherit source ownership through the normal inventory logic +and are not pruned. + +Use distinct names for isolated teams, for example: + +```ini +[source/team-a] +type = digitalocean +api_token = REPLACE_WITH_TEAM_A_READ_ONLY_TOKEN +private_ip_vrf = DigitalOcean Team-A + +[source/team-b] +type = digitalocean +api_token = REPLACE_WITH_TEAM_B_READ_ONLY_TOKEN +private_ip_vrf = DigitalOcean Team-B +``` + +Both teams can then have `10.108.0.6` in the same region without sharing an +IP record. IP matching and per-run conflict checks use the address and VRF; +prefix-length changes update the same record. Public network addresses remain +in the global table, regardless of their numerical address range. +When unset, private addresses also remain global, preserving existing behavior. +Each run logs the source name and whether private addresses use the global +table or the configured VRF, making it possible to verify the active settings. + +Enabling or changing this option does **not** move existing IPs between VRFs. +It creates or reuses records in the configured VRF and selects discovered +primary IPs from the appropriate table, preferring public addresses. +Existing global records and assignments remain unchanged; review and clean up +obsolete records manually after confirming ownership in a dry run. +Previously source-tagged global IPs may become orphaned and eligible for +normal pruning once their replacements are in a VRF. Disable pruning during +the transition and review the dry-run changes before enabling writes. +Turning the option off likewise does not move VRF records back to global. +This is per-source isolation, not per-VPC isolation: if a team has multiple +VPCs that reuse addresses, per-VPC isolation would be needed instead of this +setting. + +| DigitalOcean | NetBox | +| --- | --- | +| Region | Virtualization cluster scoped to a reused or automatically created NetBox Region on NetBox 4.2+ | +| Droplet | Virtual machine | +| Droplet size | VM vCPUs and memory | +| Droplet image distribution/name | VM platform | +| Droplet public/private IPv4 and IPv6 | VM interfaces and IP addresses | +| Droplet root disk | VM disk capacity on older NetBox; virtual disk on NetBox 3.7+ | +| Attached block volume | Additional virtual disk on NetBox 3.7+ | +| DigitalOcean resource tags | NetBox tags | +| Managed database cluster | VM inventory record with `db-` name prefix and metadata custom fields | + +DigitalOcean reports Droplet and volume disk sizes in gigabytes. NetBox 4.1+ +receives megabytes using 1024 MB per reported GB; earlier NetBox versions +receive gigabytes. Database storage, reported in MiB, is converted to whole +gigabytes on earlier NetBox versions. VM serial numbers are only sent on +NetBox 4.1+. Database +clusters are represented as one VM-like record per cluster because this +project has no native managed-database object type. Database engine, version, +size, node count, endpoint host, and port are stored in `digitalocean_*` custom +fields. Connection credentials and connection URIs are never copied. Database +endpoint hostnames are not resolved into IP addresses. + +DigitalOcean private and public network addresses are assigned to interfaces +named after their network type. The source does not create NetBox prefixes or +sites. + +VMs are matched by `digitalocean_resource_id` before falling back to name and +cluster for records without a resource ID. Duplicate VM names within a region +(case-insensitive, including database `db-` prefixes) have ` ()` +appended, with the original name shortened as needed to fit NetBox's limit. +Resource IDs preserve VM identity across renames and region changes. +If multiple VMs carry the same resource ID, the source uses the sole +non-orphaned match when all other matches have the `NetBox-synced: Orphaned` +tag. It logs the ignored orphaned records and leaves them unchanged for normal +orphan pruning. A single orphaned match can still be rediscovered. +If multiple non-orphaned matches, or only multiple orphaned matches, remain, +the source exits with the matching NetBox IDs, names, and clusters rather than +choosing arbitrarily. +Droplets with the same name in different clusters remain distinct. Interfaces +and virtual disks are matched to their resolved parent VM, not just its name. +If an earlier run moved child objects between same-named VMs, inspect the next +dry run for interface, disk, and primary-IP changes before enabling writes. + +IP assignments can be repaired automatically when the existing IP belongs to +the current source, either through this run's ownership or its persisted +`Source: ` tag. Conflicting or uncertain ownership is logged and +the assignment is skipped without stopping the run or changing the existing +assignment. Other addresses, droplets, disks, and databases continue syncing. +Skipped IPs are not selected as the discovered VM's primary IP. An existing IP +tagged for this source is retained for the run, preventing the skipped assignment +from causing orphaning/pruning; ownership tags are preserved. +Two interfaces claiming the same IP in one source run also produce an error log: +the first assignment is kept and subsequent conflicting assignments are skipped. +Source section names must remain stable for +persisted ownership tags to match. If an earlier run already removed an IP's +source tag or marked it orphaned, verify its owner and restore the correct tag +or repair its assignment manually before retrying. +Ownership-conflict errors include the existing NetBox IP ID, VRF, assigned +interface, IP source tags, current-run source, and assigned VM tags to help +identify stale assignments. VM tags alone do not authorize moving an IP. + +## Validation + +Start with a dry run: + +```sh +python netbox-sync.py -n -c settings.ini +``` + +Confirm the discovered Droplet, volume, and database cluster counts and inspect +the resulting NetBox object changes before enabling writes or pruning. +API failures stop discovery before resource inventory is modified, close the +API session, and report the HTTP status and DigitalOcean error message when +available. diff --git a/module/config/file_output.py b/module/config/file_output.py index b7191f86..4210d86b 100644 --- a/module/config/file_output.py +++ b/module/config/file_output.py @@ -19,6 +19,7 @@ from module.sources.hetzner.config import HetznerConfig from module.sources.ovirt.config import OVirtConfig from module.sources.aws.config import AWSConfig +from module.sources.digitalocean.config import DigitalOceanConfig from module.common.logging import get_logger from module.config import default_config_file_path, source_config_section_name from module.config.files import ConfigFile, ConfigFileINI, ConfigFileYAML @@ -39,6 +40,7 @@ class ConfigFileOutput(DescriptionFormatterMixin): CheckRedfishConfig, HetznerConfig, AWSConfig, + DigitalOceanConfig, OVirtConfig ] diff --git a/module/netbox/inventory.py b/module/netbox/inventory.py index af39be89..5900bd02 100644 --- a/module/netbox/inventory.py +++ b/module/netbox/inventory.py @@ -128,6 +128,14 @@ def get_by_data(self, object_type, data=None): # try to find by primary/secondary key elif data.get(object_type.primary_key) is not None: + if object_type in (NBVMInterface, NBVirtualDisk) and isinstance(data.get("virtual_machine"), NBVM): + # VM names are only unique within a cluster; use the resolved parent object. + for this_object in self.get_all_items(object_type): + if this_object.data.get("virtual_machine") is data["virtual_machine"] and \ + str(this_object.data.get(object_type.primary_key)).lower() == \ + str(data[object_type.primary_key]).lower(): + return this_object + return None object_name_to_find = None for this_object in self.get_all_items(object_type): diff --git a/module/netbox/object_classes.py b/module/netbox/object_classes.py index 9f87ce6c..1706ebfe 100644 --- a/module/netbox/object_classes.py +++ b/module/netbox/object_classes.py @@ -1511,7 +1511,6 @@ class NBRegion(NetBoxObject): object_type = "dcim.region" primary_key = "name" prune = False - read_only = True def __init__(self, *args, **kwargs): self.data_model = { diff --git a/module/sources/__init__.py b/module/sources/__init__.py index fe3e461b..ee45e4f1 100644 --- a/module/sources/__init__.py +++ b/module/sources/__init__.py @@ -12,6 +12,7 @@ from module.sources.hetzner.connection import HetznerHandler from module.sources.ovirt.connection import OVirtHandler from module.sources.aws.connection import AWSHandler +from module.sources.digitalocean.connection import DigitalOceanHandler from module.sources.check_redfish.import_inventory import CheckRedfish from module.common.logging import get_logger @@ -21,7 +22,7 @@ from module.config import source_config_section_name # list of valid sources -valid_sources = [VMWareHandler, CheckRedfish, HetznerHandler, AWSHandler, OVirtHandler] +valid_sources = [VMWareHandler, CheckRedfish, HetznerHandler, AWSHandler, DigitalOceanHandler, OVirtHandler] def validate_source(source_class_object=None, state="pre"): diff --git a/module/sources/digitalocean/__init__.py b/module/sources/digitalocean/__init__.py new file mode 100644 index 00000000..9591209d --- /dev/null +++ b/module/sources/digitalocean/__init__.py @@ -0,0 +1,8 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2020 - 2026 netbox-sync team. All rights reserved. +# +# This work is licensed under the terms of the MIT license. +# For a copy, see file LICENSE.txt included in this +# repository or visit: . + +from module.sources.digitalocean.connection import DigitalOceanHandler diff --git a/module/sources/digitalocean/client.py b/module/sources/digitalocean/client.py new file mode 100644 index 00000000..8c525380 --- /dev/null +++ b/module/sources/digitalocean/client.py @@ -0,0 +1,86 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2020 - 2026 netbox-sync team. All rights reserved. +# +# This work is licensed under the terms of the MIT license. +# For a copy, see file LICENSE.txt included in this +# repository or visit: . + +from urllib.parse import urljoin, urlparse + +import requests + + +class DigitalOceanClient: + + api_url = "https://api.digitalocean.com/v2/" + timeout = 30 + + def __init__(self, token, session=None): + self.session = session or requests.Session() + self.headers = { + "Accept": "application/json", + "Authorization": f"Bearer {token}", + "User-Agent": "netbox-sync", + } + + def get_collection(self, path, collection_name): + url = urljoin(self.api_url, path.lstrip("/")) + params = {"per_page": 200} + objects = [] + + while url is not None: + response = self.session.get( + url, + headers=self.headers, + params=params, + timeout=self.timeout, + ) + try: + response.raise_for_status() + except requests.HTTPError as exc: + detail = str(exc) + try: + error_payload = response.json() + except requests.exceptions.JSONDecodeError: + detail = f"{detail} (error response is not JSON)" + else: + if isinstance(error_payload, dict): + detail = error_payload.get("message") or detail + raise requests.HTTPError( + f"DigitalOcean API '{path}' returned HTTP {response.status_code}: {detail}", + response=response, + ) from exc + payload = response.json() + if not isinstance(payload, dict): + raise ValueError(f"DigitalOcean response for '{path}' is not a JSON object.") + collection = payload.get(collection_name) + # DigitalOcean can return {"databases": null} for accounts without databases. + if collection_name == "databases" and "databases" in payload and collection is None: + collection = [] + if not isinstance(collection, list): + raise ValueError(f"DigitalOcean response for '{path}' has no '{collection_name}' list.") + objects.extend(collection) + + links = payload.get("links") or {} + if not isinstance(links, dict): + raise ValueError(f"DigitalOcean response for '{path}' has invalid pagination links.") + pages = links.get("pages") or {} + if not isinstance(pages, dict): + raise ValueError(f"DigitalOcean response for '{path}' has invalid pagination pages.") + next_url = pages.get("next") + if next_url: + if not isinstance(next_url, str): + raise ValueError(f"DigitalOcean response for '{path}' has a non-string pagination link.") + url = urljoin(url, next_url) + parsed_url = urlparse(url) + if parsed_url.scheme != "https" or parsed_url.hostname != "api.digitalocean.com" \ + or parsed_url.port not in (None, 443) or parsed_url.username is not None: + raise ValueError("DigitalOcean API returned a pagination link outside its HTTPS API host.") + else: + url = None + params = None + + return objects + + def close(self): + self.session.close() diff --git a/module/sources/digitalocean/config.py b/module/sources/digitalocean/config.py new file mode 100644 index 00000000..ad05aafa --- /dev/null +++ b/module/sources/digitalocean/config.py @@ -0,0 +1,103 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2020 - 2026 netbox-sync team. All rights reserved. +# +# This work is licensed under the terms of the MIT license. +# For a copy, see file LICENSE.txt included in this +# repository or visit: . + +from module.common.logging import get_logger +from module.common.misc import quoted_split +from module.config import source_config_section_name +from module.config.base import ConfigBase +from module.config.option import ConfigOption + +log = get_logger() + + +class DigitalOceanConfig(ConfigBase): + + section_name = source_config_section_name + source_name_example = "my-digitalocean-example" + + def __init__(self): + self.options = [ + ConfigOption( + "enabled", + bool, + default_value=True, + description="Enable or disable the DigitalOcean source." + ), + ConfigOption( + "type", + str, + default_value="digitalocean", + description="Source type identifier. Must remain 'digitalocean'." + ), + ConfigOption( + "api_token", + str, + mandatory=True, + sensitive=True, + description="DigitalOcean API token with read access to the resources being synced." + ), + ConfigOption( + "region_mapping", + str, + config_example="fra1=Germany, nyc1=United States", + description="Optional comma-separated DigitalOcean region slug=NetBox Region name mappings. " + "Missing Regions are created on NetBox 4.2+. Unmapped regions use the " + "DigitalOcean region name and slug." + ), + ConfigOption( + "private_ip_vrf", + str, + config_example="DigitalOcean Team-A", + description="Optional NetBox VRF name for private network addresses. Reuses or creates the VRF. " + "Public addresses remain global. Unset keeps all addresses global; existing " + "global IP records are not migrated." + ), + ConfigOption( + "sync_volumes", + bool, + default_value=True, + description="Sync attached block volumes as virtual disks." + ), + ConfigOption( + "sync_databases", + bool, + default_value=True, + description="Sync managed database clusters as VM inventory records." + ), + ] + + super().__init__() + + def validate_options(self): + vrf_option = self.get_option_by_name("private_ip_vrf") + if vrf_option.value is not None: + name = vrf_option.value + if not isinstance(name, str) or not name.strip() or len(name.strip()) > 100: + log.error("Config option 'private_ip_vrf' must be a non-empty VRF name of at most 100 characters.") + self.set_validation_failed() + else: + vrf_option.set_value(name.strip()) + option = self.get_option_by_name("region_mapping") + mapping = {} + if option.value is not None: + if not isinstance(option.value, str): + log.error("Config option 'region_mapping' must be a comma-separated string of slug=name mappings.") + self.set_validation_failed() + return + + for entry in quoted_split(option.value): + slug, separator, name = entry.partition("=") + slug = slug.strip(' "\'') + name = name.strip(' "\'') + if not separator or not slug or not name or len(name) > 100 or slug in mapping: + log.error(f"Invalid region_mapping entry '{entry}': expected a unique region slug and " + "a non-empty NetBox Region name of at most 100 characters (slug=name).") + self.set_validation_failed() + continue + mapping[slug] = name + + option.set_value(mapping) diff --git a/module/sources/digitalocean/connection.py b/module/sources/digitalocean/connection.py new file mode 100644 index 00000000..f33948cd --- /dev/null +++ b/module/sources/digitalocean/connection.py @@ -0,0 +1,607 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2020 - 2026 netbox-sync team. All rights reserved. +# +# This work is licensed under the terms of the MIT license. +# For a copy, see file LICENSE.txt included in this +# repository or visit: . + +from collections import Counter +from ipaddress import ip_interface +from packaging import version +import requests + +from module.common.logging import get_logger +from module.common.misc import do_error_exit +from module.netbox import ( + NBCustomField, NBCluster, NBClusterGroup, NBClusterType, NBIPAddress, NBPlatform, + NBRegion, NBTag, NBVM, NBVMInterface, NBVirtualDisk, NBVRF, NetBoxObject, +) +from module.netbox.inventory import NetBoxInventory +from module.netbox.connection import NetBoxHandler +from module.sources.common.source_base import SourceBase +from module.sources.digitalocean.client import DigitalOceanClient +from module.sources.digitalocean.config import DigitalOceanConfig + +log = get_logger() + + +class DigitalOceanHandler(SourceBase): + + source_type = "digitalocean" + settings = DigitalOceanConfig() + + dependent_netbox_objects = [ + NBCustomField, NBCluster, NBClusterGroup, NBClusterType, NBIPAddress, NBPlatform, + NBRegion, NBTag, NBVM, NBVMInterface, NBVirtualDisk, NBVRF, + ] + + def __init__(self, name=None): + if not name: + raise ValueError(f"Invalid value for attribute 'name': '{name}'.") + + self.inventory = NetBoxInventory() + self.name = name + self.log = get_logger() + self._ip_claims = {} + self._regions_by_slug = {} + self._vm_name_counts = Counter() + self.set_source_tag() + + settings_handler = DigitalOceanConfig() + settings_handler.source_name = name + self.settings = settings_handler.parse() + + if self.settings.enabled is False: + log.info(f"Source '{name}' is currently disabled. Skipping") + return + + self.init_successful = True + + def _abort(self, message): + log.error(message) + do_error_exit(message) + + def apply(self): + self._ip_claims = {} + client = DigitalOceanClient(self.settings.api_token) + + # Fetch everything before touching inventory so API failures cannot produce + # a partial run that later looks like missing resources. + try: + regions = client.get_collection("regions", "regions") + droplets = client.get_collection("droplets", "droplets") + volumes = client.get_collection("volumes", "volumes") if self.settings.sync_volumes else [] + databases = client.get_collection("databases", "databases") if self.settings.sync_databases else [] + except (requests.RequestException, ValueError) as exc: + self._abort(f"DigitalOcean discovery failed for source '{self.name}': {exc}") + finally: + client.close() + + self._regions_by_slug = {region["slug"]: region for region in regions} + self._current_resource_ids = { + str(resource["id"]) for resource in (*droplets, *databases) if resource.get("id") is not None + } + self.log.info( + f"Connected to DigitalOcean, found {len(droplets)} droplets, " + f"{len(volumes)} volumes and {len(databases)} database clusters" + ) + if self.settings.private_ip_vrf is None: + self.log.info( + f"DigitalOcean source '{self.name}': private_ip_vrf is unset; " + "private addresses use the global table." + ) + else: + self.log.info( + f"DigitalOcean source '{self.name}': private addresses use VRF " + f"'{self.settings.private_ip_vrf}'; public addresses use the global table." + ) + + cluster_group = self.inventory.add_update_object( + NBClusterGroup, + data={ + "name": f"DigitalOcean: {self.name}", + "slug": NetBoxObject.format_slug(f"digitalocean-{self.name}"), + }, + source=self, + ) + cluster_type = self.inventory.add_update_object( + NBClusterType, + data={"name": "DigitalOcean", "slug": "digitalocean"}, + source=self, + ) + volumes_by_id = {str(volume["id"]): volume for volume in volumes} + clusters = {} + resources = [] + + for resource_type, collection in (("droplet", droplets), ("database", databases)): + for resource in collection: + cluster = self._get_cluster(resource.get("region"), cluster_group, cluster_type, clusters) + if cluster is None: + log.warning(f"DigitalOcean {resource_type} '{resource.get('name')}' has no region; skipping.") + continue + resources.append((resource_type, resource, cluster)) + self._vm_name_counts = Counter( + (cluster, self._resource_name(resource, resource_type)[:64].casefold()) + for resource_type, resource, cluster in resources + ) + for resource_type, resource, cluster in resources: + if resource_type == "droplet": + vm = self._sync_droplet(resource, cluster) + self._sync_droplet_network(vm, resource) + self._sync_droplet_disks(vm, resource, volumes_by_id) + else: + self._sync_database(resource, cluster) + + def _get_cluster(self, region, cluster_group, cluster_type, clusters): + region_slug = region.get("slug") if isinstance(region, dict) else region + region = self._regions_by_slug.get(region_slug, region) + if isinstance(region, dict): + region_slug = region.get("slug") or region.get("name") + region_name = region.get("name") or region_slug + else: + region_slug = region + region_name = region + if not region_slug: + return None + + key = str(region_slug) + if key in clusters: + return clusters[key] + + cluster_data = { + "name": f"{self.name}: {region_name}"[:100], + "type": cluster_type, + "group": cluster_group, + } + if version.parse(self.inventory.netbox_api_version) >= version.parse("4.2.0"): + mapped_name = self.settings.region_mapping.get(key) + target_name = mapped_name or region_name + target_slug = NetBoxObject.format_slug(mapped_name) if mapped_name else key + regions = self.inventory.get_all_items(NBRegion) + netbox_region = next(( + item for item in regions if item.data.get("name") == target_name + ), None) + if netbox_region is None: + netbox_region = next(( + item for item in regions + if item.data.get("slug") == target_slug or + (mapped_name is None and item.data.get("name") == key) + ), None) + if netbox_region is None: + netbox_region = self.inventory.add_object( + NBRegion, data={"name": target_name, "slug": target_slug}, source=self, + ) + elif self.source_tag in netbox_region.get_tags(): + netbox_region.set_source(self) + cluster_data["scope_type"] = NBRegion.object_type + cluster_data["scope_id"] = netbox_region + + # Scope changes must update this source's cluster rather than create a duplicate. + candidates = [ + item for item in self.inventory.get_all_items(NBCluster) + if item.data.get("name") == cluster_data["name"] and item.data.get("group") is cluster_group + ] + if len(candidates) > 1: + message = f"Multiple DigitalOcean clusters match '{cluster_data['name']}' in the source cluster group." + log.error(message) + raise ValueError(message) + if candidates: + cluster = candidates[0] + cluster.update(data=cluster_data, source=self) + else: + cluster = self.inventory.add_update_object(NBCluster, data=cluster_data, source=self) + clusters[key] = cluster + return cluster + + def _sync_droplet(self, droplet, cluster): + name = str(droplet.get("name", "")).strip() + if not name: + log.warning(f"Skipping DigitalOcean droplet without a name (ID: {droplet.get('id')}).") + return None + + size = droplet.get("size") or {} + memory = droplet.get("memory") + if memory is None: + memory = size.get("memory") + vcpus = droplet.get("vcpus") + if vcpus is None: + vcpus = size.get("vcpus") + disk_gb = droplet.get("disk") + if disk_gb is None: + disk_gb = size.get("disk") + vm_data = { + "name": self._vm_name(droplet, cluster, "droplet"), + "cluster": cluster, + "status": "active" if droplet.get("status") == "active" else "offline", + "vcpus": vcpus, + "memory": memory, + } + if version.parse(self.inventory.netbox_api_version) >= version.parse("4.1.0"): + vm_data["serial"] = str(droplet.get("id")) if droplet.get("id") is not None else None + if disk_gb is not None and version.parse(self.inventory.netbox_api_version) < version.parse("3.7.0"): + vm_data["disk"] = int(disk_gb) + + platform = self._droplet_platform(droplet.get("image")) + if platform: + platform_name = platform[:100] + vm_data["platform"] = self.inventory.add_update_object( + NBPlatform, + data={"name": platform_name, "slug": NetBoxObject.format_slug(platform_name)}, + source=self, + ) + tags = [tag[:100] for tag in (droplet.get("tags") or []) if isinstance(tag, str) and tag] + if tags: + self._ensure_tags(tags) + vm_data["tags"] = tags + + custom_fields = self._custom_fields({ + "digitalocean_resource_type": "droplet", + "digitalocean_resource_id": droplet.get("id"), + }) + if custom_fields: + vm_data["custom_fields"] = custom_fields + + return self._sync_vm(vm_data) + + @staticmethod + def _resource_name(resource, resource_type): + name = str(resource.get("name", "")).strip() + return f"db-{name}" if resource_type == "database" else name + + def _vm_name(self, resource, cluster, resource_type): + name = self._resource_name(resource, resource_type) + if self._vm_name_counts[(cluster, name[:64].casefold())] > 1: + suffix = f" ({resource['id']})" + return name[:64 - len(suffix)] + suffix + return name[:64] + + def _sync_vm(self, data): + identity = (data.get("custom_fields") or {}).get("digitalocean_resource_id") + cluster = data["cluster"] + group = cluster.data.get("group") + candidates = [ + vm for vm in self.inventory.get_all_items(NBVM) + if identity is not None and + (vm.data.get("custom_fields") or {}).get("digitalocean_resource_id") == identity and + (vm.source is self or self.source_tag in vm.get_tags() or vm.data.get("cluster") is cluster or + (group is not None and vm.data.get("cluster") is not None and + vm.data["cluster"].data.get("group") is group)) + ] + if len(candidates) > 1: + current = [vm for vm in candidates if NetBoxHandler.orphaned_tag not in vm.get_tags()] + if len(current) == 1: + for orphan in candidates: + if orphan is current[0]: + continue + log.warning( + f"Ignoring orphaned DigitalOcean VM '{orphan.get_display_name()}' " + f"(NetBox ID {orphan.nb_id}) with resource ID '{identity}' for source '{self.name}'; " + f"using '{current[0].get_display_name()}' (NetBox ID {current[0].nb_id}). " + "The orphaned VM is left for normal orphan pruning." + ) + candidates = current + else: + details = "; ".join( + f"'{vm.get_display_name()}' (NetBox ID {vm.nb_id}, " + f"cluster '{vm.data['cluster'].get_display_name() if vm.data.get('cluster') else None}', " + f"orphaned={NetBoxHandler.orphaned_tag in vm.get_tags()})" + for vm in candidates + ) + self._abort( + f"Multiple DigitalOcean VMs match resource ID '{identity}' for source '{self.name}': " + f"{details}. Resolve the duplicate records in NetBox before retrying." + ) + vm = candidates[0] if candidates else self.inventory.get_by_data(NBVM, data=data) + if not candidates and vm is not None: + existing_id = (vm.data.get("custom_fields") or {}).get("digitalocean_resource_id") + if existing_id is not None and existing_id != identity and existing_id in self._current_resource_ids: + vm = None + data["name"] = self._unused_vm_name(data["name"], cluster, vm, identity) + if vm is None: + return self.inventory.add_object(NBVM, data=data, source=self) + vm.update(data=data, source=self) + return vm + + def _unused_vm_name(self, name, cluster, vm, identity): + """Add the resource ID if another VM in the cluster still uses the name, e.g. one of a deleted resource.""" + if identity is None: + return name + suffix = f" ({identity})" + if name.endswith(suffix): + return name + for other in self.inventory.get_all_items(NBVM): + if other is not vm and other.data.get("cluster") is cluster and \ + str(other.data.get("name", "")).casefold() == name.casefold(): + return name[:64 - len(suffix)] + suffix + return name + + @staticmethod + def _droplet_platform(image): + if not isinstance(image, dict): + return None + distribution = image.get("distribution") + image_name = image.get("name") + if image_name and distribution and distribution.lower() not in image_name.lower(): + return f"{distribution} {image_name}" + return image_name or distribution + + def _sync_droplet_network(self, vm, droplet): + if vm is None: + return + + networks = droplet.get("networks") or {} + primary = {4: None, 6: None} + primary_is_public = {4: False, 6: False} + for family in ("v4", "v6"): + for address in networks.get(family, []) or []: + ip_value = address.get("ip_address") + if ip_value is None: + continue + ip_value = str(ip_value) + interface_name = str(address.get("type") or "network").lower() + try: + ip_obj = self._parse_network_address(ip_value, address) + except ValueError: + log.warning(f"Invalid IP address '{ip_value}' on DigitalOcean droplet '{vm.data['name']}'.") + continue + + interface = self.inventory.add_update_object( + NBVMInterface, + data={ + "name": interface_name[:64], + "virtual_machine": vm, + "enabled": True, + }, + source=self, + ) + vrf = self._get_private_ip_vrf() if interface_name == "private" else None + ip_object = self._sync_ip(ip_obj, interface, vrf=vrf) + if ip_object is None: + continue + ip_version = ip_obj.version + is_public = interface_name == "public" + if primary[ip_version] is None or (is_public and not primary_is_public[ip_version]): + primary[ip_version] = ip_object + primary_is_public[ip_version] = is_public + + primary_data = {} + if primary[4] is not None: + primary_data["primary_ip4"] = primary[4] + if primary[6] is not None: + primary_data["primary_ip6"] = primary[6] + if primary_data: + vm.update(data=primary_data, source=self) + + @staticmethod + def _parse_network_address(ip_value, address): + prefix = address.get("netmask") + if prefix is None: + prefix = address.get("cidr") + if prefix is None: + prefix = 32 if ":" not in ip_value else 128 + return ip_interface(f"{ip_value}/{prefix}") + + def _get_private_ip_vrf(self): + name = self.settings.private_ip_vrf + if name is None: + return None + candidates = [vrf for vrf in self.inventory.get_all_items(NBVRF) if vrf.data.get("name") == name] + if len(candidates) > 1: + self._abort(f"Multiple NetBox VRFs match private_ip_vrf '{name}' for source '{self.name}'.") + if candidates: + vrf = candidates[0] + if self.source_tag in vrf.get_tags(): + vrf.set_source(self) + return vrf + return self.inventory.add_object(NBVRF, data={"name": name}, source=self) + + def _sync_ip(self, ip_obj, interface, vrf=None): + claim_key = (str(ip_obj.ip), vrf) + previous_claim = self._ip_claims.get(claim_key) + if previous_claim is not None and previous_claim is not interface: + message = (f"DigitalOcean source '{self.name}' reports IP '{ip_obj}' on multiple interfaces. " + f"Skipping assignment to '{interface.get_display_name()}'; keeping the first assignment.") + log.error(message) + return None + + ip_data = { + "address": str(ip_obj), + "assigned_object_type": "virtualization.vminterface", + "assigned_object_id": interface, + } + if vrf is not None: + ip_data["vrf"] = vrf + candidates = [ + item for item in self.inventory.get_all_items(NBIPAddress) + if ip_interface(item.data["address"]).ip == ip_obj.ip and item.data.get("vrf") is vrf + ] + if len(candidates) > 1: + assigned = [item for item in candidates if item.data.get("assigned_object_id") is interface] + if len(assigned) != 1: + vrf_name = vrf.get_display_name() if vrf is not None else "global" + hint = "" + if vrf is None and ip_obj.ip.is_private: + hint = " Set 'private_ip_vrf' to keep DigitalOcean private addresses apart." + log.error( + f"Multiple NetBox IPs match '{ip_obj.ip}' in VRF '{vrf_name}' for source '{self.name}': " + f"{[item.nb_id for item in candidates]}. Skipping assignment to " + f"'{interface.get_display_name()}'; resolve the duplicate records.{hint}" + ) + return None + candidates = assigned + existing = candidates[0] if candidates else None + if existing is not None: + assigned_object = existing.data.get("assigned_object_id") + if assigned_object is not None and assigned_object is not interface: + source_tags = {tag for tag in existing.get_tags() if tag.startswith("Source: ")} + same_source = existing.source is self or self.source_tag in source_tags + other_source = ( + (existing.source is not None and existing.source is not self) or + bool(source_tags - {self.source_tag}) + ) + if not same_source or other_source: + assigned_vm = ( + assigned_object.data.get("virtual_machine") + if isinstance(assigned_object, NBVMInterface) else None + ) + vrf = existing.data.get("vrf") + vrf_name = vrf.get_display_name() if isinstance(vrf, NetBoxObject) else vrf + owner_name = getattr(existing.source, "name", None) + owner_vm_tags = sorted(assigned_vm.get_tags()) if assigned_vm is not None else [] + assigned_name = ( + assigned_object.get_display_name() + if isinstance(assigned_object, NetBoxObject) else str(assigned_object) + ) + assigned_cluster = assigned_vm.data.get("cluster") if assigned_vm is not None else None + assigned_cluster_name = ( + assigned_cluster.get_display_name() if assigned_cluster is not None else None + ) + message = ( + f"DigitalOcean IP '{ip_obj}' has conflicting or uncertain source ownership; " + f"cannot reassign it for source '{self.name}'. " + f"Existing NetBox IP ID {existing.nb_id}, VRF={vrf_name!r}, " + f"assigned to '{assigned_name}', " + f"assigned interface ID {getattr(assigned_object, 'nb_id', None)}, " + f"assigned VM ID {assigned_vm.nb_id if assigned_vm is not None else None}, " + f"assigned cluster={assigned_cluster_name!r}, target interface ID {interface.nb_id}, " + f"IP source tags={sorted(source_tags)}, current-run source={owner_name!r}, " + f"assigned VM tags={owner_vm_tags}. " + f"Skipping assignment to '{interface.get_display_name()}'; resolve the existing assignment." + ) + log.error(message) + if existing.source is None and self.source_tag in source_tags: + # Retain previously managed IPs without changing their assignment or ownership tags. + existing.set_source(self) + return None + log.info(f"Repairing DigitalOcean IP '{ip_obj}' assignment for source '{self.name}' " + f"to interface '{interface.get_display_name()}'.") + if existing is None: + result = self.inventory.add_object(NBIPAddress, data=ip_data, source=self) + else: + existing.update(data=ip_data, source=self) + result = existing + self._ip_claims[claim_key] = interface + return result + + def _sync_droplet_disks(self, vm, droplet, volumes_by_id): + if vm is None: + return + + disk_gb = droplet.get("disk") + if disk_gb is None: + disk_gb = (droplet.get("size") or {}).get("disk") + disk_factor = 1024 if version.parse(self.inventory.netbox_api_version) >= version.parse("4.1.0") else 1 + total_disk = int(disk_gb or 0) * disk_factor + attached_volumes = [ + volumes_by_id[str(volume_id)] + for volume_id in (droplet.get("volume_ids") or []) + if str(volume_id) in volumes_by_id + ] + total_disk += sum(int(volume.get("size_gigabytes", 0)) * disk_factor for volume in attached_volumes) + + if version.parse(self.inventory.netbox_api_version) < version.parse("3.7.0"): + vm.update(data={"disk": total_disk}, source=self) + return + + if disk_gb is not None: + self._sync_virtual_disk( + vm, "root", int(disk_gb) * disk_factor, "DigitalOcean Droplet root disk" + ) + for volume in attached_volumes: + self._sync_virtual_disk( + vm, + f"do-{volume['id']}"[:64], + int(volume.get("size_gigabytes", 0)) * disk_factor, + volume.get("name"), + ) + + def _sync_database(self, database, cluster): + name = str(database.get("name", "")).strip() + if not name: + log.warning(f"Skipping DigitalOcean database cluster without a name (ID: {database.get('id')}).") + return + + vm_data = { + "name": self._vm_name(database, cluster, "database"), + "cluster": cluster, + "status": "active" if database.get("status") == "online" else "offline", + "comments": "DigitalOcean managed database cluster", + } + if version.parse(self.inventory.netbox_api_version) >= version.parse("4.1.0"): + vm_data["serial"] = str(database.get("id")) if database.get("id") is not None else None + engine = database.get("engine") + db_version = database.get("version") + platform = " ".join(str(value) for value in (engine, db_version) if value) + if platform: + platform_name = platform[:100] + vm_data["platform"] = self.inventory.add_update_object( + NBPlatform, + data={"name": platform_name, "slug": NetBoxObject.format_slug(platform_name)}, + source=self, + ) + tags = [tag[:100] for tag in (database.get("tags") or []) if isinstance(tag, str) and tag] + if tags: + self._ensure_tags(tags) + vm_data["tags"] = tags + + connection = database.get("private_connection") or database.get("connection") or {} + custom_fields = self._custom_fields({ + "digitalocean_resource_type": "database", + "digitalocean_resource_id": database.get("id"), + "digitalocean_engine": engine, + "digitalocean_version": db_version, + "digitalocean_size": database.get("size"), + "digitalocean_node_count": database.get("num_nodes"), + "digitalocean_endpoint": connection.get("host"), + "digitalocean_port": connection.get("port"), + }) + if custom_fields: + vm_data["custom_fields"] = custom_fields + vm = self._sync_vm(vm_data) + + storage_size = database.get("storage_size_mib") + if storage_size is not None: + storage_size = int(storage_size) + if version.parse(self.inventory.netbox_api_version) < version.parse("4.1.0"): + storage_size //= 1024 + if version.parse(self.inventory.netbox_api_version) < version.parse("3.7.0"): + vm.update(data={"disk": storage_size}, source=self) + else: + self._sync_virtual_disk(vm, "database-storage", storage_size, "Managed database storage") + + def _custom_fields(self, values): + custom_fields = {} + for field_name, value in values.items(): + if value is None: + continue + self.add_update_custom_field({ + "name": field_name, + "label": field_name.replace("digitalocean_", "DO ").replace("_", " ")[:50], + "object_types": [NBVM.object_type], + "type": "text", + "description": f"DigitalOcean metadata for source '{self.name}'", + }) + custom_fields[field_name] = str(value) + return custom_fields + + def _ensure_tags(self, tags): + for tag in tags: + if not isinstance(tag, str) or not tag: + continue + self.inventory.add_update_object( + NBTag, + data={"name": tag, "slug": NetBoxObject.format_slug(tag)}, + source=self, + ) + + def _sync_virtual_disk(self, vm, name, size, description=None): + self.inventory.add_update_object( + NBVirtualDisk, + data={ + "name": name, + "virtual_machine": vm, + "size": size, + "description": description, + }, + source=self, + ) diff --git a/settings-example.ini b/settings-example.ini index 76f435ae..0e2765e8 100644 --- a/settings-example.ini +++ b/settings-example.ini @@ -755,6 +755,33 @@ regions = auto ;organization_account_ids = 111122223333 ; Leave organization_external_id unset unless the member trust policy requires it. +[source/my-digitalocean-example] + +; Enable or disable the DigitalOcean source. +;enabled = True + +; Source type identifier. Must remain 'digitalocean'. +;type = digitalocean + +; DigitalOcean API token with read access to the resources being synced. +api_token = + +; Optional comma-separated DigitalOcean region slug=NetBox Region name mappings. Missing +; Regions are created on NetBox 4.2+. Unmapped regions use the DigitalOcean region name +; and slug. +;region_mapping = fra1=Germany, nyc1=United States + +; Optional NetBox VRF name for private network addresses. Reuses or creates the VRF. +; Public addresses remain global. Unset keeps all addresses global; existing global IP +; records are not migrated. +;private_ip_vrf = DigitalOcean Team-A + +; Sync attached block volumes as virtual disks. +;sync_volumes = True + +; Sync managed database clusters as VM inventory records. +;sync_databases = True + [source/my-ovirt-example] ; Optional SDK: pip install -r requirements-ovirt.txt diff --git a/tests/test_digitalocean_source.py b/tests/test_digitalocean_source.py new file mode 100644 index 00000000..92e3db03 --- /dev/null +++ b/tests/test_digitalocean_source.py @@ -0,0 +1,1198 @@ +from configparser import ConfigParser as INIParser +from copy import deepcopy +from ipaddress import ip_interface +from pathlib import Path +from unittest.mock import Mock + +import pytest +import requests + +from module.config.file_output import ConfigFileOutput +from module.config.files import ConfigFileINI, ConfigFileYAML +from module.config.parser import ConfigParser +from module.netbox import ( + NBCluster, NBClusterGroup, NBCustomField, NBIPAddress, NBPlatform, NBRegion, NBTag, NBVM, NBVMInterface, + NBVirtualDisk, NBVRF, +) +from module.netbox.connection import NetBoxHandler +from module.sources import instantiate_sources, valid_sources +from module.sources.aws.config import AWSConfig +from module.sources.aws.connection import AWSHandler +from module.sources.digitalocean.config import DigitalOceanConfig +from module.sources.digitalocean.connection import DigitalOceanHandler +from module.sources.digitalocean.client import DigitalOceanClient + + +@pytest.fixture +def digitalocean(inventory, monkeypatch): + parser = ConfigParser() + monkeypatch.setattr(parser, "content", { + "source": {"cloud": {"type": "digitalocean", "api_token": "test-token"}} + }) + monkeypatch.setattr(parser, "parsing_finished", True) + client = Mock() + client.get_collection.side_effect = lambda path, key: { + "regions": [{"slug": "nyc1", "name": "New York 1"}], + "droplets": [dict( + id=123, + name="web-1", + status="active", + region={"slug": "nyc1", "name": "New York 1"}, + memory=2048, + vcpus=2, + disk=25, + image={"distribution": "Ubuntu", "name": "Ubuntu 24.04"}, + tags=["production", "web"], + volume_ids=["volume-1"], + networks={ + "v4": [ + {"ip_address": "10.10.0.5", "netmask": "255.255.255.0", "type": "private"}, + {"ip_address": "203.0.113.10", "netmask": "255.255.255.0", "type": "public"}, + ], + "v6": [ + {"ip_address": "2001:db8::5", "cidr": 64, "type": "public"}, + ], + }, + )], + "volumes": [dict( + id="volume-1", + name="database-data", + size_gigabytes=20, + droplet_ids=[123], + )], + "databases": [dict( + id="database-uuid", + name="orders", + region="nyc1", + status="online", + engine="pg", + version="16", + size="db-s-2vcpu-4gb", + num_nodes=2, + storage_size_mib=102400, + private_connection={ + "host": "db-private.example.org", + "port": 25060, + "uri": "postgres://user:secret@db-private.example.org:25060/db", + }, + connection={ + "host": "db-public.example.org", + "port": 25060, + "password": "do-not-copy", + }, + tags=["production"], + )], + }[path] + monkeypatch.setattr( + "module.sources.digitalocean.connection.DigitalOceanClient", + Mock(return_value=client), + ) + return inventory, client + + +def test_source_registration_and_settings_generation(monkeypatch): + assert DigitalOceanHandler in valid_sources + assert AWSHandler in valid_sources + assert DigitalOceanConfig in ConfigFileOutput.source_config_list + assert AWSConfig in ConfigFileOutput.source_config_list + assert DigitalOceanHandler.implements("digitalocean") + + for file_type in (ConfigFileINI, ConfigFileYAML): + output = ConfigFileOutput.__new__(ConfigFileOutput) + output.config_file_type = file_type + output.comment_prefix = file_type.comment_prefix + output.lines = [] + output.format() + text = "\n".join(output.lines) + assert "my-digitalocean-example" in text + assert "sync_databases" in text + assert "api_token" in text + assert "region_mapping" in text + + +def test_configured_source_is_instantiated(digitalocean): + inventory, _ = digitalocean + + source, = instantiate_sources() + + assert isinstance(source, DigitalOceanHandler) + assert source.init_successful + assert source.name == "cloud" + assert inventory.source_list == [source] + + +def test_settings_example_includes_all_digitalocean_options(): + parser = INIParser() + text = (Path(__file__).parents[1] / "settings-example.ini").read_text() + parser.read_string(text) + section_name = f"source/{DigitalOceanConfig.source_name_example}" + + assert section_name in parser + assert "source/my-aws-example" in parser + section = text.split(f"[{section_name}]", 1)[1] + for option in DigitalOceanConfig().options: + assert any( + line.lstrip(";").startswith(f"{option.key} =") + for line in section.splitlines() + ), option.key + + +@pytest.mark.parametrize("mapping, expected", [ + (None, {}), + ("fra1=Germany, nyc1=United States", {"fra1": "Germany", "nyc1": "United States"}), + (' fra1 = "Germany", nyc1 = "New York, USA" ', {"fra1": "Germany", "nyc1": "New York, USA"}), +]) +def test_region_mapping_configuration(mapping, expected): + config = DigitalOceanConfig() + config.source_name = "cloud" + config.config_content = {"source": {"cloud": {"api_token": "test-token", "region_mapping": mapping}}} + + assert config.parse(do_log=False).region_mapping == expected + + +@pytest.mark.parametrize("mapping", [ + "fra1", "=Germany", "fra1=", "fra1=Germany,", + "fra1=Germany,fra1=Europe", ["fra1=Germany"], "fra1=" + "x" * 101, +]) +def test_invalid_region_mapping_is_rejected(mapping, caplog): + config = DigitalOceanConfig() + config.source_name = "cloud" + config.config_content = {"source": {"cloud": {"api_token": "test-token", "region_mapping": mapping}}} + + with pytest.raises(SystemExit): + config.parse(do_log=False) + + assert "region_mapping" in caplog.text + + +@pytest.mark.parametrize("region, name, slug", [ + ({"slug": "fra1", "name": "Frankfurt 1"}, "Frankfurt 1", "fra1"), + ("fra1", "fra1", "fra1"), +]) +def test_missing_region_is_created_and_reused(digitalocean, region, name, slug, caplog): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + + cluster = source._get_cluster(region, None, None, {}) + netbox_region = cluster.data["scope_id"] + assert netbox_region.data["name"] == name + assert netbox_region.data["slug"] == slug + assert netbox_region.read_only is False + assert {"name", "slug"} <= set(netbox_region.updated_items) + assert cluster.data["scope_type"] == NBRegion.object_type + assert source._get_cluster(region, None, None, {}) is cluster + assert len(inventory.get_all_items(NBRegion)) == 1 + assert "Unable to determine second key" not in caplog.text + + +@pytest.mark.parametrize("existing_name, existing_slug", [ + ("Germany", "de"), ("Federal Republic", "germany"), (None, None), +]) +def test_region_mapping_overrides_provider_match(digitalocean, existing_name, existing_slug): + inventory, _ = digitalocean + provider_region = inventory.add_object( + NBRegion, data={"name": "Frankfurt 1", "slug": "fra1"}, read_from_netbox=True, + ) + existing = None + if existing_name is not None: + existing = inventory.add_object( + NBRegion, data={"name": existing_name, "slug": existing_slug}, read_from_netbox=True, + ) + source = DigitalOceanHandler("cloud") + source.settings.region_mapping = {"fra1": "Germany", "fra2": "Germany"} + cluster = source._get_cluster({"slug": "fra1", "name": "Frankfurt 1"}, None, None, {}) + + mapped_region = cluster.data["scope_id"] + assert mapped_region is not provider_region + if existing is not None: + assert mapped_region is existing + assert mapped_region.data["name"] == existing_name + assert mapped_region.data["slug"] == existing_slug + assert not mapped_region.updated_items + else: + assert mapped_region.data["name"] == "Germany" + assert mapped_region.data["slug"] == "germany" + other_cluster = source._get_cluster({"slug": "fra2", "name": "Frankfurt 2"}, None, None, {}) + assert other_cluster.data["scope_id"] is mapped_region + assert len(inventory.get_all_items(NBRegion)) == 2 + + +@pytest.mark.parametrize("previously_unscoped", [True, False]) +def test_existing_source_cluster_is_rescoped_without_duplicates(digitalocean, previously_unscoped): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + source.apply() + cluster = inventory.get_all_items(NBCluster)[0] + original_vm = inventory.get_all_items(NBVM)[0] + if previously_unscoped: + cluster.data.pop("scope_id") + cluster.data.pop("scope_type") + + source.settings.region_mapping = {"nyc1": "United States"} + source.apply() + + assert inventory.get_all_items(NBCluster) == [cluster] + assert original_vm.data["cluster"] is cluster + assert cluster.data["scope_id"].data["name"] == "United States" + + +def test_old_netbox_does_not_create_region_scopes(digitalocean): + inventory, _ = digitalocean + inventory.netbox_api_version = "4.1.11" + source = DigitalOceanHandler("cloud") + source.settings.region_mapping = {"fra1": "Germany"} + + cluster = source._get_cluster({"slug": "fra1", "name": "Frankfurt 1"}, None, None, {}) + + assert "scope_id" not in cluster.data + assert "scope_type" not in cluster.data + assert not inventory.get_all_items(NBRegion) + + +@pytest.mark.parametrize("name, slug", [ + ("Frankfurt 1", "frankfurt"), ("Existing Frankfurt", "fra1"), ("fra1", "existing"), +]) +def test_unmapped_region_reuses_existing_name_or_slug(digitalocean, name, slug): + inventory, _ = digitalocean + existing = inventory.add_object( + NBRegion, data={"name": name, "slug": slug}, read_from_netbox=True, + ) + source = DigitalOceanHandler("cloud") + + cluster = source._get_cluster({"slug": "fra1", "name": "Frankfurt 1"}, None, None, {}) + + assert cluster.data["scope_id"] is existing + assert existing.data["name"] == name + assert existing.data["slug"] == slug + assert not existing.updated_items + assert len(inventory.get_all_items(NBRegion)) == 1 + + +def test_region_is_written_before_cluster_with_integer_scope_id(digitalocean): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + cluster = source._get_cluster({"slug": "fra1", "name": "Frankfurt 1"}, None, None, {}) + region = cluster.data["scope_id"] + handler = NetBoxHandler.__new__(NetBoxHandler) + handler.inventory = inventory + handler.request = Mock(side_effect=lambda object_type, req_type, data, nb_id: { + **data, "id": 101 if object_type is NBRegion else 201, + }) + + handler.update_instance() + + calls = handler.request.call_args_list + assert [call.args[0] for call in calls] == [NBRegion, NBCluster] + assert all(call.kwargs["req_type"] == "POST" for call in calls) + assert calls[0].kwargs["data"]["name"] == "Frankfurt 1" + assert calls[1].kwargs["data"]["scope_id"] == 101 + assert calls[1].kwargs["data"]["scope_type"] == NBRegion.object_type + assert cluster.data["scope_id"] is region + assert not region.is_new + assert not cluster.is_new + assert not region.updated_items + assert not cluster.updated_items + handler.request.reset_mock() + handler.update_instance() + handler.request.assert_not_called() + + +def test_ambiguous_source_clusters_are_rejected(digitalocean, caplog): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + source.apply() + cluster = inventory.get_all_items(NBCluster)[0] + inventory.add_object(NBCluster, data=dict(cluster.data), read_from_netbox=True) + + with pytest.raises(ValueError, match="Multiple DigitalOcean clusters"): + source.apply() + + assert "Multiple DigitalOcean clusters" in caplog.text + assert len(inventory.get_all_items(NBCluster)) == 2 + + +def test_digitalocean_discovery_maps_droplets_volumes_and_database_clusters(digitalocean): + inventory, client = digitalocean + inventory.add_object( + NBRegion, + data={"name": "New York 1", "slug": "nyc1"}, + read_from_netbox=True, + ) + source = DigitalOceanHandler("cloud") + assert source.init_successful + inventory.resolve_relations() + source.apply() + + client.close.assert_called_once() + assert [call.args for call in client.get_collection.call_args_list] == [ + ("regions", "regions"), + ("droplets", "droplets"), + ("volumes", "volumes"), + ("databases", "databases"), + ] + assert len(inventory.get_all_items(NBCluster)) == 1 + assert len(inventory.get_all_items(NBClusterGroup)) == 1 + cluster = inventory.get_all_items(NBCluster)[0] + assert cluster.data["scope_type"] == NBRegion.object_type + assert cluster.data["scope_id"].data["slug"] == "nyc1" + assert len(inventory.get_all_items(NBVM)) == 2 + assert len(inventory.get_all_items(NBVMInterface)) == 2 + assert len(inventory.get_all_items(NBIPAddress)) == 3 + assert len(inventory.get_all_items(NBVirtualDisk)) == 3 + + droplet = next(vm for vm in inventory.get_all_items(NBVM) if vm.data["name"] == "web-1") + assert droplet.data["status"] == "active" + assert droplet.data["serial"] == "123" + assert droplet.data["vcpus"] == 2 + assert droplet.data["memory"] == 2048 + assert droplet.data["platform"].data["name"] == "Ubuntu 24.04" + assert droplet.data["primary_ip4"].data["address"] == "203.0.113.10/24" + assert droplet.data["primary_ip6"].data["address"] == "2001:db8::5/64" + assert {tag.data["name"] for tag in droplet.data["tags"]} == {"production", "web"} + assert { + disk.data["name"]: disk.data["size"] + for disk in inventory.get_all_items(NBVirtualDisk) + if disk.data["virtual_machine"] is droplet + } == {"root": 25600, "do-volume-1": 20480} + + database = next(vm for vm in inventory.get_all_items(NBVM) if vm.data["name"] == "db-orders") + assert database.data["status"] == "active" + assert database.data["serial"] == "database-uuid" + assert database.data["platform"].data["name"] == "pg 16" + assert database.data["custom_fields"]["digitalocean_resource_type"] == "database" + assert database.data["custom_fields"]["digitalocean_engine"] == "pg" + assert database.data["custom_fields"]["digitalocean_version"] == "16" + assert database.data["custom_fields"]["digitalocean_node_count"] == "2" + assert database.data["custom_fields"]["digitalocean_endpoint"] == "db-private.example.org" + assert database.data["custom_fields"]["digitalocean_port"] == "25060" + assert "secret" not in repr(database.data) + assert len(inventory.get_all_items(NBCustomField)) == 8 + assert {tag.data["name"] for tag in inventory.get_all_items(NBTag)} == { + "production", "web", + } + + source.apply() + assert len(inventory.get_all_items(NBCluster)) == 1 + assert len(inventory.get_all_items(NBVM)) == 2 + assert len(inventory.get_all_items(NBVirtualDisk)) == 3 + assert len(inventory.get_all_items(NBPlatform)) == 2 + assert len(inventory.get_all_items(NBClusterGroup)) == 1 + assert len(inventory.get_all_items(NBTag)) == 2 + + +def test_same_named_droplets_in_different_clusters_keep_their_children(digitalocean): + inventory, client = digitalocean + droplets = [ + { + "id": index, + "name": "prod-web-FE-2", + "status": status, + "region": {"slug": slug, "name": region_name}, + "disk": 25, + "networks": {"v4": [ + {"ip_address": address, "cidr": 24, "type": "public"}, + ]}, + } + for index, slug, region_name, address, status in [ + (1, "nyc3", "New York 3", "203.0.113.1", "off"), + (2, "nyc1", "New York 1", "203.0.113.2", "active"), + ] + ] + client.get_collection.side_effect = lambda path, key: droplets if path == "droplets" else [] + source = DigitalOceanHandler("cloud") + + for _ in range(2): + source.apply() + vms = inventory.get_all_items(NBVM) + assert len(vms) == 2 + assert len(inventory.get_all_items(NBVMInterface)) == 2 + assert len(inventory.get_all_items(NBVirtualDisk)) == 2 + for vm in vms: + ip = vm.data["primary_ip4"] + interface = ip.data["assigned_object_id"] + assert interface.data["virtual_machine"] is vm + assert len(vm.get_virtual_disks()) == 1 + assert vm.get_virtual_disks()[0].data["virtual_machine"] is vm + + +@pytest.mark.parametrize("object_type", [NBVMInterface, NBVirtualDisk]) +def test_vm_children_match_resolved_parent_not_duplicate_name(inventory, object_type): + clusters = [ + inventory.add_object(NBCluster, data={"name": name}, read_from_netbox=True) + for name in ("New York 3", "New York 1") + ] + vms = [ + inventory.add_object(NBVM, data={"name": "same-name", "cluster": cluster}, read_from_netbox=True) + for cluster in clusters + ] + children = [ + inventory.add_object( + object_type, data={"name": "public", "virtual_machine": vm}, read_from_netbox=True, + ) + for vm in vms + ] + + for vm, child in zip(vms, children): + assert inventory.add_update_object( + object_type, data={"name": "PUBLIC", "virtual_machine": vm}, + ) is child + assert child.data["virtual_machine"] is vm + assert "virtual_machine" not in child.updated_items + + +@pytest.mark.parametrize("ownership", ["same-tag", "same-run", "other-tag", "unknown", "mixed-tags", "other-source"]) +@pytest.mark.parametrize("vrf_name", [None, "Team private"]) +def test_ip_conflict_repairs_only_confirmed_same_source(digitalocean, ownership, vrf_name, caplog): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + vrf = inventory.add_object(NBVRF, data={"name": vrf_name}) if vrf_name else None + vms = [ + inventory.add_object(NBVM, data={"name": name}, read_from_netbox=True) + for name in ("old-vm", "new-vm") + ] + interfaces = [ + inventory.add_object( + NBVMInterface, data={"name": "public", "virtual_machine": vm}, read_from_netbox=True, + ) + for vm in vms + ] + ip = inventory.add_object(NBIPAddress, data={ + "id": 19223, "address": "203.0.113.1/24", + "vrf": vrf, + "assigned_object_type": NBVMInterface.object_type, "assigned_object_id": interfaces[0], + }, read_from_netbox=True) + if ownership in ("same-tag", "mixed-tags", "other-source"): + ip.add_tags(source.source_tag) + if ownership in ("other-tag", "mixed-tags"): + ip.add_tags("Source: another-account") + if ownership == "same-run": + ip.source = source + if ownership == "other-source": + ip.source = Mock() + vms[0].data["primary_ip4"] = ip + + if ownership in ("same-tag", "same-run"): + assert source._sync_ip(ip_interface("203.0.113.1/24"), interfaces[1], vrf) is ip + assert ip.data["assigned_object_id"] is interfaces[1] + assert ip.source is source + assert "primary_ip4" in vms[0].unset_items + else: + assert source._sync_ip(ip_interface("203.0.113.1/24"), interfaces[1], vrf) is None + assert "ownership" in caplog.text + assert "NetBox IP ID 19223" in caplog.text + assert "public (old-vm)" in caplog.text + assert "IP source tags=" in caplog.text + assert f"VRF={vrf_name!r}" in caplog.text + assert ip.data["assigned_object_id"] is interfaces[0] + assert vms[0].data["primary_ip4"] is ip + assert "primary_ip4" not in vms[0].unset_items + if ownership == "mixed-tags": + assert ip.source is source + assert "Source: another-account" in ip.get_tags() + inventory.source_list = [source] + inventory.tag_all_the_things(Mock( + primary_tag="NetBox-synced", orphaned_tag="NetBox-synced: Orphaned", + )) + assert "NetBox-synced: Orphaned" not in ip.get_tags() + elif ownership == "other-source": + assert ip.source is not source + else: + assert ip.source is None + + +def test_repaired_ip_is_not_orphaned(digitalocean): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + source.apply() + ip = inventory.get_by_data(NBIPAddress, data={"address": "203.0.113.10/24"}) + target_interface = ip.data["assigned_object_id"] + wrong_vm = inventory.add_object(NBVM, data={"name": "wrong-vm"}, read_from_netbox=True) + wrong_interface = inventory.add_object( + NBVMInterface, data={"name": "public", "virtual_machine": wrong_vm}, read_from_netbox=True, + ) + ip.data["assigned_object_id"] = wrong_interface + ip.source = None + ip.add_tags(["NetBox-synced", source.source_tag, "NetBox-synced: Orphaned"]) + inventory.source_list = [source] + + source.apply() + handler = Mock( + primary_tag="NetBox-synced", orphaned_tag="NetBox-synced: Orphaned", + ) + inventory.tag_all_the_things(handler) + + assert ip.data["assigned_object_id"] is target_interface + assert ip.source is source + assert source.source_tag in ip.get_tags() + assert "NetBox-synced: Orphaned" not in ip.get_tags() + assert target_interface.data["virtual_machine"].data["primary_ip4"] is ip + + +@pytest.mark.parametrize("second_prefix", [24, 32]) +def test_conflicting_droplet_ip_claims_skip_and_continue(digitalocean, second_prefix, caplog): + inventory, client = digitalocean + droplets = [ + {"id": index, "name": name, "region": "nyc1", "networks": { + "v4": [{"ip_address": "203.0.113.1", "cidr": prefix, "type": "public"}], + }} + for index, name, prefix in [(1, "first", 24), (2, "second", second_prefix), (3, "third", 24)] + ] + droplets[2]["networks"]["v4"][0]["ip_address"] = "203.0.113.2" + client.get_collection.side_effect = lambda path, key: droplets if path == "droplets" else [] + source = DigitalOceanHandler("cloud") + + source.apply() + assert "multiple interfaces" in caplog.text + + ip = inventory.get_all_items(NBIPAddress)[0] + assert ip.data["assigned_object_id"].data["virtual_machine"].data["name"] == "first" + vms = {vm.data["name"]: vm for vm in inventory.get_all_items(NBVM)} + assert "primary_ip4" not in vms["second"].data + assert vms["third"].data["primary_ip4"].data["address"] == "203.0.113.2/24" + + +def test_ownership_conflict_continues_other_addresses_and_disks(digitalocean, caplog): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + old_vm = inventory.add_object(NBVM, data={"name": "old"}, read_from_netbox=True) + interface = inventory.add_object(NBVMInterface, data={ + "name": "private", "virtual_machine": old_vm, + }, read_from_netbox=True) + existing = inventory.add_object(NBIPAddress, data={ + "id": 1234, "address": "10.10.0.5/24", + "assigned_object_type": NBVMInterface.object_type, "assigned_object_id": interface, + }, read_from_netbox=True) + source.apply() + assert "Skipping" in caplog.text + assert existing.data["assigned_object_id"] is interface + droplet = next(vm for vm in inventory.get_all_items(NBVM) if vm.data["name"] == "web-1") + assert droplet.data["primary_ip4"].data["address"] == "203.0.113.10/24" + assert droplet.data["primary_ip6"].data["address"] == "2001:db8::5/64" + assert len(droplet.get_virtual_disks()) == 2 + assert len(inventory.get_all_items(NBVM)) == 3 + + +def test_api_failure_closes_session_before_inventory_changes(digitalocean): + inventory, client = digitalocean + client.get_collection.side_effect = RuntimeError("API request failed") + source = DigitalOceanHandler("cloud") + + with pytest.raises(RuntimeError, match="API request failed"): + source.apply() + + client.close.assert_called_once() + assert not inventory.get_all_items(NBCluster) + assert not inventory.get_all_items(NBVM) + + +def test_database_sync_can_be_disabled(digitalocean): + inventory, client = digitalocean + source = DigitalOceanHandler("cloud") + source.settings.sync_databases = False + source.settings.sync_volumes = False + source.apply() + + assert [call.args for call in client.get_collection.call_args_list] == [ + ("regions", "regions"), ("droplets", "droplets"), + ] + assert all(vm.data["name"] != "db-orders" for vm in inventory.get_all_items(NBVM)) + + +def test_digitalocean_client_paginates_and_uses_bearer_auth(): + response_one = Mock() + response_one.json.return_value = { + "droplets": [{"id": 1}], + "links": {"pages": {"next": "https://api.digitalocean.com/v2/droplets?page=2"}}, + } + response_two = Mock() + response_two.json.return_value = {"droplets": [{"id": 2}], "links": {"pages": {}}} + session = Mock() + session.get.side_effect = [response_one, response_two] + + client = DigitalOceanClient("test-token", session=session) + + assert client.get_collection("droplets", "droplets") == [{"id": 1}, {"id": 2}] + first_call, second_call = session.get.call_args_list + assert first_call.kwargs["headers"]["Authorization"] == "Bearer test-token" + assert first_call.kwargs["params"] == {"per_page": 200} + assert second_call.kwargs["params"] is None + + +@pytest.mark.parametrize("databases", [None, []]) +def test_digitalocean_client_accepts_empty_databases(databases): + response = Mock() + response.json.return_value = {"databases": databases} + session = Mock() + session.get.return_value = response + client = DigitalOceanClient("test-token", session=session) + + assert client.get_collection("databases", "databases") == [] + response.raise_for_status.assert_called_once() + session.get.assert_called_once() + + +@pytest.mark.parametrize("path, payload", [ + ("databases", {}), + ("databases", {"databases": {}}), + ("databases", {"databases": ""}), + ("databases", {"databases": False}), + ("droplets", {"droplets": None}), + ("volumes", {"volumes": None}), +]) +def test_digitalocean_client_rejects_invalid_collections(path, payload): + response = Mock() + response.json.return_value = payload + session = Mock() + session.get.return_value = response + client = DigitalOceanClient("test-token", session=session) + + with pytest.raises(ValueError, match=f"has no '{path}' list"): + client.get_collection(path, path) + + +def test_digitalocean_client_rejects_external_pagination_link(): + response = Mock() + response.json.return_value = { + "droplets": [], + "links": {"pages": {"next": "https://attacker.example/steal"}}, + } + session = Mock() + session.get.return_value = response + client = DigitalOceanClient("test-token", session=session) + + with pytest.raises(ValueError, match="pagination link"): + client.get_collection("droplets", "droplets") + + session.get.assert_called_once() + + +def persist_inventory(inventory): + for items in inventory.base_structure.values(): + for item in items: + item.is_new = False + item.updated_items.clear() + item.unset_items.clear() + item.source = None + + +@pytest.mark.parametrize("name", ["web-1", "W" * 64]) +def test_duplicate_names_keep_resource_identity_and_children(digitalocean, name): + inventory, client = digitalocean + first = deepcopy(client.get_collection("droplets", "droplets")[0]) + second = deepcopy(first) + first["name"] = name + second.update(id=456, name=name.lower(), disk=50, vcpus=4, memory=4096, volume_ids=[]) + second["networks"] = {"v4": [ + {"ip_address": "203.0.113.20", "cidr": 24, "type": "public"}, + ]} + payload = { + "droplets": [first, second], + "volumes": client.get_collection("volumes", "volumes"), + "databases": [], + "regions": client.get_collection("regions", "regions"), + } + client.get_collection.side_effect = lambda path, key: payload[path] + source = DigitalOceanHandler("cloud") + for _ in range(2): + source.apply() + vms = inventory.get_all_items(NBVM) + assert len(vms) == 2 + by_id = {vm.data["custom_fields"]["digitalocean_resource_id"]: vm for vm in vms} + for resource in (first, second): + vm = by_id[str(resource["id"])] + suffix = f" ({resource['id']})" + assert vm.data["name"] == resource["name"][:64 - len(suffix)] + suffix + assert vm.data["serial"] == str(resource["id"]) + assert vm.data["vcpus"] == resource["vcpus"] + assert vm.data["memory"] == resource["memory"] + assert vm.get_virtual_disks()[0].data["size"] == resource["disk"] * 1024 + assert vm.data["primary_ip4"].data["assigned_object_id"].data["virtual_machine"] is vm + assert "primary_ip6" not in by_id["456"].data + assert len(by_id["123"].get_virtual_disks()) == 2 + assert len(by_id["456"].get_virtual_disks()) == 1 + if _ == 1: + assert all(not vm.updated_items for vm in vms) + handler = NetBoxHandler.__new__(NetBoxHandler) + handler.inventory = inventory + handler.request = Mock() + handler.update_instance() + handler.request.assert_not_called() + persist_inventory(inventory) + payload["droplets"].reverse() + + +def test_database_and_droplet_name_collision(digitalocean): + inventory, client = digitalocean + payload = {path: client.get_collection(path, path) for path in ("droplets", "volumes", "databases", "regions")} + payload["droplets"][0]["name"] = "db-orders" + client.get_collection.side_effect = lambda path, key: payload[path] + source = DigitalOceanHandler("cloud") + source.apply() + vms = inventory.get_all_items(NBVM) + assert {vm.data["name"] for vm in vms} == {"db-orders (123)", "db-orders (database-uuid)"} + assert len(vms) == 2 + persist_inventory(inventory) + payload["droplets"][0]["name"] = "renamed-droplet" + source.apply() + assert inventory.get_all_items(NBVM) == vms + assert vms[0].data["name"] == "renamed-droplet" + assert vms[1].data["name"] == "db-orders" + + +def test_database_only_region_stays_stable_when_droplet_appears(digitalocean): + inventory, client = digitalocean + payload = {path: client.get_collection(path, path) for path in ("droplets", "volumes", "databases", "regions")} + droplet = payload["droplets"].pop() + droplet["region"] = {"slug": "sfo3", "name": "San Francisco 3"} + payload["databases"][0]["region"] = "sfo3" + payload["regions"] = [{"slug": "sfo3", "name": "San Francisco 3"}] + client.get_collection.side_effect = lambda path, key: payload[path] + client.get_collection.reset_mock() + source = DigitalOceanHandler("cloud") + source.apply() + cluster, = inventory.get_all_items(NBCluster) + database, = inventory.get_all_items(NBVM) + assert cluster.data["name"] == "cloud: San Francisco 3" + payload["droplets"].append(droplet) + client.get_collection.side_effect = lambda path, key: payload[path] + source.apply() + assert inventory.get_all_items(NBCluster) == [cluster] + assert database in inventory.get_all_items(NBVM) + assert len(inventory.get_all_items(NBVM)) == 2 + assert sum(call.args == ("regions", "regions") for call in client.get_collection.call_args_list) == 2 + + +@pytest.mark.parametrize("api_version, factor", [("3.6.0", 1), ("3.7.0", 1), ("4.0.0", 1), ("4.1.0", 1024)]) +def test_disk_units_and_serial_follow_netbox_version(digitalocean, api_version, factor): + inventory, _ = digitalocean + inventory.netbox_api_version = api_version + source = DigitalOceanHandler("cloud") + source.apply() + droplet, database = inventory.get_all_items(NBVM) + assert ("serial" in droplet.data) == (factor == 1024) + assert ("serial" in database.data) == (factor == 1024) + if api_version == "3.6.0": + assert droplet.data["disk"] == 45 + assert database.data["disk"] == 100 + else: + assert {disk.data["name"]: disk.data["size"] for disk in inventory.get_all_items(NBVirtualDisk)} == { + "root": 25 * factor, "do-volume-1": 20 * factor, "database-storage": 100 * factor, + } + persist_inventory(inventory) + source.apply() + assert not droplet.updated_items + assert not database.updated_items + + +@pytest.mark.parametrize("resource_type", ["droplet", "database"]) +def test_resource_id_matches_before_name_and_cluster(digitalocean, resource_type): + inventory, client = digitalocean + payload = {path: client.get_collection(path, path) for path in ("droplets", "volumes", "databases", "regions")} + payload["regions"].append({"slug": "sfo3", "name": "San Francisco 3"}) + client.get_collection.side_effect = lambda path, key: payload[path] + source = DigitalOceanHandler("cloud") + source.apply() + vm = next(vm for vm in inventory.get_all_items(NBVM) + if vm.data["custom_fields"]["digitalocean_resource_type"] == resource_type) + children = vm.get_virtual_disks() + persist_inventory(inventory) + collection = "droplets" if resource_type == "droplet" else "databases" + payload[collection][0].update(name="renamed", region="sfo3") + source.apply() + assert len(inventory.get_all_items(NBVM)) == 2 + assert vm.data["name"] == ("renamed" if resource_type == "droplet" else "db-renamed") + assert vm.data["cluster"].data["name"] == "cloud: San Francisco 3" + assert vm.get_virtual_disks() == children + + +@pytest.mark.parametrize("resource_type", ["droplet", "database"]) +def test_recreated_resource_reuses_vm_of_deleted_one(digitalocean, resource_type): + inventory, client = digitalocean + payload = {path: client.get_collection(path, path) for path in ("droplets", "volumes", "databases", "regions")} + client.get_collection.side_effect = lambda path, key: payload[path] + source = DigitalOceanHandler("cloud") + source.apply() + vms = inventory.get_all_items(NBVM) + vm = next(vm for vm in vms if vm.data["custom_fields"]["digitalocean_resource_type"] == resource_type) + name = vm.data["name"] + persist_inventory(inventory) + collection = "droplets" if resource_type == "droplet" else "databases" + payload[collection][0]["id"] = 789 if resource_type == "droplet" else "new-database-uuid" + source.apply() + assert inventory.get_all_items(NBVM) == vms + assert vm.data["name"] == name + assert vm.data["custom_fields"]["digitalocean_resource_id"] == str(payload[collection][0]["id"]) + assert vm.data["serial"] == str(payload[collection][0]["id"]) + + +def test_rename_to_name_of_deleted_resource_keeps_names_unique(digitalocean): + inventory, client = digitalocean + payload = {path: client.get_collection(path, path) for path in ("droplets", "volumes", "databases", "regions")} + second = deepcopy(payload["droplets"][0]) + second.update(id=456, name="web-2", volume_ids=[], networks={}) + payload["droplets"].append(second) + client.get_collection.side_effect = lambda path, key: payload[path] + source = DigitalOceanHandler("cloud") + source.apply() + first_vm, second_vm = (next(vm for vm in inventory.get_all_items(NBVM) if vm.data["name"] == name) + for name in ("web-1", "web-2")) + persist_inventory(inventory) + payload["droplets"] = [payload["droplets"][0]] + payload["droplets"][0]["name"] = "web-2" + for _ in range(2): + source.apply() + assert first_vm.data["name"] == "web-2 (123)" + assert second_vm.data["name"] == "web-2" + persist_inventory(inventory) + inventory.base_structure[NBVM.name].remove(second_vm) + source.apply() + assert first_vm.data["name"] == "web-2" + + +def test_vm_without_resource_id_is_adopted_by_name_and_cluster(digitalocean): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + source.apply() + droplet = inventory.get_all_items(NBVM)[0] + droplet.data["custom_fields"] = {} + persist_inventory(inventory) + source.apply() + assert inventory.get_all_items(NBVM)[0] is droplet + assert len(inventory.get_all_items(NBVM)) == 2 + assert droplet.data["custom_fields"]["digitalocean_resource_id"] == "123" + + +def test_duplicate_resource_ids_fail_explicitly(digitalocean, capsys): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + source.apply() + vm = inventory.get_all_items(NBVM)[0] + inventory.add_object(NBVM, data=dict(vm.data), read_from_netbox=True) + with pytest.raises(SystemExit) as exc: + source.apply() + assert exc.value.code == 1 + assert "Multiple DigitalOcean VMs match resource ID" in capsys.readouterr().err + + +@pytest.mark.parametrize("orphan_first", [False, True]) +def test_orphaned_duplicate_resource_id_does_not_block_current_vm(digitalocean, orphan_first, caplog): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + source.apply() + current = inventory.get_all_items(NBVM)[0] + stale = inventory.add_object(NBVM, data={ + **current.data, "id": 9001, "name": "prod-status-server", + "primary_ip4": None, "primary_ip6": None, + }, read_from_netbox=True) + stale.add_tags(NetBoxHandler.orphaned_tag) + persist_inventory(inventory) + stale_data = deepcopy(stale.data["custom_fields"]) + if orphan_first: + inventory.base_structure[NBVM.name].reverse() + source.apply() + assert len(inventory.get_all_items(NBVM)) == 3 + assert current.source is source + assert not current.updated_items + assert stale.source is None + assert not stale.updated_items + assert stale.data["custom_fields"] == stale_data + assert NetBoxHandler.orphaned_tag in stale.get_tags() + assert "9001" in caplog.text + assert "prod-status-server" in caplog.text + assert current.data["primary_ip4"].data["assigned_object_id"].data["virtual_machine"] is current + + +def test_only_orphaned_resource_id_match_can_be_reactivated(digitalocean): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + source.apply() + vm = inventory.get_all_items(NBVM)[0] + vm.add_tags(NetBoxHandler.orphaned_tag) + persist_inventory(inventory) + source.apply() + assert vm.source is source + assert len(inventory.get_all_items(NBVM)) == 2 + + +def test_multiple_orphaned_resource_id_matches_remain_ambiguous(digitalocean, capsys): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + source.apply() + vm = inventory.get_all_items(NBVM)[0] + duplicate = inventory.add_object(NBVM, data={ + **vm.data, "id": 9002, "name": "other-orphan", + }, read_from_netbox=True) + for item in (vm, duplicate): + item.add_tags(NetBoxHandler.orphaned_tag) + persist_inventory(inventory) + with pytest.raises(SystemExit): + source.apply() + assert "9002" in capsys.readouterr().err + assert duplicate.source is None + + +def test_managed_region_keeps_source_tag_on_next_run(digitalocean): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + inventory.source_list = [source] + tags = Mock(primary_tag="NetBox-synced", orphaned_tag="NetBox-synced: Orphaned") + source.apply() + inventory.tag_all_the_things(tags) + region, = inventory.get_all_items(NBRegion) + assert source.source_tag in region.get_tags() + persist_inventory(inventory) + source.apply() + inventory.tag_all_the_things(tags) + assert source.source_tag in region.get_tags() + assert "NetBox-synced: Orphaned" not in region.get_tags() + + +@pytest.mark.parametrize("status, message", [(401, "Unable to authenticate you"), (429, "API rate limit exceeded")]) +def test_api_http_error_exits_cleanly_with_provider_message(digitalocean, status, message, capsys, caplog, + monkeypatch): + inventory, _ = digitalocean + response = requests.Response() + response.status_code = status + response._content = ('{"message": "' + message + '"}').encode() + session = Mock() + session.get.return_value = response + client = DigitalOceanClient("test-token", session=session) + monkeypatch.setattr("module.sources.digitalocean.connection.DigitalOceanClient", Mock(return_value=client)) + source = DigitalOceanHandler("cloud") + with pytest.raises(SystemExit) as exc: + source.apply() + assert exc.value.code == 1 + error = capsys.readouterr().err + assert message in error + assert str(status) in error + assert message in caplog.text + session.close.assert_called_once() + assert not inventory.get_all_items(NBCluster) + assert not inventory.get_all_items(NBVM) + + +def test_api_non_json_http_error_retains_status(): + response = requests.Response() + response.status_code = 502 + response._content = b"Bad gateway" + session = Mock() + session.get.return_value = response + client = DigitalOceanClient("test-token", session=session) + with pytest.raises(requests.HTTPError, match="HTTP 502"): + client.get_collection("droplets", "droplets") + + +@pytest.mark.parametrize("name", [" ", "x" * 101]) +def test_invalid_private_ip_vrf_is_rejected(name, caplog): + config = DigitalOceanConfig() + config.source_name = "cloud" + config.config_content = {"source": {"cloud": {"api_token": "test-token", "private_ip_vrf": name}}} + with pytest.raises(SystemExit): + config.parse(do_log=False) + assert "private_ip_vrf" in caplog.text + + +def test_private_ip_vrf_is_opt_in_and_trimmed(): + config = DigitalOceanConfig() + config.source_name = "cloud" + config.config_content = {"source": {"cloud": {"api_token": "test-token"}}} + assert config.parse(do_log=False).private_ip_vrf is None + config = DigitalOceanConfig() + config.source_name = "cloud" + config.config_content = {"source": {"cloud": {"api_token": "test-token"}}} + config.config_content["source"]["cloud"]["private_ip_vrf"] = " Team private " + assert config.parse(do_log=False).private_ip_vrf == "Team private" + + +def test_overlapping_team_private_ips_are_isolated_without_migrating_global_ip(digitalocean): + inventory, _ = digitalocean + first = DigitalOceanHandler("cloud") + second = DigitalOceanHandler("cloud") + second.name = "other-team" + second.set_source_tag() + first.settings.private_ip_vrf = "DigitalOcean Team-A" + second.settings.private_ip_vrf = "DigitalOcean Team-B" + existing_vrf = inventory.add_object( + NBVRF, data={"name": first.settings.private_ip_vrf}, read_from_netbox=True, + ) + old_vm = inventory.add_object(NBVM, data={"name": "old-global-owner"}, read_from_netbox=True) + old_interface = inventory.add_object( + NBVMInterface, data={"name": "private", "virtual_machine": old_vm}, read_from_netbox=True, + ) + global_ip = inventory.add_object(NBIPAddress, data={ + "address": "10.108.0.6/20", "assigned_object_type": NBVMInterface.object_type, + "assigned_object_id": old_interface, + }, read_from_netbox=True) + vms = [inventory.add_object(NBVM, data={"name": name}) for name in ("team-a-monitoring", "team-b-portal1")] + droplet = {"networks": { + "v4": [{"ip_address": "10.108.0.6", "cidr": 20, "type": "private"}], + "v6": [{"ip_address": "2001:db8::6", "cidr": 64, "type": "private"}], + }} + for _ in range(2): + for source, vm in zip((first, second), vms): + source._sync_droplet_network(vm, droplet) + for key in ("primary_ip4", "primary_ip6"): + ip = vm.data[key] + assert ip.data["vrf"].data["name"] == source.settings.private_ip_vrf + assert ip.data["assigned_object_id"].data["virtual_machine"] is vm + assert ip.source is source + assert len(inventory.get_all_items(NBIPAddress)) == 5 + assert len(inventory.get_all_items(NBVRF)) == 2 + assert first._get_private_ip_vrf() is existing_vrf + assert existing_vrf.source is first + assert global_ip.data["assigned_object_id"] is old_interface + assert global_ip.data.get("vrf") is None + assert not global_ip.updated_items + if _ == 1: + assert all(not vm.updated_items for vm in vms) + persist_inventory(inventory) + + +def test_configured_private_vrf_keeps_public_ips_global(digitalocean): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + source.settings.private_ip_vrf = "Team private" + source.apply() + vm = inventory.get_all_items(NBVM)[0] + assert vm.data["primary_ip4"].data.get("vrf") is None + assert vm.data["primary_ip6"].data.get("vrf") is None + private_ip = next(ip for ip in inventory.get_all_items(NBIPAddress) if ip.data["address"] == "10.10.0.5/24") + assert private_ip.data["vrf"].data["name"] == "Team private" + inventory.source_list = [source] + tags = Mock(primary_tag="NetBox-synced", orphaned_tag="NetBox-synced: Orphaned") + inventory.tag_all_the_things(tags) + vrf = private_ip.data["vrf"] + persist_inventory(inventory) + source.apply() + inventory.tag_all_the_things(tags) + assert source.source_tag in vrf.get_tags() + assert not vm.updated_items + + +def test_ip_prefix_change_matches_same_address_and_vrf(digitalocean): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + vm = inventory.add_object(NBVM, data={"name": "test"}) + interface = inventory.add_object(NBVMInterface, data={"name": "private", "virtual_machine": vm}) + vrf = inventory.add_object(NBVRF, data={"name": "private"}) + ip = source._sync_ip(ip_interface("10.108.0.6/20"), interface, vrf) + assert source._sync_ip(ip_interface("10.108.0.6/24"), interface, vrf) is ip + assert ip.data["address"] == "10.108.0.6/24" + assert len(inventory.get_all_items(NBIPAddress)) == 1 + + +def test_duplicate_ip_in_one_vrf_is_skipped(digitalocean, caplog): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + source._ip_claims = {} + vm = inventory.add_object(NBVM, data={"name": "test"}) + interface = inventory.add_object(NBVMInterface, data={"name": "private", "virtual_machine": vm}) + duplicates = [ + inventory.add_object(NBIPAddress, data={"id": index, "address": "10.108.0.6/20"}, read_from_netbox=True) + for index in (1, 2) + ] + assert source._sync_ip(ip_interface("10.108.0.6/20"), interface) is None + assert "Multiple NetBox IPs match '10.108.0.6'" in caplog.text + assert "private_ip_vrf" in caplog.text + assert len(inventory.get_all_items(NBIPAddress)) == 2 + assert all(ip.data.get("assigned_object_id") is None for ip in duplicates) + + +def test_duplicate_ip_uses_the_one_on_the_interface(digitalocean, caplog): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + source._ip_claims = {} + vm = inventory.add_object(NBVM, data={"name": "test"}) + interface = inventory.add_object(NBVMInterface, data={"name": "public", "virtual_machine": vm}) + inventory.add_object(NBIPAddress, data={"id": 1, "address": "198.51.100.7/32", "role": "anycast"}, + read_from_netbox=True) + own = inventory.add_object(NBIPAddress, data={ + "id": 2, "address": "198.51.100.7/24", "assigned_object_type": "virtualization.vminterface", + "assigned_object_id": interface, + }, read_from_netbox=True) + assert source._sync_ip(ip_interface("198.51.100.7/24"), interface) is own + assert "Multiple NetBox IPs" not in caplog.text + assert len(inventory.get_all_items(NBIPAddress)) == 2 + + +def test_duplicate_global_ip_does_not_stop_the_run(digitalocean, caplog): + inventory, _ = digitalocean + for index in (1, 2): + inventory.add_object(NBIPAddress, data={"id": index, "address": "10.10.0.5/24"}, read_from_netbox=True) + DigitalOceanHandler("cloud").apply() + assert "Multiple NetBox IPs match '10.10.0.5'" in caplog.text + vm = next(vm for vm in inventory.get_all_items(NBVM) if vm.data["name"] == "web-1") + assert vm.data["primary_ip4"].data["address"] == "203.0.113.10/24" + assert len(vm.get_virtual_disks()) == 2 + + +def test_ip_claims_are_scoped_by_vrf_including_global(digitalocean, caplog): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + interfaces = [] + for name in ("global", "private", "conflict"): + vm = inventory.add_object(NBVM, data={"name": name}) + interfaces.append(inventory.add_object(NBVMInterface, data={"name": "private", "virtual_machine": vm})) + vrf = inventory.add_object(NBVRF, data={"name": "Team private"}) + address = ip_interface("10.108.0.6/20") + global_ip = source._sync_ip(address, interfaces[0]) + private_ip = source._sync_ip(address, interfaces[1], vrf) + assert global_ip is not private_ip + assert source._sync_ip(address, interfaces[2], vrf) is None + assert "multiple interfaces" in caplog.text + assert global_ip.data["assigned_object_id"] is interfaces[0] + assert private_ip.data["assigned_object_id"] is interfaces[1] + + +def test_duplicate_private_vrf_names_exit_cleanly(digitalocean, capsys): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + source.settings.private_ip_vrf = "Team private" + for _ in range(2): + inventory.add_object(NBVRF, data={"name": "Team private"}, read_from_netbox=True) + with pytest.raises(SystemExit): + source._get_private_ip_vrf() + assert "Multiple NetBox VRFs" in capsys.readouterr().err + + +def test_private_vrf_is_written_before_ip_with_integer_vrf_id(digitalocean): + inventory, _ = digitalocean + source = DigitalOceanHandler("cloud") + source.settings.private_ip_vrf = "Team private" + vm = inventory.add_object(NBVM, data={"id": 501, "name": "test"}, read_from_netbox=True) + interface = inventory.add_object(NBVMInterface, data={ + "id": 601, "name": "private", "virtual_machine": vm, + }, read_from_netbox=True) + vrf = source._get_private_ip_vrf() + ip = source._sync_ip(ip_interface("10.108.0.6/20"), interface, vrf) + handler = NetBoxHandler.__new__(NetBoxHandler) + handler.inventory = inventory + handler.request = Mock(side_effect=lambda object_type, req_type, data, nb_id: { + **data, "id": 701 if object_type is NBVRF else 801, + }) + handler.update_instance() + calls = handler.request.call_args_list + assert [call.args[0] for call in calls] == [NBVRF, NBIPAddress] + assert calls[1].kwargs["data"]["vrf"] == 701 + assert calls[1].kwargs["data"]["assigned_object_id"] == 601 + assert ip.data["vrf"] is vrf + handler.request.reset_mock() + handler.update_instance() + handler.request.assert_not_called() + + +@pytest.mark.parametrize("vrf_name", [None, "Team private"]) +def test_private_vrf_configuration_is_logged_without_null_vrf_noise(digitalocean, vrf_name, caplog): + inventory, _ = digitalocean + caplog.set_level("INFO", logger="NetBox-Sync") + source = DigitalOceanHandler("cloud") + source.settings.private_ip_vrf = vrf_name + source.apply() + if vrf_name is None: + assert "private_ip_vrf is unset" in caplog.text + else: + assert "private addresses use VRF 'Team private'" in caplog.text + caplog.clear() + persist_inventory(inventory) + source.apply() + assert "Found unset key 'vrf'" not in caplog.text