uncloud/uncloud_django_based/uncloud/uncloud_net/models.py

import uuid
import ipaddress

from django.db import models
from django.contrib.auth import get_user_model
from django.core.validators import MinValueValidator, MaxValueValidator


from uncloud_pay.models import Product, RecurringPeriod
from uncloud.models import UncloudModel, UncloudStatus


class MACAdress(models.Model):
    default_prefix = 0x420000000000

class VPNPool(UncloudModel):
    """
    Network address pools from which VPNs can be created
    """

    uuid = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)

    network = models.GenericIPAddressField(unique=True)
    network_size = models.IntegerField(validators=[MinValueValidator(0),
                                                   MaxValueValidator(128)])

    subnetwork_size = models.IntegerField(validators=[
                                              MinValueValidator(0),
                                              MaxValueValidator(128)
                                          ])

    vpn_hostname = models.CharField(max_length=256)

    wireguard_private_key = models.CharField(max_length=48)

    @property
    def num_maximum_networks(self):
        """
        sample:
        network_size = 40
        subnetwork_size = 48
        maximum_networks = 2^(48-40)

        2nd sample:
        network_size = 8
        subnetwork_size = 24
        maximum_networks = 2^(24-8)
        """

        return 2**(subnetwork_size - network_size)

    @property
    def used_networks(self):
        return self.vpnnetworkreservation_set.objects.filter(vpnpool=self, status='used')

    @property
    def num_used_networks(self):
        return len(self.used_networks)

    @property
    def num_free_networks(self):
        return self.num_maximum_networks - self.num_used_networks

    @property
    def next_free_network(self):
        free_net = self.vpnnetworkreservation_set.objects.filter(vpnpool=self,
                                                                 status='free')

        used_net = self.vpnnetworkreservation_set.objects.filter(vpnpool=self,
                                                                 status='used')

        this_net = ipaddress.ip_network("{}/{}".format(
            self.network, self.network_size))


        if num_free_networks == 0:
            raise Exception("No free networks")

        if len(free_net) > 0:
            return free_net[0].address

        if len(used_net) > 0:
            """
            sample:

            pool = 2a0a:e5c1:200::/40
            last_used = 2a0a:e5c1:204::/48

            next:
            """

            last_net = ipaddress.ip_network(self.used_networks.last())
            last_net_ip = last_net[0]

            if last_net_ip.version == 6:
                offset_to_next = 2**(128 - self.subnetwork_size)
            elif last_net_ip.version == 4:
                offset_to_next = 2**(32 - self.subnetwork_size)

            next_net_ip = last_net_ip + offset_to_next

            return next_net_ip



class VPNNetworkReservation(UncloudModel):
    """
     This class tracks the used VPN networks. It will be deleted, when the product is cancelled.
     """
    vpnpool = models.ForeignKey(VPNPool,
                                 on_delete=models.CASCADE)
    address = models.GenericIPAddressField(primary_key=True)

    status = models.CharField(max_length=256,
                              default='used',
                              choices = (
                                  ('used', 'used'),
                                  ('free', 'free')
                              )
    )


class VPNNetwork(Product):
    """
    A selected network. Used for tracking reservations / used networks
    """
    network = models.ForeignKey(VPNNetworkReservation,
                               on_delete=models.CASCADE,
                               editable=False)

    wireguard_public_key = models.CharField(max_length=48)