Calculate IP space in LAN Switch Interconnect, add IPAM steps to workflows

gso/cli/imports.py

@@ -42,13 +42,11 @@ from gso.utils.shared_enums import SBPType, Vendor
 from gso.utils.types.base_site import BaseSiteValidatorModel
 from gso.utils.types.interfaces import BandwidthString, LAGMember, LAGMemberList, PhysicalPortCapacity
 from gso.utils.types.ip_address import (
-    AddressSpace,
     IPAddress,
     IPv4AddressType,
-    IPV4Netmask,
-    IPv4NetworkType,
+    IPv4Netmask,
     IPv6AddressType,
-    IPV6Netmask,
+    IPv6Netmask,
     PortNumber,
 )
 from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID
@@ -275,9 +273,9 @@ class L3CoreServiceImportModel(BaseModel):
         vlan_id: VLAN_ID
         custom_firewall_filters: bool = False
         ipv4_address: IPv4AddressType
-        ipv4_mask: IPV4Netmask
+        ipv4_mask: IPv4Netmask
         ipv6_address: IPv6AddressType
-        ipv6_mask: IPV6Netmask
+        ipv6_mask: IPv6Netmask
         is_multi_hop: bool = True
         bgp_peers: list["L3CoreServiceImportModel.BaseBGPPeer"]
         v4_bfd_settings: "L3CoreServiceImportModel.BFDSettingsModel"
@@ -315,7 +313,6 @@ class LanSwitchInterconnectRouterSideImportModel(BaseModel):
     node: UUIDstr
     ae_iface: str
     ae_members: LAGMemberList[LAGMember]
-    ipv4_address: IPv4AddressType
 
 
 class LanSwitchInterconnectSwitchSideImportModel(BaseModel):
@@ -324,15 +321,12 @@ class LanSwitchInterconnectSwitchSideImportModel(BaseModel):
     switch: UUIDstr
     ae_iface: str
     ae_members: LAGMemberList[LAGMember]
-    ipv4_address: IPv4AddressType
 
 
 class LanSwitchInterconnectImportModel(BaseModel):
     """Import LAN Switch Interconnect model."""
 
     lan_switch_interconnect_description: str
-    lan_switch_interconnect_ip_network: IPv4NetworkType | None
-    address_space: AddressSpace
     minimum_links: int
     router_side: LanSwitchInterconnectRouterSideImportModel
     switch_side: LanSwitchInterconnectSwitchSideImportModel

gso/migrations/versions/2024-11-04_e854e0c35e20_update_lan_switch_interconnect.py

