WIP: Introduce new IP address attributes to remove intern_ip

serveradmin/serverdb/forms.py

@@ -24,7 +24,7 @@ def clean(self):
         # It makes no sense to add inet or supernet attributes to hosts of
         # ip_addr_type null because they would have to be empty anyways.
         inet_attribute = (
-            self.cleaned_data['attribute'].type in ('inet', 'supernet') and
+            self.cleaned_data['attribute'].type in ('inet', 'inet4', 'inet6', 'supernet') and
             self.instance.servertype.ip_addr_type == 'null'
         )
         if inet_attribute:

serveradmin/serverdb/migrations/0017_auto_20231206_0908.py

@@ -0,0 +1,53 @@
+# Generated by Django 3.2.20 on 2023-12-06 15:08
+
+from django.db import migrations, models
+import django.db.models.deletion
+import netfields.fields
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('serverdb', '0016_optional_servertype_for_relation'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='attribute',
+            name='supernet',
+            field=models.BooleanField(default=False),
+        ),
+        migrations.AlterField(
+            model_name='attribute',
+            name='type',
+            field=models.CharField(choices=[('string', 'string'), ('boolean', 'boolean'), ('relation', 'relation'), ('reverse', 'reverse'), ('number', 'number'), ('inet', 'inet'), ('inet4', 'inet4'), ('inet6', 'inet6'), ('macaddr', 'macaddr'), ('date', 'date'), ('datetime', 'datetime'), ('supernet', 'supernet'), ('domain', 'domain')], max_length=32),
+        ),
+        migrations.CreateModel(
+            name='ServerInet6Attribute',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('value', netfields.fields.InetAddressField(max_length=39)),
+                ('attribute', models.ForeignKey(db_index=False, limit_choices_to={'type': 'inet6'}, on_delete=django.db.models.deletion.CASCADE, to='serverdb.attribute')),
+                ('server', models.ForeignKey(db_index=False, on_delete=django.db.models.deletion.CASCADE, to='serverdb.server')),
+            ],
+            options={
+                'db_table': 'server_inet6_attribute',
+                'unique_together': {('server', 'attribute', 'value')},
+                'index_together': {('attribute', 'value')},
+            },
+        ),
+        migrations.CreateModel(
+            name='ServerInet4Attribute',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('value', netfields.fields.InetAddressField(max_length=39)),
+                ('attribute', models.ForeignKey(db_index=False, limit_choices_to={'type': 'inet4'}, on_delete=django.db.models.deletion.CASCADE, to='serverdb.attribute')),
+                ('server', models.ForeignKey(db_index=False, on_delete=django.db.models.deletion.CASCADE, to='serverdb.server')),
+            ],
+            options={
+                'db_table': 'server_inet4_attribute',
+                'unique_together': {('server', 'attribute', 'value')},
+                'index_together': {('attribute', 'value')},
+            },
+        ),
+    ]

serveradmin/serverdb/models.py

@@ -13,7 +13,7 @@
     IPv6Interface,
     ip_network,
     IPv4Network,
-    IPv6Network,
+    IPv6Network, AddressValueError, NetmaskValueError,
 )
 from typing import Union
 
@@ -38,6 +38,8 @@
     'reverse': str,
     'number': lambda x: float(x) if '.' in str(x) else int(x),
     'inet': lambda x: inet_to_python(x),
+    'inet4': lambda x: inet4_to_python(x),
+    'inet6': lambda x: inet6_to_python(x),
     'macaddr': EUI,
     'date': str,
     'datetime': str,
@@ -77,6 +79,46 @@ def get_choices(types):
     return zip(*([types] * 2))
 
 
+def is_supernet_consistent(ip_address, server):
+    """Check if requested IP address is consistent with supernets for all other IP address of a server."""
+
+    if ip_address.version == 4:
+        server_attribute_cls = ServerInet4Attribute
+    else:
+        server_attribute_cls = ServerInet6Attribute
+
+    # A server belongs to different supernets depending on network servertype
+    supernet_attributes = Attribute.objects.filter(
+        type='supernet',
+        servertype_attributes__servertype_id=server.servertype_id,
+    )
+    supernet_servertypes = [x.target_servertype for x in supernet_attributes]
+
+    for supernet_servertype in supernet_servertypes:
+        supernet_1, _, _ = server.get_supernet(supernet_servertype)  # The result is guaranteed to be consistent
+
+        if supernet_1 is None:
+            continue
+
+        # Check the supernet of the new IP address
+        try:
+            supernet_2 = server_attribute_cls.objects.get(
+                value__net_contains_or_equals=ip_address,
+                server__servertype__ip_addr_type='network',
+                server__servertype_id=supernet_servertype,
+            ).server
+        except server_attribute_cls.DoesNotExist:
+            # There are some provider networks which are valid only for one address family.
+            # TODO: Enforce validation once the "local" network is gone.
+            continue
+
+        if supernet_1 != supernet_2:
+            raise ValidationError(
+                f'Non-matching {supernet_servertype} {supernet_2} for IP address {ip_address}, '
+                f'other IP addresses of {server.hostname} are in {supernet_1}'
+        )
+
+
 # TODO: Make validators out of the methods is_ip_address, is_unique and
 #       is_network and attach them to the model fields validators.
 def is_ip_address(ip_interface: Union[IPv4Interface, IPv6Interface]) -> None:
