Add infra_ipv6 attribute to devices

alembic/versions/2776e92f7b02_add_infra_ipv6_address_field_to_device.py

@@ -0,0 +1,28 @@
+"""Add infra IPv6 address field to Device
+
+Revision ID: 2776e92f7b02
+Revises: adcce7d9baaa
+Create Date: 2023-03-17 15:26:18.242019
+
+"""
+from alembic import op
+import sqlalchemy as sa
+import sqlalchemy_utils
+
+# revision identifiers, used by Alembic.
+revision = '2776e92f7b02'
+down_revision = 'adcce7d9baaa'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('device', sa.Column('infra_ipv6', sqlalchemy_utils.types.ip_address.IPAddressType(length=50), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('device', 'infra_ipv6')
+    # ### end Alembic commands ###

alembic/versions/2f9faee221a7_add_ipv6_gw_field_to_mgmtdomain.py

@@ -0,0 +1,24 @@
+"""Add ipv6_gw field to Mgmtdomain
+
+Revision ID: 2f9faee221a7
+Revises: b7629362583c
+Create Date: 2022-10-26 13:52:12.466111
+
+"""
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "2f9faee221a7"
+down_revision = "b7629362583c"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column("mgmtdomain", sa.Column("ipv6_gw", sa.Unicode(43)))
+
+
+def downgrade():
+    op.drop_column("mgmtdomain", "ipv6_gw")

alembic/versions/adcce7d9baaa_add_device_secondary_management_ip_column.py

@@ -0,0 +1,28 @@
+""""Add secondary_management_ip to Device
+
+Revision ID: adcce7d9baaa
+Revises: 2f9faee221a7
+Create Date: 2023-01-11 15:18:12.188994
+
+"""
+import sqlalchemy as sa
+import sqlalchemy_utils
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "adcce7d9baaa"
+down_revision = "2f9faee221a7"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "device",
+        sa.Column("secondary_management_ip", sqlalchemy_utils.types.ip_address.IPAddressType(length=50), nullable=True),
+    )
+
+
+def downgrade():
+    op.drop_column("device", "secondary_management_ip")

docker/api/config/api.yml

@@ -9,5 +9,6 @@ certpath: /tmp/devicecerts/
 global_unique_vlans: True
 init_mgmt_timeout: 30
 mgmtdomain_reserved_count: 5
+mgmtdomain_primary_ip_version: 4
 commit_confirmed_mode: 1
 commit_confirmed_timeout: 300

docs/configuration/index.rst

@@ -38,6 +38,9 @@ Defines parameters for the API:
   each defined management domain when assigning new management IP addresses to devices.
   Defaults to 5 (e.g. meaning 10.0.0.1 through 10.0.0.5 would remain unassigned on
   a domain for 10.0.0.0/24).
+- mgmtdomain_primary_ip_version: For dual stack management domains, this setting
+  defines whether IP version 4 or 6 is preferred when an access device's primary
+  management address is assigned. The only valid values are therefore 4 and 6.
 - commit_confirmed_mode: Integer specifying default commit confirm mode
   (see :ref:`commit_confirm_modes`). Defaults to 1.
 - commit_confirmed_timeout: Time to wait before rolling back an unconfirmed commit,

src/cnaas_nms/api/device.py

@@ -73,6 +73,7 @@
         "description": fields.String(required=False),
         "management_ip": fields.String(required=False),
         "infra_ip": fields.String(required=False),
+        "infra_ipv6": fields.String(required=False),
         "dhcp_ip": fields.String(required=False),
         "serial": fields.String(required=False),
         "ztp_mac": fields.String(required=False),

src/cnaas_nms/api/mgmtdomain.py

@@ -1,4 +1,4 @@
-from ipaddress import IPv4Interface
+from ipaddress import IPv4Interface, IPv6Interface
 from typing import Optional
 
 from flask import request
@@ -29,6 +29,7 @@
         "device_b": fields.String(required=True),
         "vlan": fields.Integer(required=True),
         "ipv4_gw": fields.String(required=True),
+        "ipv6_gw": fields.String(required=True),
         "description": fields.String(required=False),
     },
 )
