Add add_next_subnet (subnet auto-add)

add_next_subnet tries to add a subnet in a parent subnet,
given the parent subnet and wanted prefixlen.

Signed-off-by: Damien Claisse <[email protected]>

ipam/client/abstractipam.py

@@ -18,6 +18,10 @@ def add_next_ip(self, subnet, dnsname, description):
     def get_next_free_ip(self, subnet):
         raise NotImplementedError()
 
+    @abstractmethod
+    def add_next_subnet(self, parent_subnet, prefixlen, description):
+        raise NotImplementedError()
+
     @abstractmethod
     def get_hostname_by_ip(self, ip):
         raise NotImplementedError()

ipam/client/backends/phpipam.py

@@ -6,11 +6,25 @@
 
 DEFAULT_IPAM_DB_TYPE = 'mysql'
 
+DEFAULT_SUBNET_OPTIONS = {
+    # Default permissions: read-only for guests and operators
+    'permissions': '{"2":"1","3":"1"}',
+    'vlan_id': 0,
+    'vrf_id': 0,
+}
+
 
 class PHPIPAM(AbstractIPAM):
 
     def __init__(self, params):
         dbtype = DEFAULT_IPAM_DB_TYPE
+        subnet_options = DEFAULT_SUBNET_OPTIONS.copy()
+        for (option, value) in DEFAULT_SUBNET_OPTIONS.items():
+            param_name = 'subnet_{}'.format(option)
+            if params.get(param_name):
+                value = params[param_name]
+            subnet_options[option] = value
+        self.subnet_options = subnet_options
         section_name = 'Production'
         if 'section_name' in params:
             section_name = params['section_name']
@@ -141,6 +155,86 @@ def get_allocated_ips_by_subnet_id(self, subnetid):
         iplist = [ip_address(int(ip[0])) for ip in self.cur]
         return iplist
 
