| README.md | ●●●●● patch | view | raw | blame | history | |
| lib/Ipam.py | ●●●●● patch | view | raw | blame | history | |
| test/IpamPoolTest.py | ●●●●● patch | view | raw | blame | history |
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 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,6 +75,7 @@ else: address = ipaddress.ip_address(address) 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: @@ -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) 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')