+"""Update LAN Switch Interconnect.
+
+Revision ID: e854e0c35e20
+Revises: 0e7e7d749617
+Create Date: 2024-11-04 17:21:14.612740
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = 'e854e0c35e20'
+down_revision = '0e7e7d749617'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("""
+DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectSwitchSideBlock', 'LanSwitchInterconnectRouterSideBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv4_address'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectSwitchSideBlock', 'LanSwitchInterconnectRouterSideBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('ipv4_address'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('address_space'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('address_space'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('lan_switch_interconnect_ip_network'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('LanSwitchInterconnectBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('lan_switch_interconnect_ip_network'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instance_values WHERE subscription_instance_values.resource_type_id IN (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('address_space', 'lan_switch_interconnect_ip_network'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM resource_types WHERE resource_types.resource_type IN ('address_space', 'lan_switch_interconnect_ip_network')
+    """))
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+

gso/oss-params-example.json

@@ -51,6 +51,13 @@
       "domain_name": ".geantip",
       "dns_view": "default",
       "network_view": "default"
+    },
+    "LAN_SWITCH_INTERCONNECT": {
+      "V4": {"containers":  ["10.2.0.0/16"], "networks":  [], "mask": 24},
+      "V6": {"containers":  [], "networks": [], "mask": 126},
+      "domain_name": ".geant.net",
+      "dns_view": "default",
+      "network_view": "default"
     }
   },
   "MONITORING": {

gso/products/product_blocks/lan_switch_interconnect.py

@@ -6,7 +6,6 @@ from orchestrator.types import SubscriptionLifecycle
 from gso.products.product_blocks.router import RouterBlock, RouterBlockInactive, RouterBlockProvisioning
 from gso.products.product_blocks.switch import SwitchBlock, SwitchBlockInactive, SwitchBlockProvisioning
 from gso.utils.types.interfaces import LAGMemberList
-from gso.utils.types.ip_address import AddressSpace, IPv4AddressType, IPv4NetworkType
 
 
 class LanSwitchInterconnectInterfaceBlockInactive(
@@ -48,7 +47,6 @@ class LanSwitchInterconnectRouterSideBlockInactive(
     node: RouterBlockInactive
     ae_iface: str | None = None
     ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockInactive]
-    ipv4_address: IPv4AddressType | None = None
 
 
 class LanSwitchInterconnectRouterSideBlockProvisioning(
@@ -59,7 +57,6 @@ class LanSwitchInterconnectRouterSideBlockProvisioning(
     node: RouterBlockProvisioning
     ae_iface: str | None = None
     ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning]  # type: ignore[assignment]
-    ipv4_address: IPv4AddressType | None
 
 
 class LanSwitchInterconnectRouterSideBlock(
@@ -70,7 +67,6 @@ class LanSwitchInterconnectRouterSideBlock(
     node: RouterBlock
     ae_iface: str
     ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock]  # type: ignore[assignment]
-    ipv4_address: IPv4AddressType | None
 
 
 class LanSwitchInterconnectSwitchSideBlockInactive(
@@ -83,7 +79,6 @@ class LanSwitchInterconnectSwitchSideBlockInactive(
     switch: SwitchBlockInactive
     ae_iface: str | None = None
     ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockInactive]
-    ipv4_address: IPv4AddressType | None = None
 
 
 class LanSwitchInterconnectSwitchSideBlockProvisioning(
@@ -94,7 +89,6 @@ class LanSwitchInterconnectSwitchSideBlockProvisioning(
     switch: SwitchBlockProvisioning
     ae_iface: str | None = None
     ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlockProvisioning]  # type: ignore[assignment]
-    ipv4_address: IPv4AddressType | None
 
 
 class LanSwitchInterconnectSwitchSideBlock(
@@ -105,7 +99,6 @@ class LanSwitchInterconnectSwitchSideBlock(
     switch: SwitchBlock
     ae_iface: str
     ae_members: LAGMemberList[LanSwitchInterconnectInterfaceBlock]  # type: ignore[assignment]
-    ipv4_address: IPv4AddressType | None
 
 
 class LanSwitchInterconnectBlockInactive(
@@ -116,8 +109,6 @@ class LanSwitchInterconnectBlockInactive(
     """A :term:`LAN` Switch Interconnect that's currently inactive, see :class:`LanSwitchInterconnectBlock`."""
 
     lan_switch_interconnect_description: str | None = None
-    lan_switch_interconnect_ip_network: IPv4NetworkType | None = None
-    address_space: AddressSpace | None = None
     minimum_links: int | None = None
     router_side: LanSwitchInterconnectRouterSideBlockInactive
     switch_side: LanSwitchInterconnectSwitchSideBlockInactive
@@ -129,8 +120,6 @@ class LanSwitchInterconnectBlockProvisioning(
     """A :term:`LAN` Switch Interconnect that's currently being provisioned, see :class:`LanSwitchInterconnectBlock`."""
 
     lan_switch_interconnect_description: str | None = None
-    lan_switch_interconnect_ip_network: IPv4NetworkType | None
-    address_space: AddressSpace | None = None
     minimum_links: int | None = None
     router_side: LanSwitchInterconnectRouterSideBlockProvisioning
     switch_side: LanSwitchInterconnectSwitchSideBlockProvisioning
@@ -141,10 +130,6 @@ class LanSwitchInterconnectBlock(LanSwitchInterconnectBlockProvisioning, lifecyc
 
     #: A human-readable description of this :term:`LAN` Switch Interconnect.
     lan_switch_interconnect_description: str
-    #: The :term:`IP` resources for this :term:`VLAN` Switch Interconnect.
-    lan_switch_interconnect_ip_network: IPv4NetworkType | None
-    #: The address space of the :term:`VLAN` Switch Interconnect. It can be private or public.
-    address_space: AddressSpace
     #: The minimum amount of links the :term:`LAN` Switch Interconnect should consist of.
     minimum_links: int
     #: The router side of the :term:`LAN` Switch Interconnect.

gso/products/product_blocks/service_binding_port.py

@@ -14,7 +14,7 @@ from gso.products.product_blocks.bgp_session import (
 )
 from gso.products.product_blocks.edge_port import EdgePortBlock, EdgePortBlockInactive, EdgePortBlockProvisioning
 from gso.utils.shared_enums import SBPType
-from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
+from gso.utils.types.ip_address import IPv4AddressType, IPv4Netmask, IPv6AddressType, IPv6Netmask
 from gso.utils.types.virtual_identifiers import VLAN_ID
 
 
@@ -60,9 +60,9 @@ class ServiceBindingPortInactive(
     vlan_id: VLAN_ID | None = None
     sbp_type: SBPType | None = None
     ipv4_address: IPv4AddressType | None = None
-    ipv4_mask: IPV4Netmask | None = None
+    ipv4_mask: IPv4Netmask | None = None
     ipv6_address: IPv6AddressType | None = None
-    ipv6_mask: IPV6Netmask | None = None
+    ipv6_mask: IPv6Netmask | None = None
     custom_firewall_filters: bool | None = None
     geant_sid: str | None = None
     bgp_session_list: list[BGPSessionInactive] = Field(default_factory=list)
@@ -78,9 +78,9 @@ class ServiceBindingPortProvisioning(ServiceBindingPortInactive, lifecycle=[Subs
     vlan_id: VLAN_ID | None = None
     sbp_type: SBPType
     ipv4_address: IPv4AddressType | None = None
-    ipv4_mask: IPV4Netmask | None = None
+    ipv4_mask: IPv4Netmask | None = None
     ipv6_address: IPv6AddressType | None = None
-    ipv6_mask: IPV6Netmask | None = None
+    ipv6_mask: IPv6Netmask | None = None
     custom_firewall_filters: bool
     geant_sid: str
     bgp_session_list: list[BGPSessionProvisioning]  # type: ignore[assignment]
@@ -101,11 +101,11 @@ class ServiceBindingPort(ServiceBindingPortProvisioning, lifecycle=[Subscription
     #: If layer 3, IPv4 resources.
     ipv4_address: IPv4AddressType | None = None
     #: IPV4 subnet mask.
-    ipv4_mask: IPV4Netmask | None = None
+    ipv4_mask: IPv4Netmask | None = None
     #: If layer 3, IPv6 resources.
     ipv6_address: IPv6AddressType | None = None
     #: IPV6 subnet mask.
-    ipv6_mask: IPV6Netmask | None = None
+    ipv6_mask: IPv6Netmask | None = None
     #: Any custom firewall filters that the partner may require.
     custom_firewall_filters: bool
     #: The GÉANT service ID of this binding port.

gso/services/infoblox.py

@@ -10,7 +10,7 @@ from infoblox_client.exceptions import (
 )
 
 from gso.settings import IPAMParams, load_oss_params
-from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
+from gso.utils.types.ip_address import IPv4AddressType, IPv4Netmask, IPv4NetworkType, IPv6AddressType, IPv6Netmask
 
 logger = getLogger(__name__)
 NULL_MAC = "00:00:00:00:00:00"
@@ -41,7 +41,7 @@ def _allocate_network(  # noqa: PLR0917
     conn: connector.Connector,
     dns_view: str,
     network_view: str,
-    netmask: int,
+    netmask: IPv4Netmask | IPv6Netmask,
     containers: list[str],
     comment: str | None = "",
 ) -> ipaddress.IPv4Network | ipaddress.IPv6Network:
@@ -73,6 +73,21 @@ def _allocate_network(  # noqa: PLR0917
     raise AllocationError(msg)
 
 
+def create_v4_network_by_ip(
+    dns_view: str, network_view: str, network: IPv4NetworkType, comment: str | None = ""
+) -> None:
+    """Register an IPv4 network at the given location. Raises an :class:`AllocationError` on failure."""
+    conn, _ = _setup_connection()
+    created_net = objects.NetworkV4.create(
+        conn, network=network, view=dns_view, network_view=network_view, comment=comment
+    )
+    if created_net.response != "Infoblox Object created":
+        msg = f"Failed to allocate network at {network}. Response from Netbox: {created_net.response}"
+        raise AllocationError(msg)
+    msg = f"Successfully registered new network at {network}"
+    logger.debug(msg)
+
+
 def hostname_available(hostname: str) -> bool:
     """Check whether a hostname is still available **in Infoblox**.
 
@@ -234,38 +249,48 @@ def allocate_host(
 
 def create_host_by_ip(
     hostname: str,
-    ipv4_address: IPv4AddressType,
-    ipv6_address: IPv6AddressType,
     service_type: str,
     comment: str,
+    ipv4_address: IPv4AddressType | None = None,
+    ipv6_address: IPv6AddressType | None = None,
 ) -> None:
     """Create a new host record with a given IPv4 and IPv6 address.
 
     :param str hostname: The :term:`FQDN` of the new host.
-    :param IPv4Address ipv4_address: The IPv4 address of the new host.
-    :param IPv6Address ipv6_address: The IPv6 address of the new host.
     :param str service_type: The relevant service type, used to deduce the correct ``dns_view`` and ``network_view`` in
                              Infoblox.
     :param str comment: The comment stored in this Infoblox record, most likely the relevant ``subscription_id`` in
                         :term:`GSO`.
+    :param IPv4Address ipv4_address: The IPv4 address of the new host.
+    :param IPv6Address ipv6_address: The IPv6 address of the new host.
     """
     if not hostname_available(hostname):
         msg = f"FQDN '{hostname}' is already in use, allocation aborted."
         raise AllocationError(msg)
+    if not (ipv4_address or ipv6_address):
+        msg = "At least one IP address must be given to create a record for. Received None."
+        raise AllocationError(msg)
 
     conn, oss = _setup_connection()
-    ipv6_object = objects.IP.create(ip=str(ipv6_address), mac=NULL_MAC, configure_for_dhcp=False)
-    ipv4_object = objects.IP.create(ip=str(ipv4_address), mac=NULL_MAC, configure_for_dhcp=False)
     dns_view = getattr(oss, service_type).dns_view
     network_view = getattr(oss, service_type).network_view
 
     # This needs to be done in two steps, otherwise only one of the IP addresses is stored.
+    if ipv6_address:  # We first register the IPv6 record.
+        ipv6_object = objects.IP.create(ip=str(ipv6_address), mac=NULL_MAC, configure_for_dhcp=False)
         objects.HostRecord.create(
             conn, ip=ipv6_object, name=hostname, comment=comment, view=dns_view, network_view=network_view
         )
+        if ipv4_address:  # If set, we also register the IPv4 record.
             new_host = find_host_by_fqdn(hostname)
+            ipv4_object = objects.IP.create(ip=str(ipv4_address), mac=NULL_MAC, configure_for_dhcp=False)
             new_host.ipv4addrs = [ipv4_object]
             new_host.update()
+    else:  # IPv6 is not set, it must therefore be IPv4 only.
+        ipv4_object = objects.IP.create(ip=str(ipv4_address), mac=NULL_MAC, configure_for_dhcp=False)
+        objects.HostRecord.create(
+            conn, ip=ipv4_object, name=hostname, comment=comment, view=dns_view, network_view=network_view
+        )
 
 
 def find_host_by_ip(ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> objects.HostRecord | None:
@@ -295,11 +320,7 @@ def find_host_by_fqdn(fqdn: str) -> objects.HostRecord:
     :type fqdn: str
     """
     conn, _ = _setup_connection()
-    return objects.HostRecord.search(
-        conn,
-        name=fqdn,
-        return_fields=["ipv4addrs", "name", "view", "aliases", "comment"],
-    )
+    return objects.HostRecord.search(conn, name=fqdn, return_fields=["ipv4addrs", "name", "view", "aliases", "comment"])
 
 
 def find_v6_host_by_fqdn(fqdn: str) -> objects.HostRecordV6:

gso/settings.py

@@ -15,7 +15,7 @@ from pydantic import EmailStr
 from pydantic_forms.types import strEnum
 from pydantic_settings import BaseSettings
 
-from gso.utils.types.ip_address import IPV4Netmask, IPV6Netmask, PortNumber
+from gso.utils.types.ip_address import IPv4Netmask, IPv6Netmask, PortNumber
 
 logger = logging.getLogger(__name__)
 
@@ -65,7 +65,7 @@ class V4NetworkParams(BaseSettings):
 
     containers: list[ipaddress.IPv4Network]
     networks: list[ipaddress.IPv4Network]
-    mask: IPV4Netmask
+    mask: IPv4Netmask
 
 
 class V6NetworkParams(BaseSettings):
@@ -73,7 +73,7 @@ class V6NetworkParams(BaseSettings):
 
     containers: list[ipaddress.IPv6Network]
     networks: list[ipaddress.IPv6Network]
-    mask: IPV6Netmask
+    mask: IPv6Netmask
 
 
 class ServiceNetworkParams(BaseSettings):
@@ -98,6 +98,7 @@ class IPAMParams(BaseSettings):
     GEANT_IP: ServiceNetworkParams
     SI: ServiceNetworkParams
     LT_IAS: ServiceNetworkParams
+    LAN_SWITCH_INTERCONNECT: ServiceNetworkParams
 
 
 class MonitoringSNMPV2Params(BaseSettings):

gso/utils/helpers.py

@@ -2,6 +2,7 @@
 
 import random
 import re
+from ipaddress import IPv4Network
 from typing import TYPE_CHECKING
 from uuid import UUID
 
@@ -18,7 +19,7 @@ from gso.services.partners import get_all_partners
 from gso.services.subscriptions import is_virtual_circuit_id_available
 from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import PhysicalPortCapacity
-from gso.utils.types.ip_address import IPv4AddressType
+from gso.utils.types.ip_address import IPv4AddressType, IPv4NetworkType
 from gso.utils.types.virtual_identifiers import VC_ID
 
 if TYPE_CHECKING:
@@ -123,6 +124,17 @@ def generate_fqdn(hostname: str, site_name: str, country_code: str) -> str:
     return f"{hostname}.{site_name.lower()}.{country_code.lower()}{oss.IPAM.LO.domain_name}"
 
 
+def generate_lan_switch_interconnect_subnet(site_internal_id: int) -> IPv4NetworkType:
+    """Generate an IPv4 network in which a :term:`LAN` Switch Interconnect resides, given a Site internal ID."""
+    ipam_oss = settings.load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT
+
+    result = str(ipam_oss.V4.containers[0]).split(".")[:2]  # Take the first two octets from the IPv4 network.
+    result.append(str(site_internal_id))  # Append the side ID as the third octet.
+    result.append(f"0/{ipam_oss.V4.mask}")  # Append the fourth octet, together with the netmask.
+
+    return IPv4Network(".".join(result))
+
+
 def generate_inventory_for_routers(
     router_role: RouterRole,
     exclude_routers: list[str] | None = None,

gso/utils/types/ip_address.py

@@ -4,7 +4,6 @@ import ipaddress
 from typing import Annotated, Any
 
 from pydantic import AfterValidator, Field, PlainSerializer
-from pydantic_forms.types import strEnum
 from typing_extensions import Doc
 
 
@@ -40,8 +39,8 @@ IPv6AddressType = Annotated[ipaddress.IPv6Address, PlainSerializer(_str, return_
 IPv6NetworkType = Annotated[ipaddress.IPv6Network, PlainSerializer(_str, return_type=str, when_used="always")]
 IPAddress = Annotated[str, AfterValidator(validate_ipv4_or_ipv6)]
 IPNetwork = Annotated[str, AfterValidator(validate_ipv4_or_ipv6_network)]
-IPV4Netmask = Annotated[int, Field(ge=0, le=32), Doc("A valid netmask for an IPv4 network or address.")]
-IPV6Netmask = Annotated[int, Field(ge=0, le=128), Doc("A valid netmask for an IPv6 network or address.")]
+IPv4Netmask = Annotated[int, Field(ge=0, le=32), Doc("A valid netmask for an IPv4 network or address.")]
+IPv6Netmask = Annotated[int, Field(ge=0, le=128), Doc("A valid netmask for an IPv6 network or address.")]
 PortNumber = Annotated[
     int,
     Field(
@@ -53,10 +52,3 @@ PortNumber = Annotated[
         "and can therefore not be selected for permanent allocation."
     ),
 ]
-
-
-class AddressSpace(strEnum):
-    """Types of address space. Can be private or public."""
-
-    PRIVATE = "PRIVATE"
-    PUBLIC = "PUBLIC"

gso/workflows/iptrunk/create_iptrunk.py

@@ -482,7 +482,7 @@ def register_dns_records(subscription: IptrunkInactive) -> State:
         ipv4_addr = subscription.iptrunk.iptrunk_ipv4_network[index]
         ipv6_addr = subscription.iptrunk.iptrunk_ipv6_network[index + 1]
 
-        infoblox.create_host_by_ip(fqdn, ipv4_addr, ipv6_addr, "TRUNK", str(subscription.subscription_id))
+        infoblox.create_host_by_ip(fqdn, "TRUNK", str(subscription.subscription_id), ipv4_addr, ipv6_addr)
 
     return {"subscription": subscription}
 

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -738,7 +738,7 @@ def update_ipam(subscription: Iptrunk, replace_index: int, new_node: Router, new
     #  And in with the new
     new_fqdn = f"{new_lag_interface}-0.{new_node.router.router_fqdn}"
     comment = str(subscription.subscription_id)
-    infoblox.create_host_by_ip(new_fqdn, v4_addr, v6_addr, "TRUNK", comment)
+    infoblox.create_host_by_ip(new_fqdn, "TRUNK", comment, v4_addr, v6_addr)
 
     return {"subscription": subscription}
 

gso/workflows/l3_core_service/create_imported_l3_core_service.py

@@ -21,7 +21,7 @@ from gso.products.product_types.l3_core_service import ImportedL3CoreServiceInac
 from gso.services.partners import get_partner_by_name
 from gso.services.subscriptions import get_product_id_by_name
 from gso.utils.shared_enums import SBPType
-from gso.utils.types.ip_address import IPAddress, IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
+from gso.utils.types.ip_address import IPAddress, IPv4AddressType, IPv4Netmask, IPv6AddressType, IPv6Netmask
 from gso.utils.types.virtual_identifiers import VLAN_ID
 
 
@@ -56,9 +56,9 @@ def initial_input_form_generator() -> FormGenerator:
         vlan_id: VLAN_ID
         custom_firewall_filters: bool = False
         ipv4_address: IPv4AddressType
-        ipv4_mask: IPV4Netmask
+        ipv4_mask: IPv4Netmask
         ipv6_address: IPv6AddressType
-        ipv6_mask: IPV6Netmask
+        ipv6_mask: IPv6Netmask
         rtbh_enabled: bool = True
         is_multi_hop: bool = True
         bgp_peers: list[BaseBGPPeer]

gso/workflows/l3_core_service/create_l3_core_service.py

@@ -25,7 +25,7 @@ from gso.utils.helpers import (
     partner_choice,
 )
 from gso.utils.shared_enums import APType, SBPType
-from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
+from gso.utils.types.ip_address import IPv4AddressType, IPv4Netmask, IPv6AddressType, IPv6Netmask
 from gso.utils.types.tt_number import TTNumber
 from gso.utils.types.virtual_identifiers import VLAN_ID
 
@@ -119,12 +119,12 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         divider: Divider = Field(None, exclude=True)
         v4_label: Label = Field("IPV4 SBP interface params", exclude=True)
         ipv4_address: IPv4AddressType
-        ipv4_mask: IPV4Netmask
+        ipv4_mask: IPv4Netmask
         v4_bfd_settings: BFDSettingsForm
         divider2: Divider = Field(None, exclude=True)
         v6_label: Label = Field("IPV6 SBP interface params", exclude=True)
         ipv6_address: IPv6AddressType
-        ipv6_mask: IPV6Netmask
+        ipv6_mask: IPv6Netmask
         v6_bfd_settings: BFDSettingsForm
         divider3: Divider = Field(None, exclude=True)
         v4_bgp_peer: IPv4BGPPeer

gso/workflows/l3_core_service/modify_l3_core_service.py

@@ -21,7 +21,7 @@ from gso.products.product_types.edge_port import EdgePort
 from gso.products.product_types.l3_core_service import L3CoreService
 from gso.utils.helpers import active_edge_port_selector
 from gso.utils.shared_enums import APType, SBPType
-from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
+from gso.utils.types.ip_address import IPv4AddressType, IPv4Netmask, IPv6AddressType, IPv6Netmask
 from gso.utils.types.virtual_identifiers import VLAN_ID
 
 
@@ -158,9 +158,9 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
             # occur since it's a layer 3 service. The ignore statements are there to put our type checker at ease.
             vlan_id: VLAN_ID = current_sbp.vlan_id  # type: ignore[assignment]
             ipv4_address: IPv4AddressType = current_sbp.ipv4_address  # type: ignore[assignment]
-            ipv4_mask: IPV4Netmask = current_sbp.ipv4_mask  # type: ignore[assignment]
+            ipv4_mask: IPv4Netmask = current_sbp.ipv4_mask  # type: ignore[assignment]
             ipv6_address: IPv6AddressType = current_sbp.ipv6_address  # type: ignore[assignment]
-            ipv6_mask: IPV6Netmask = current_sbp.ipv6_mask  # type: ignore[assignment]
+            ipv6_mask: IPv6Netmask = current_sbp.ipv6_mask  # type: ignore[assignment]
             custom_firewall_filters: bool = current_sbp.custom_firewall_filters
             v4_bfd_settings: BFDInputModel = BFDInputModel(
                 bfd_enabled=current_sbp.v4_bfd_settings.bfd_enabled,

gso/workflows/lan_switch_interconnect/create_imported_lan_switch_interconnect.py

@@ -21,14 +21,11 @@ from gso.products.product_types.router import Router
 from gso.products.product_types.switch import Switch
 from gso.services.partners import get_partner_by_name
 from gso.services.subscriptions import get_product_id_by_name
-from gso.utils.types.ip_address import AddressSpace, IPv4NetworkType
 
 
 def _initial_input_form_generator() -> FormGenerator:
     class ImportLanSwitchInterconnect(FormPage):
         lan_switch_interconnect_description: str
-        lan_switch_interconnect_ip_network: IPv4NetworkType | None
-        address_space: AddressSpace
         minimum_links: int
         router_side: LanSwitchInterconnectRouterSideImportModel
         switch_side: LanSwitchInterconnectSwitchSideImportModel
@@ -51,16 +48,12 @@ def create_subscription(partner: str) -> State:
 def initialize_subscription(
     subscription: ImportedLanSwitchInterconnectInactive,
     lan_switch_interconnect_description: str,
-    lan_switch_interconnect_ip_network: IPv4NetworkType | None,
-    address_space: AddressSpace,
     minimum_links: int,
     router_side: dict,
     switch_side: dict,
 ) -> State:
     """Initialize the subscription using input data."""
     subscription.lan_switch_interconnect.lan_switch_interconnect_description = lan_switch_interconnect_description
-    subscription.lan_switch_interconnect.lan_switch_interconnect_ip_network = lan_switch_interconnect_ip_network
-    subscription.lan_switch_interconnect.address_space = address_space
     subscription.lan_switch_interconnect.minimum_links = minimum_links
 
     router_block = Router.from_subscription(router_side.pop("node")).router

gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py

 """A creation workflow for creating a new interconnect between a switch and a router."""
 
+from ipaddress import IPv4Address
 from typing import Annotated
 from uuid import uuid4
 
@@ -7,6 +8,7 @@ from annotated_types import Len
 from orchestrator.forms import FormPage
 from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
+from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.workflow import StepList, begin, done, step, workflow
 from orchestrator.workflows.steps import resync, set_status, store_process_subscription
 from orchestrator.workflows.utils import wrap_create_initial_input_form
@@ -19,12 +21,15 @@ from gso.products.product_blocks.lan_switch_interconnect import (
 from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnectInactive
 from gso.products.product_types.router import Router
 from gso.products.product_types.switch import Switch
+from gso.services.infoblox import create_host_by_ip, create_v4_network_by_ip
 from gso.services.partners import get_partner_by_name
+from gso.settings import load_oss_params
 from gso.utils.helpers import (
     active_router_selector,
     active_switch_selector,
     available_interfaces_choices,
     available_lags_choices,
+    generate_lan_switch_interconnect_subnet,
 )
 from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import (
@@ -35,7 +40,6 @@ from gso.utils.types.interfaces import (
     PhysicalPortCapacity,
     validate_interface_names_are_unique,
 )
-from gso.utils.types.ip_address import AddressSpace
 from gso.utils.types.tt_number import TTNumber
 from gso.workflows.shared import create_summary_form
 
@@ -48,7 +52,6 @@ def _initial_input_form(product_name: str) -> FormGenerator:
         partner: ReadOnlyField("GEANT", default_type=str)  # type: ignore[valid-type]
         router_side: active_router_selector()  # type: ignore[valid-type]
         switch_side: active_switch_selector()  # type: ignore[valid-type]
-        address_space: AddressSpace
         description: str
         minimum_link_count: int
         divider: Divider
@@ -107,7 +110,6 @@ def _initial_input_form(product_name: str) -> FormGenerator:
         "partner",
         "router_side",
         "switch_side",
-        "address_space",
         "description",
         "minimum_link_count",
         "vlan_id",
@@ -133,7 +135,6 @@ def create_subscription(product: UUIDstr, partner: str) -> State:
 def initialize_subscription(
     subscription: LanSwitchInterconnectInactive,
     description: str,
-    address_space: AddressSpace,
     minimum_link_count: int,
     router_side: UUIDstr,
     router_side_iface: JuniperPhyInterface,
@@ -144,7 +145,6 @@ def initialize_subscription(
 ) -> State:
     """Update the product model with all input from the operator."""
     subscription.lan_switch_interconnect.lan_switch_interconnect_description = description
-    subscription.lan_switch_interconnect.address_space = address_space
     subscription.lan_switch_interconnect.minimum_links = minimum_link_count
     subscription.lan_switch_interconnect.router_side.node = Router.from_subscription(router_side).router
     subscription.lan_switch_interconnect.router_side.ae_iface = router_side_iface
@@ -162,6 +162,47 @@ def initialize_subscription(
     return {"subscription": subscription}
 
 
+@step("Register network in IPAM")
+def register_dns_records_network(subscription: LanSwitchInterconnectInactive) -> State:
+    """Add :term:`DNS` records in :term:`IPAM`."""
+    router_site = subscription.lan_switch_interconnect.router_side.node.router_site
+    if not router_site or not router_site.site_internal_id:
+        msg = "Site internal ID not set. Cannot continue."
+        raise ProcessFailureError(msg, details=router_site)
+
+    new_network = generate_lan_switch_interconnect_subnet(router_site.site_internal_id)
+    ipam_oss_params = load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT
+    create_v4_network_by_ip(
+        ipam_oss_params.dns_view, ipam_oss_params.network_view, new_network, str(subscription.subscription_id)
+    )
+
+    return {"ipam_registrations": {"network": new_network}}
+
+
+@step("Register devices in IPAM")
+def register_dns_records_devices(
+    subscription: LanSwitchInterconnectInactive, subscription_id: UUIDstr, ipam_registrations: dict[str, str]
+) -> State:
+    """Register :term:`DNS` records for both switch and router side in :term:`IPAM`."""
+    switch_hostname = subscription.lan_switch_interconnect.switch_side.switch.fqdn
+    router_hostname = (
+        f"{subscription.lan_switch_interconnect.router_side.ae_iface}."
+        f"{subscription.lan_switch_interconnect.router_side.node.router_fqdn}"
+    )
+    if not (switch_hostname and router_hostname):
+        msg = "Missing switch or router hostname, cannot continue."
+        raise ProcessFailureError(msg, details=subscription.lan_switch_interconnect)
+
+    ip_network_prefix = ipam_registrations["network"].split(".")[:3]  # Take the first three octets of the network.
+    switch_side_ip = IPv4Address(".".join([*ip_network_prefix, "10"]))  # Add .10 as the fourth octet of the switch.
+    router_side_ip = IPv4Address(".".join([*ip_network_prefix, "1"]))  # Add .1 as the fourth octet of the router.
+
+    create_host_by_ip(switch_hostname, "LAN_SWITCH_INTERCONNECT", subscription_id, switch_side_ip)
+    create_host_by_ip(router_hostname, "LAN_SWITCH_INTERCONNECT", subscription_id, router_side_ip)
+
+    return {"ipam_registrations": {switch_hostname: switch_side_ip, router_hostname: router_side_ip}}
+
+
 @workflow(
     "Create LAN Switch Interconnect",
     initial_input_form=wrap_create_initial_input_form(_initial_input_form),
@@ -174,6 +215,8 @@ def create_lan_switch_interconnect() -> StepList:
         >> create_subscription
         >> store_process_subscription(Target.CREATE)
         >> initialize_subscription
+        >> register_dns_records_network
+        >> register_dns_records_devices
         >> set_status(SubscriptionLifecycle.ACTIVE)
         >> resync
         >> done

gso/workflows/lan_switch_interconnect/terminate_lan_switch_interconnect.py

@@ -4,13 +4,15 @@ from orchestrator import begin, workflow
 from orchestrator.forms import FormPage
 from orchestrator.targets import Target
 from orchestrator.types import SubscriptionLifecycle, UUIDstr
-from orchestrator.workflow import StepList, done
+from orchestrator.workflow import StepList, done, step
 from orchestrator.workflows.steps import resync, set_status, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from pydantic_forms.types import FormGenerator
 from pydantic_forms.validators import Label
 
 from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
+from gso.services.infoblox import delete_host_by_fqdn, delete_network
+from gso.utils.helpers import generate_lan_switch_interconnect_subnet
 from gso.utils.types.tt_number import TTNumber
 
 
@@ -31,6 +33,25 @@ def _input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     return {"subscription": lan_switch_interconnect}
 
 
+@step("Release IPAM resources")
+def clean_up_ipam(subscription: LanSwitchInterconnect) -> None:
+    """Clear up :term:`IPAM` resources in :term:`DNS`.
+
+    First, remove :term:`DNS` records for both the router and switch side of the interconnect.
+    Second, remove the registered network for this interconnect.
+    """
+    delete_host_by_fqdn(subscription.lan_switch_interconnect.switch_side.switch.fqdn)
+    delete_host_by_fqdn(
+        f"{subscription.lan_switch_interconnect.router_side.ae_iface}."
+        f"{subscription.lan_switch_interconnect.router_side.node.router_fqdn}"
+    )
+    delete_network(
+        generate_lan_switch_interconnect_subnet(
+            subscription.lan_switch_interconnect.router_side.node.router_site.site_internal_id
+        )
+    )
+
+
 @workflow(
     "Terminate LAN Switch Interconnect",
     initial_input_form=wrap_modify_initial_input_form(_input_form_generator),
@@ -42,6 +63,7 @@ def terminate_lan_switch_interconnect() -> StepList:
         begin
         >> store_process_subscription(Target.TERMINATE)
         >> unsync
+        >> clean_up_ipam
         >> set_status(SubscriptionLifecycle.TERMINATED)
         >> resync
         >> done

gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py

@@ -3,11 +3,39 @@
 from typing import Any
 
 from orchestrator.targets import Target
+from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.workflow import StepList, begin, done, step, workflow
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 
+from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
+from gso.services.infoblox import find_host_by_fqdn, find_network_by_cidr
 from gso.services.lso_client import LSOState, anonymous_lso_interaction
+from gso.utils.helpers import generate_lan_switch_interconnect_subnet
+
+
+@step("Validate IPAM configuration")
+def validate_ipam(subscription: LanSwitchInterconnect) -> None:
+    """Validate that :term:`DNS` records are registered in :term:`IPAM` correctly."""
+    switch_fqdn = subscription.lan_switch_interconnect.switch_side.switch.fqdn
+    router_fqdn = (
+        f"{subscription.lan_switch_interconnect.router_side.ae_iface}."
+        f"{subscription.lan_switch_interconnect.router_side.node.router_fqdn}"
+    )
+
+    for fqdn in [switch_fqdn, router_fqdn]:
+        host_record = find_host_by_fqdn(fqdn)
+        if not host_record or str(subscription.subscription_id) not in host_record.comment:
+            msg = "DNS record is incorrectly configured in IPAM, please investigate this manually!"
+            raise ProcessFailureError(msg)
+
+    lan_interconnect_network = generate_lan_switch_interconnect_subnet(
+        subscription.lan_switch_interconnect.router_side.node.router_site.site_internal_id
+    )
+    network_record = find_network_by_cidr(lan_interconnect_network)
+    if not network_record or str(subscription.subscription_id) not in network_record.comment:
+        msg = "LAN Switch Interconnect network is incorrectly configured in IPAM, please investigate this manually!"
+        raise ProcessFailureError(msg)
 
 
 @step("Check config for drift")
@@ -15,7 +43,14 @@ def validate_config(subscription: dict[str, Any]) -> LSOState:
     """Workflow step for running a playbook that checks whether config has drifted."""
     return {
         "playbook_name": "lan_switch_interconnect.yaml",
-        "inventory": {"all": {"hosts": {None: None}}},
+        "inventory": {
+            "all": {
+                "hosts": {
+                    subscription["lan_switch_interconnect"]["switch_side"]["switch"]["fqdn"]: None,
+                    subscription["lan_switch_interconnect"]["router_side"]["node"]["router_fqdn"]: None,
+                }
+            }
+        },
         "extra_vars": {
             "subscription_json": subscription,
             "verb": "deploy",
@@ -34,6 +69,7 @@ def validate_lan_switch_interconnect() -> StepList:
         begin
         >> store_process_subscription(Target.SYSTEM)
         >> unsync
+        >> validate_ipam
         >> anonymous_lso_interaction(validate_config)
         >> resync
         >> done

test/cli/test_imports.py

@@ -30,7 +30,6 @@ from gso.products.product_types.site import Site
 from gso.utils.helpers import generate_unique_vc_id, iso_from_ipv4
 from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import PhysicalPortCapacity
-from gso.utils.types.ip_address import AddressSpace
 
 
 ##############
@@ -230,8 +229,6 @@ def lan_switch_interconnect_data(temp_file, faker, switch_subscription_factory,
     def _lan_switch_interconnect_data(**kwargs):
         lan_switch_interconnect_data = {
             "lan_switch_interconnect_description": faker.sentence(),
-            "lan_switch_interconnect_ip_network": str(faker.ipv4_network()),
-            "address_space": AddressSpace.PUBLIC,
             "minimum_links": 1,
             "router_side": {
                 "node": router_subscription_factory(),
@@ -240,7 +237,6 @@ def lan_switch_interconnect_data(temp_file, faker, switch_subscription_factory,
                     {"interface_name": faker.network_interface(), "interface_description": faker.sentence()}
                     for _ in range(2)
                 ],
-                "ipv4_address": faker.ipv4(),
             },
             "switch_side": {
                 "switch": switch_subscription_factory(),
@@ -249,7 +245,6 @@ def lan_switch_interconnect_data(temp_file, faker, switch_subscription_factory,
                     {"interface_name": faker.network_interface(), "interface_description": faker.sentence()}
                     for _ in range(2)
                 ],
-                "ipv4_address": faker.ipv4(),
             },
         }
         lan_switch_interconnect_data.update(**kwargs)

test/fixtures/lan_switch_interconnect_fixtures.py

@@ -18,7 +18,6 @@ from gso.products.product_types.lan_switch_interconnect import (
 from gso.products.product_types.router import Router
 from gso.products.product_types.switch import Switch
 from gso.services.subscriptions import get_product_id_by_name
-from gso.utils.types.ip_address import AddressSpace, IPv4AddressType, IPv4NetworkType
 
 
 @pytest.fixture()
@@ -31,17 +30,13 @@ def lan_switch_interconnect_subscription_factory(
         status: SubscriptionLifecycle | None = None,
         start_date: str | None = "2024-01-01T10:20:30+01:02",
         lan_switch_interconnect_description: str | None = None,
-        lan_switch_interconnect_ip_network: IPv4NetworkType | None = None,
-        address_space: AddressSpace | None = None,
         minimum_links: int | None = None,
         router_side_node: UUIDstr | None = None,
         router_side_ae_iface: str | None = None,
         router_side_ae_members: list[dict[str, str]] | None = None,
-        router_side_ipv4_address: IPv4AddressType | None = None,
         switch_side_switch: UUIDstr | None = None,
         switch_side_ae_iface: str | None = None,
         switch_side_ae_members: list[dict[str, str]] | None = None,
-        switch_side_ipv4_address: IPv4AddressType | None = None,
         *,
         is_imported: bool | None = True,
     ) -> UUIDstr:
@@ -70,24 +65,18 @@ def lan_switch_interconnect_subscription_factory(
         subscription.lan_switch_interconnect.lan_switch_interconnect_description = (
             lan_switch_interconnect_description or faker.sentence()
         )
-        subscription.lan_switch_interconnect.lan_switch_interconnect_ip_network = (
-            lan_switch_interconnect_ip_network or faker.ipv4_network()
-        )
-        subscription.lan_switch_interconnect.address_space = address_space or AddressSpace.PRIVATE
         subscription.lan_switch_interconnect.minimum_links = minimum_links or 1
         subscription.lan_switch_interconnect.router_side = LanSwitchInterconnectRouterSideBlockInactive.new(
             uuid4(),
             node=router_side_node or Router.from_subscription(router_subscription_factory()).router,
             ae_iface=router_side_ae_iface or faker.network_interface(),
             ae_members=router_side_ae_members,
-            ipv4_address=router_side_ipv4_address or faker.ipv4(),
         )
         subscription.lan_switch_interconnect.switch_side = LanSwitchInterconnectSwitchSideBlockInactive.new(
             uuid4(),
             switch=switch_side_switch or Switch.from_subscription(switch_subscription_factory()).switch,
             ae_iface=switch_side_ae_iface or faker.network_interface(),
             ae_members=switch_side_ae_members,
-            ipv4_address=switch_side_ipv4_address or faker.ipv4(),
         )
 
         subscription = SubscriptionModel.from_other_lifecycle(subscription, SubscriptionLifecycle.ACTIVE)