From 0700f847c59d44155ad89917b52dd58dc89f3b73 Mon Sep 17 00:00:00 2001
From: Jacek Kowalski <Jacek@jacekk.info>
Date: Wed, 11 Mar 2026 19:24:05 +0000
Subject: [PATCH] Allocate point-to-point addresses when option ptp=1 is specified
---
lib/Ipam.py | 23 ++++++++++++++---------
test/IpamPoolTest.py | 16 ++++++++++++++++
README.md | 16 ++++++++++++++++
3 files changed, 46 insertions(+), 9 deletions(-)
diff --git a/README.md b/README.md
index 717d384..c4360c0 100644
--- a/README.md
+++ b/README.md
@@ -26,6 +26,22 @@
Check out [`test_integration.sh`](test_integration.sh) for more examples.
+## Options
+
+To use options, add `--ipam-opt option=value` as an argument of `docker network create`:
+
+```bash
+docker network create --ipam-driver jacekkow/pyipam:latest --ipam-opt ptp=1 new-network
+```
+
+Available options:
+
+`ptp=1`
+
+When set addresses with netmask /32 (IPv4) or /128 (IPv6) are handed out.
+In this mode all IP addresses are handed out from the subnet,
+including ones that would be "network address" and "broadcast address"!
+
## Manual packaging
In order to test this module in development environment, you can build it
diff --git a/lib/Ipam.py b/lib/Ipam.py
index 12f4a56..2cffed2 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,10 @@
if not self.subpool.subnet_of(self.pool):
raise InputValidationException('Subpool must be a subnet of pool')
+ 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 +63,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,10 +75,11 @@
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')
@@ -86,7 +88,10 @@
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)
diff --git a/test/IpamPoolTest.py b/test/IpamPoolTest.py
index 287bc53..2dda0a5 100644
--- a/test/IpamPoolTest.py
+++ b/test/IpamPoolTest.py
@@ -294,3 +294,19 @@
pool.deallocate('fe80::2')
self.assertEqual(pool.allocate('fe80::2'), 'fe80::2/125')
self.assertEqual(pool.allocate(), 'fe80::1/125')
+
+
+class TestPoolPointToPoint(unittest.TestCase):
+ def test_pool_allocate_ptp_ipv4(self):
+ pool = Pool(pool='127.0.0.0/30', options={'ptp': '1'})
+ self.assertEqual(pool.allocate(), '127.0.0.0/32')
+ self.assertEqual(pool.allocate(), '127.0.0.1/32')
+ self.assertEqual(pool.allocate(), '127.0.0.2/32')
+ self.assertEqual(pool.allocate(), '127.0.0.3/32')
+
+ def test_pool_allocate_ptp_ipv6(self):
+ pool = Pool(pool='fe80::/126', options={'ptp': '1'})
+ self.assertEqual(pool.allocate(), 'fe80::/128')
+ self.assertEqual(pool.allocate(), 'fe80::1/128')
+ self.assertEqual(pool.allocate(), 'fe80::2/128')
+ self.assertEqual(pool.allocate(), 'fe80::3/128')
--
Gitblit v1.10.0