+    def add_next_subnet(self, parent_subnet, prefixlen, description):
+        """
+        Find a subnet prefixlen-wide in parent_subnet, insert it into IPAM,
+        and return it.
+        """
+        if prefixlen <= parent_subnet.prefixlen:
+            raise ValueError('Parent subnet {} is too small to add new '
+                             'subnet with prefixlen {}!'
+                             ''.format(parent_subnet, prefixlen))
+
+        parent_subnet_id = self.find_subnet_id(parent_subnet)
+        if not parent_subnet_id:
+            raise ValueError('Unable to get subnet id from database '
+                             'for parent subnet {}'.format(parent_subnet))
+
+        parent_subnet_used_ips = self.get_allocated_ips_by_subnet_id(
+            parent_subnet_id)
+        if len(parent_subnet_used_ips):
+            raise ValueError('Parent subnet {} must not contain any '
+                             'allocated IP address!'.format(parent_subnet))
+
+        subnet = self._get_next_free_subnet(parent_subnet, parent_subnet_id,
+                                            prefixlen)
+        if not subnet:
+            raise ValueError('No more space to add a new subnet with '
+                             'prefixlen {} in {}!'.format(
+                                prefixlen, parent_subnet))
+
+        # Everything is in order, insert our subnet in IPAM
+        self.cur.execute(
+            'INSERT INTO subnets '
+            '(subnet, mask, sectionId, description, vrfId, '
+            'masterSubnetId, vlanId, permissions) '
+            'VALUES (\'{:d}\', \'{}\', \'{}\', \'{}\', '
+            '\'{}\', \'{}\', \'{}\', \'{}\')'.format(
+                int(subnet.network_address),
+                subnet.prefixlen,
+                self.section_id,
+                description,
+                self.subnet_options['vrf_id'],
+                parent_subnet_id,
+                self.subnet_options['vlan_id'],
+                self.subnet_options['permissions']))
+        return subnet
+
+    def _get_next_free_subnet(self, subnet, subnet_id, prefixlen):
+        """
+        Find next free prefixlen-wide subnet in a given subnet.
+        """
+        allocated_subnets = self._get_allocated_subnets(
+            subnet_id, prefixlen)
+
+        for candidate_subnet in subnet.subnets(new_prefix=prefixlen):
+            is_overlapping = False
+            # A candidate subnet is free if it doesn't overlap any other
+            # allocated subnet
+            for allocated_subnet in allocated_subnets:
+                if candidate_subnet.overlaps(allocated_subnet):
+                    is_overlapping = True
+                    # Since one subnet is overlapping, don't check the others
+                    break
+            if not is_overlapping:
+                return candidate_subnet
+        return None
+
+    def _get_allocated_subnets(self, subnet_id, prefixlen):
+        """
+        Return list of unavailable children subnets in a parent subnet,
+        given its id.
+        """
+        self.cur.execute('SELECT subnet, mask FROM subnets '
+                         'WHERE masterSubnetId={} '
+                         'ORDER BY subnet ASC'.format(subnet_id))
+
+        allocated_subnets = [
+            ip_network('{}/{}'.format(ip_address(int(row[0])), int(row[1])))
+            for row in self.cur
+        ]
+        return allocated_subnets
+
     def delete_ip(self, ipaddress):
         """Delete an IP address in IPAM. ipaddress must be an
         instance of ip_interface with correct prefix length.

ipam/client/tests/data/db.sql

@@ -81,6 +81,8 @@ INSERT INTO "subnets" VALUES (4,'168034304','31',2,'TEST /31 SUBNET GROUP',0,0,1
 INSERT INTO "subnets" VALUES (5,'168099840','31',2,'TEST /31 SUBNET GROUP',0,0,1,10,0,'{"2":"1","3":"1"}',0,0,NULL);
 INSERT INTO "subnets" VALUES (6,'42540488161975842760550356425300246592','125',2,'TEST IPv6 /126 SUBNET',0,0,1,10,0,'{"2":"1","3":"1"}',0,0,NULL);
 INSERT INTO "subnets" VALUES (7,'42540488161975842760550356425300246608','127',2,'TEST IPv6 /127 SUBNET',0,0,1,10,0,'{"2":"1","3":"1"}',0,0,NULL);
+INSERT INTO "subnets" VALUES (8,'168427520','24',2,'TEST IPv4 /24 SUBNET',0,0,1,10,0,'{"2":"1","3":"1"}',0,0,NULL);
+INSERT INTO "subnets" VALUES (9,'42540766411362381960998550477184434176','48',2,'TEST IPv6 /48 SUBNET',0,0,1,10,0,'{"2":"1","3":"1"}',0,0,NULL);
 CREATE INDEX "subnets_subnet" ON "subnets" ("subnet");
 CREATE INDEX "sections_id" ON "sections" ("id");
 CREATE INDEX "ipaddresses_dns_name" ON "ipaddresses" ("dns_name");

ipam/client/tests/test_phpipam.py

@@ -54,6 +54,22 @@ def test_db_fail():
         PHPIPAM(params)
 
 
+def test_subnet_options(testdb, testphpipam):
+
+    assert testphpipam.subnet_options['vlan_id'] == 0
+    assert testphpipam.subnet_options['vrf_id'] == 0
+
+    params = {'section_name': 'Production', 'dbtype': 'sqlite',
+              'database_uri': testdb, 'subnet_vlan_id': 42,
+              'subnet_vrf_id': 43, 'subnet_permissions': 'test'}
+
+    testipam = PHPIPAM(params)
+
+    assert testipam.subnet_options['vlan_id'] == 42
+    assert testipam.subnet_options['vrf_id'] == 43
+    assert testipam.subnet_options['permissions'] == 'test'
+
+
 def test_set_section_id(testphpipam):
     testphpipam.set_section_id(42)
     assert testphpipam.section_id == 42
@@ -268,6 +284,63 @@ def test_add_next_ip(testphpipam):
     assert "is full" in str(excinfo.value)
 
 
+def test_add_next_subnet(testphpipam):
+    parent_subnet4 = ip_network('10.10.0.0/24')
+    parent_subnet6 = ip_network('2001:db8:42::/48')
+    for i in list(range(0, 3 + 1)):
+
+        description = 'add_next_subnet generated subnet4 {}'.format(i)
+        target_subnet = ip_network('10.10.0.{}/28'.format(16*i))
+
+        test_subnet = testphpipam.add_next_subnet(
+            parent_subnet4, 28, description)
+        assert target_subnet == test_subnet
+        assert test_subnet.prefixlen == 28
+        test_subnet = testphpipam.get_subnet_by_desc(description)
+        assert test_subnet['subnet'] == target_subnet
+        assert test_subnet['description'] == description
+
+        description = 'add_next_subnet generated subnet6 {}'.format(i)
+        target_subnet = ip_network('2001:db8:42::{:x}/120'.format(256*i))
+
+        test_subnet = testphpipam.add_next_subnet(
+            parent_subnet6, 120, description)
+        assert target_subnet == test_subnet
+        test_subnet = testphpipam.get_subnet_by_desc(description)
+        assert test_subnet['subnet'] == target_subnet
+        assert test_subnet['description'] == description
+
+    description = 'add_next_subnet generated subnet 14'
+    target_subnet = ip_network('10.10.0.128/25')
+
+    test_subnet = testphpipam.add_next_subnet(parent_subnet4, 25, description)
+    assert target_subnet == test_subnet
+    test_subnet = testphpipam.get_subnet_by_desc(description)
+    assert test_subnet['subnet'] == target_subnet
+    assert test_subnet['description'] == description
+
+    with pytest.raises(ValueError) as excinfo:
+        testphpipam.add_next_subnet(parent_subnet4, 25, 'err')
+    assert 'No more space to add a new subnet' in str(excinfo.value)
+
+    parent_subnet4 = ip_network('10.42.0.0/28')
+    with pytest.raises(ValueError) as excinfo:
+        testphpipam.add_next_subnet(parent_subnet4, 28, 'err')
+    assert 'too small to add new subnet' in str(excinfo.value)
+
+    parent_subnet4 = ip_network('10.42.0.64/28')
+    with pytest.raises(ValueError) as excinfo:
+        testphpipam.add_next_subnet(parent_subnet4, 29, 'err')
+    assert 'Unable to get subnet id' in str(excinfo.value)
+
+    parent_subnet4 = ip_network('10.10.0.0/24')
+    testphpipam.add_next_ip(parent_subnet4, 'test ip', 'test ip')
+
+    with pytest.raises(ValueError) as excinfo:
+        testphpipam.add_next_subnet(parent_subnet4, 28, 'err')
+    assert 'must not contain any allocated IP address' in str(excinfo.value)
+
+
 def test_delete_ip(testphpipam):
     iplist = testphpipam.get_ip_list_by_desc('test ip #2')
     assert iplist == [{'ip': ip_address('10.1.0.2'),

setup.py

@@ -3,12 +3,12 @@
 setup(
     name='ipam-client',
     packages=find_packages(exclude=['contrib', 'docs', 'tests*']),
-    version='0.2',
+    version='0.3',
     description='IPAM abstraction layer library',
     author='Criteo',
     author_email='[email protected]',
     url='https://github.com/criteo/ipam-client',
-    download_url='https://github.com/criteo/ipam-client/archive/v0.2.tar.gz',
+    download_url='https://github.com/criteo/ipam-client/archive/v0.3.tar.gz',
     keywords=['ipam', 'phpipam'],
     license='Apache',
     classifiers=[