@@ -37,6 +38,7 @@
 class f_mgmtdomain(BaseModel):
     vlan: Optional[int] = vlan_id_schema_optional
     ipv4_gw: Optional[str] = None
+    ipv6_gw: Optional[str] = None
     description: Optional[str] = None
 
     @validator("ipv4_gw")
@@ -56,6 +58,24 @@ def ipv4_gw_valid_address(cls, v, values, **kwargs):
 
         return v
 
+    @validator("ipv6_gw")
+    @classmethod
+    def ipv6_gw_valid_address(cls, v, values, **kwargs):
+        try:
+            addr = IPv6Interface(v)
+            prefix_len = int(addr.network.prefixlen)
+        except Exception:  # noqa: S110
+            raise ValueError("Invalid ipv6_gw received. Must be correct IPv6 address with mask")
+        else:
+            if addr.ip == addr.network.network_address:
+                raise ValueError("Specify gateway address, not subnet address")
+            if addr.ip == addr.network.broadcast_address:
+                raise ValueError("Specify gateway address, not broadcast address")
+            if prefix_len >= 126 or prefix_len <= 63:
+                raise ValueError("Bad prefix length {} for management network".format(prefix_len))
+
+        return v
+
 
 class MgmtdomainByIdApi(Resource):
     @jwt_required
@@ -163,12 +183,20 @@ def post(self):
             except ValidationError as e:
                 errors += parse_pydantic_error(e, f_mgmtdomain, json_data)
 
-            required_keys = ["device_a", "device_b", "vlan", "ipv4_gw"]
-            if all([key in data for key in required_keys]) and all([key in json_data for key in required_keys]):
+            required_keys_1 = ["device_a", "device_b", "vlan", "ipv4_gw"]
+            required_keys_2 = ["device_a", "device_b", "vlan", "ipv6_gw"]
+            required_in_data = all(key in data for key in required_keys_1) or all(
+                key in data for key in required_keys_2
+            )
+            required_in_json_data = all(key in json_data for key in required_keys_1) or all(
+                key in json_data for key in required_keys_2
+            )
+            if required_in_data and required_in_json_data:
                 new_mgmtd = Mgmtdomain()
                 new_mgmtd.device_a = data["device_a"]
                 new_mgmtd.device_b = data["device_b"]
                 new_mgmtd.ipv4_gw = data["ipv4_gw"]
+                new_mgmtd.ipv6_gw = data["ipv6_gw"]
                 new_mgmtd.vlan = data["vlan"]
                 try:
                     session.add(new_mgmtd)
@@ -184,7 +212,11 @@ def post(self):
                 device_b.synchronized = False
                 return empty_result(status="success", data={"added_mgmtdomain": new_mgmtd.as_dict()}), 200
             else:
-                errors.append("Not all required inputs were found: {}".format(", ".join(required_keys)))
+                errors.append(
+                    "Not all required inputs were found: {} OR {}".format(
+                        ", ".join(required_keys_1), ", ".join(required_keys_2)
+                    )
+                )
                 return empty_result("error", errors), 400
 
 

src/cnaas_nms/app_settings.py

@@ -2,7 +2,7 @@
 from typing import Optional
 
 import yaml
-from pydantic import BaseSettings, PostgresDsn
+from pydantic import BaseSettings, PostgresDsn, validator
 
 
 class AppSettings(BaseSettings):
@@ -49,10 +49,18 @@ class ApiSettings(BaseSettings):
     GLOBAL_UNIQUE_VLANS: bool = True
     INIT_MGMT_TIMEOUT: int = 30
     MGMTDOMAIN_RESERVED_COUNT: int = 5
+    MGMTDOMAIN_PRIMARY_IP_VERSION: int = 4
     COMMIT_CONFIRMED_MODE: int = 1
     COMMIT_CONFIRMED_TIMEOUT: int = 300
     SETTINGS_OVERRIDE: Optional[dict] = None
 
+    @validator("MGMTDOMAIN_PRIMARY_IP_VERSION")
+    @classmethod
+    def primary_ip_version_is_valid(cls, version: int) -> int:
+        if version not in (4, 6):
+            raise ValueError("must be either 4 or 6")
+        return version
+
 
 def construct_api_settings() -> ApiSettings:
     api_config = Path("/etc/cnaas-nms/api.yml")