@@ -106,16 +148,30 @@ def is_unique_ip(ip_interface: Union[IPv4Interface, IPv6Interface],
     :return:
     """
 
+    if ip_interface.version == 4:
+        server_attribute_cls = ServerInet4Attribute
+    else:
+        server_attribute_cls = ServerInet6Attribute
+
     # We avoid querying the duplicate hosts here and giving the user
     # detailed information because checking with exists is cheaper than
     # querying the server and this is a validation and should be fast.
     has_duplicates = (
+        # TODO: Remove "intern_ip" support.
         Server.objects.filter(intern_ip=ip_interface).exclude(
             Q(servertype__ip_addr_type='network') |
             Q(server_id=object_id)
         ).exists() or
+        # TODO: Remove "primary_ip6" support.
         ServerInetAttribute.objects.filter(value=ip_interface).exclude(
-            server__servertype__ip_addr_type='network').exists())
+            Q(server__servertype__ip_addr_type='network') |
+            Q(server_id=object_id)
+        ).exists() or
+        server_attribute_cls.objects.filter(value=ip_interface).exclude(
+            Q(server__servertype__ip_addr_type='network') |
+            Q(server_id=object_id)
+        ).exists()
+    )
     if has_duplicates:
         raise ValidationError(
             'An object with {0} already exists'.format(str(ip_interface)))
@@ -151,6 +207,20 @@ def inet_to_python(obj: object) -> Union[IPv4Interface, IPv6Interface]:
     except ValueError as error:
         raise ValidationError(str(error))
 
+# WARNING: called only for edit->commit, not for commit!
+def inet4_to_python(obj: object) -> IPv4Interface:
+    try:
+        return IPv4Interface(obj)
+    except (AddressValueError, NetmaskValueError):
+        raise ValidationError(f'{obj} does not appear to be an IPv4 interface')
+
+
+def inet6_to_python(obj: object) -> IPv6Interface:
+    try:
+        return IPv6Interface(obj)
+    except (AddressValueError, NetmaskValueError):
+        raise ValidationError(f'{obj} does not appear to be an IPv6 interface')
+
 
 def network_overlaps(ip_interface: Union[IPv4Interface, IPv6Interface],
                      servertype_id: str, object_id: int) -> None:
@@ -226,6 +296,7 @@ def __init__(self, *args, **kwargs):
         max_length=32,
         choices=get_choices(ATTRIBUTE_TYPES.keys()),
     )
+    supernet = models.BooleanField(null=False, default=False)
     multi = models.BooleanField(null=False, default=False)
     hovertext = models.TextField(null=False, blank=True, default='')
     group = models.CharField(
@@ -438,10 +509,57 @@ def __str__(self):
         return self.hostname
 
     def get_supernet(self, servertype):
-        return Server.objects.get(
+        """Get a supernet of given servertype for the current server.
+
+        This function will check all IP addresses of a server which have the "supernet" feature enabled.
+        If data is inconsistent, an exception is raised.
+        No matching network for just some of the addresses does not mean inconsistency.
+        """
+
+        supernet_1 = None
+        supernet_ip_address = None
+        supernet_attribute = None
+
+        # TODO: Remove "intern_ip" support. Just remove this block of code below.
+        supernet_1 = Server.objects.get(
             servertype=servertype,
+            # It should probably match on ip_addr_type too, but we will remove this soon anyway.
             intern_ip__net_contains_or_equals=self.intern_ip,
         )
+        supernet_ip_address = self.intern_ip
+        supernet_attribute = None # Magic value for intern_ip
+
+        for server_attribute_cls in (ServerInet4Attribute, ServerInet6Attribute):
+            ip_addresses = server_attribute_cls.objects.filter(
+                server_id=self.server_id,
+                attribute__supernet=True,
+            )
+
+            # TODO: How to net_contains_or_equals for iterable?
+            for ip_address in ip_addresses:
+                try:
+                    attr = server_attribute_cls.objects.get(
+                        value__net_contains_or_equals=ip_address.value,
+                        server__servertype__ip_addr_type='network',
+                        server__servertype_id=servertype.servertype_id,
+                    )
+                    if not attr.attribute.supernet:
+                        raise ValidationError(f'Not a supernet: {servertype}!')
+                    supernet_2 = attr.server
+                    # TODO: Shouldn't we check that the requested servertype really point
+                except server_attribute_cls.DoesNotExist:
+                    continue
+                else:
+                    # Always trust the 1st found network
+                    if supernet_1 is None:
+                        supernet_1 = supernet_2
+                        supernet_ip_address = ip_address.value
+                        supernet_attribute = ip_addresses.attribute
+                    # Verify that all found networks match the 1st found one.
+                    elif supernet_1 != supernet_2:
+                        raise ValidationError(f'Can\'t determine {servertype} for {self.hostname}!')
+
+        return (supernet_1, supernet_ip_address, supernet_attribute)
 
     def clean(self):
         super(Server, self).clean()
@@ -474,6 +592,9 @@ def clean(self):
                 network_overlaps(self.intern_ip, self.servertype.servertype_id,
                                  self.server_id)
 
+            if ip_addr_type != 'null':
+                is_supernet_consistent(self.intern_ip, self)
+
     def get_attributes(self, attribute):
         model = ServerAttribute.get_model(attribute.type)
         return model.objects.filter(server=self, attribute=attribute)
@@ -520,6 +641,10 @@ def get_model(attribute_type):
             return ServerNumberAttribute
         if attribute_type == 'inet':
             return ServerInetAttribute
+        if attribute_type == 'inet4':
+            return ServerInet4Attribute
+        if attribute_type == 'inet6':
+            return ServerInet6Attribute
         if attribute_type == 'macaddr':
             return ServerMACAddressAttribute
         if attribute_type == 'date':
@@ -695,6 +820,96 @@ def clean(self):
             network_overlaps(self.value, self.server.servertype_id,
                              self.server.server_id)
 
+        is_supernet_consistent(self.value, self.server)
+
+
+class ServerInet4Attribute(ServerAttribute):
+    attribute = models.ForeignKey(
+        Attribute,
+        db_index=False,
+        on_delete=models.CASCADE,
+        limit_choices_to=dict(type='inet4'),
+    )
+    value = netfields.InetAddressField()
+
+    class Meta:
+        app_label = 'serverdb'
+        db_table = 'server_inet4_attribute'
+        unique_together = [['server', 'attribute', 'value']]
+        index_together = [['attribute', 'value']]
+
+    def clean(self):
+        super(ServerAttribute, self).clean()
+
+        if type(self.value) != IPv4Interface:
+            self.value = inet4_to_python(self.value)
+
+        # Get the ip_addr_type of the servertype
+        ip_addr_type = self.server.servertype.ip_addr_type
+
+        if ip_addr_type == 'null':
+            # A Servertype with ip_addr_type "null" and attributes of type
+            # inet must be denied per configuration. This is just a safety net
+            # in case e.g. somebody creates them programmatically.
+            raise ValidationError(
+                _('%(attribute_id)s must be null'), code='invalid value',
+                params={'attribute_id': self.attribute_id})
+        elif ip_addr_type == 'host':
+            is_ip_address(self.value)
+            is_unique_ip(self.value, self.server.server_id)
+        elif ip_addr_type == 'loadbalancer':
+            is_ip_address(self.value)
+        elif ip_addr_type == 'network':
+            is_network(self.value)
+            network_overlaps(self.value, self.server.servertype_id,
+                             self.server.server_id)
+
+        is_supernet_consistent(self.value, self.server)
+
+
+class ServerInet6Attribute(ServerAttribute):
+    attribute = models.ForeignKey(
+        Attribute,
+        db_index=False,
+        on_delete=models.CASCADE,
+        limit_choices_to=dict(type='inet6'),
+    )
+    value = netfields.InetAddressField()
+
+    class Meta:
+        app_label = 'serverdb'
+        db_table = 'server_inet6_attribute'
+        unique_together = [['server', 'attribute', 'value']]
+        index_together = [['attribute', 'value']]
+
+    def clean(self):
+        super(ServerAttribute, self).clean()
+
+        if type(self.value) != IPv6Interface:
+            self.value = inet6_to_python(self.value)
+
+        # Get the ip_addr_type of the servertype
+        ip_addr_type = self.server.servertype.ip_addr_type
+
+        if ip_addr_type == 'null':
+            # A Servertype with ip_addr_type "null" and attributes of type
+            # inet must be denied per configuration. This is just a safety net
+            # in case e.g. somebody creates them programmatically.
+            raise ValidationError(
+                _('%(attribute_id)s must be null'), code='invalid value',
+                params={'attribute_id': self.attribute_id})
+        elif ip_addr_type == 'host':
+            is_ip_address(self.value)
+            is_unique_ip(self.value, self.server.server_id)
+        elif ip_addr_type == 'loadbalancer':
+            is_ip_address(self.value)
+        elif ip_addr_type == 'network':
+            is_network(self.value)
+            network_overlaps(self.value, self.server.servertype_id,
+                             self.server.server_id)
+
+        is_supernet_consistent(self.value, self.server)
+
 
 class ServerMACAddressAttribute(ServerAttribute):
     attribute = models.ForeignKey(