From 1878defa477812781b78f00f5e389304217de698 Mon Sep 17 00:00:00 2001
From: Jacek Kowalski <Jacek@jacekk.info>
Date: Sun, 03 May 2026 12:45:19 +0000
Subject: [PATCH] Persist IPAM pool data
---
lib/Ipam.py | 33 +++++++++++++++++++++------------
1 files changed, 21 insertions(+), 12 deletions(-)
diff --git a/lib/Ipam.py b/lib/Ipam.py
index 12f4a56..73f4f67 100644
--- a/lib/Ipam.py
+++ b/lib/Ipam.py
@@ -12,8 +12,7 @@
def __init__(self, pool: str = None, options: dict = None, subPool: str = None, v6: bool = None):
if pool == '':
pool = None
- if options is None:
- options = {}
+ self.options = options or {}
if subPool == '':
subPool = None
@@ -42,8 +41,11 @@
if not self.subpool.subnet_of(self.pool):
raise InputValidationException('Subpool must be a subnet of pool')
+ self.validate = self.options.get('validate', '1') == '1'
+ self.ptp = self.options.get('ptp', '0') == '1'
+
self.allocations = set()
- self.current = self.subpool.hosts()
+ self.current = self.subpool.hosts() if not self.ptp else self.subpool.__iter__()
self.v6 = isinstance(self.pool, ipaddress.IPv6Network)
@@ -62,7 +64,7 @@
for address in self.current:
if not self._is_allocated(address):
return address
- self.current = self.subpool.hosts()
+ self.current = self.subpool.hosts() if not self.ptp else self.subpool.__iter__()
for address in self.current:
if not self._is_allocated(address):
return address
@@ -74,19 +76,23 @@
else:
address = ipaddress.ip_address(address)
- if self.pool.network_address == address:
- raise InputValidationException('Cannot allocate network address to a host')
- if not self.v6 and self.pool.broadcast_address == address:
- raise InputValidationException('Cannot allocate broadcast address to a host')
+ if not self.ptp:
+ if self.pool.network_address == address:
+ raise InputValidationException('Cannot allocate network address to a host')
+ if not self.v6 and self.pool.broadcast_address == address:
+ raise InputValidationException('Cannot allocate broadcast address to a host')
if address not in self.pool:
raise InputValidationException('Requested address does not belong to a pool')
address = str(address)
- if self._is_allocated(address):
+ if self.validate and self._is_allocated(address):
raise InputValidationException('Requested address {} is already used'.format(address))
self.allocations.add(address)
- return '{}/{}'.format(address, self.pool.prefixlen)
+ prefixlen = self.pool.prefixlen
+ if self.ptp:
+ prefixlen = 128 if self.v6 else 32
+ return '{}/{}'.format(address, prefixlen)
def deallocate(self, address: str):
address = ipaddress.ip_address(address)
@@ -114,7 +120,10 @@
check[str(pool)] = pool
return str(pool)
- def get_pool(self, pool: str) -> Pool:
+ def has_pool(self, pool_id: str) -> bool:
+ return pool_id in self.pools or pool_id in self.pools6
+
+ def get_pool(self, pool: str | Pool) -> Pool:
pool = str(pool)
if pool in self.pools:
return self.pools[pool]
@@ -123,7 +132,7 @@
else:
raise InputValidationException('Unknown pool {}'.format(pool))
- def remove_pool(self, pool: str):
+ def remove_pool(self, pool: str | Pool):
pool = str(pool)
if pool in self.pools:
del self.pools[pool]
--
Gitblit v1.10.0