@@ -79,6 +87,7 @@ def construct_api_settings() -> ApiSettings:
             GLOBAL_UNIQUE_VLANS=config.get("global_unique_vlans", True),
             INIT_MGMT_TIMEOUT=config.get("init_mgmt_timeout", 30),
             MGMTDOMAIN_RESERVED_COUNT=config.get("mgmtdomain_reserved_count", 5),
+            MGMTDOMAIN_PRIMARY_IP_VERSION=config.get("mgmtdomain_primary_ip_version", 4),
             COMMIT_CONFIRMED_MODE=config.get("commit_confirmed_mode", 1),
             COMMIT_CONFIRMED_TIMEOUT=config.get("commit_confirmed_timeout", 300),
             SETTINGS_OVERRIDE=config.get("settings_override", None),

src/cnaas_nms/db/device.py

@@ -77,8 +77,10 @@ class Device(cnaas_nms.db.base.Base):
     site = relationship("Site")
     description = Column(Unicode(255))
     management_ip = Column(IPAddressType)
+    secondary_management_ip = Column(IPAddressType)
     dhcp_ip = Column(IPAddressType)
     infra_ip = Column(IPAddressType)
+    infra_ipv6 = Column(IPAddressType)
     oob_ip = Column(IPAddressType)
     serial = Column(String(64))
     ztp_mac = Column(String(12))
@@ -105,7 +107,7 @@ def as_dict(self) -> dict:
                 value = value.name
             elif issubclass(value.__class__, cnaas_nms.db.base.Base):
                 continue
-            elif issubclass(value.__class__, ipaddress.IPv4Address):
+            elif issubclass(value.__class__, ipaddress._BaseAddress):
                 value = str(value)
             elif issubclass(value.__class__, datetime.datetime):
                 value = str(value)
@@ -376,38 +378,29 @@ def validate(cls, new_entry=True, **kwargs):
         if "description" in kwargs:
             data["description"] = kwargs["description"]
 
-        if "management_ip" in kwargs:
-            if kwargs["management_ip"]:
-                try:
-                    addr = ipaddress.IPv4Address(kwargs["management_ip"])
-                except Exception:
-                    errors.append("Invalid management_ip received. Must be correct IPv4 address.")
-                else:
-                    data["management_ip"] = addr
-            else:
-                data["management_ip"] = None
-
-        if "infra_ip" in kwargs:
-            if kwargs["infra_ip"]:
-                try:
-                    addr = ipaddress.IPv4Address(kwargs["infra_ip"])
-                except Exception:
-                    errors.append("Invalid infra_ip received. Must be correct IPv4 address.")
+        for ip_field in ("management_ip", "secondary_management_ip", "infra_ip", "dhcp_ip"):
+            if ip_field in kwargs:
+                if kwargs[ip_field]:
+                    try:
+                        addr = ipaddress.ip_address(kwargs[ip_field])
+                    except Exception:
+                        errors.append("Invalid {} received. Must be a valid IP address.".format(ip_field))
+                    else:
+                        data[ip_field] = addr
                 else:
-                    data["infra_ip"] = addr
-            else:
-                data["infra_ip"] = None
-
-        if "dhcp_ip" in kwargs:
-            if kwargs["dhcp_ip"]:
-                try:
-                    addr = ipaddress.IPv4Address(kwargs["dhcp_ip"])
-                except Exception:
-                    errors.append("Invalid dhcp_ip received. Must be correct IPv4 address.")
+                    data[ip_field] = None
+
+        for ipv6_field in ("infra_ipv6",):
+            if ipv6_field in kwargs:
+                if kwargs[ipv6_field]:
+                    try:
+                        addr = ipaddress.IPv6Address(kwargs[ipv6_field])
+                    except Exception:
+                        errors.append("Invalid {} received. Must be a valid IPv6 address.".format(ipv6_field))
+                    else:
+                        data[ipv6_field] = addr
                 else:
-                    data["dhcp_ip"] = addr
-            else:
-                data["dhcp_ip"] = None
+                    data[ipv6_field] = None
 
         if "serial" in kwargs:
             try: