Skip to content

S
slapos.core
Commit 1bc59b59 authored by Xavier Thompson's avatar Xavier Thompson
Browse files
Options

slap/standalone: Use IPv6 range when available 

See merge request nexedi/slapos.core!538
parents 8f9a4ffd 3da9c477
Hide whitespace changes
Inline Side-by-side
Showing with 257 additions and 82 deletions
slapos/format.py
View file @ 1bc59b59
......@@ -1464,9 +1464,15 @@ def parse_computer_definition(conf, definition_path):
address_list.append(dict(addr=address, netmask=netmask))
if computer_definition.has_option(section, 'ipv6_range'):
ipv6_range_network = computer_definition.get(section, 'ipv6_range')
addr, netmask = ipv6_range_network.split('/')
netmask = netmaskFromLenIPv6(int(netmask))
ipv6_range = {'addr' : address, 'netmask' : netmask, 'network' : ipv6_range_network}
ipv6_range_addr, ipv6_range_prefixlen = ipv6_range_network.split('/')
ipv6_range_prefixlen = int(ipv6_range_prefixlen)
ipv6_range_netmask = netmaskFromLenIPv6(ipv6_range_prefixlen)
ipv6_range = {
'addr' : ipv6_range_addr,
'netmask' : ipv6_range_netmask,
'network' : ipv6_range_network,
'prefixlen': ipv6_range_prefixlen,
}
else:
ipv6_range = {}
tap = Tap(computer_definition.get(section, 'network_interface'))
......
slapos/grid/promise/generic.py
View file @ 1bc59b59
......@@ -493,12 +493,8 @@ class GenericPromise(with_metaclass(ABCMeta, object)):
))
elif (not self.__is_tested and not check_anomaly) or \
(not self.__is_anomaly_detected and check_anomaly):
# Anomaly or Test is disabled on this promise, send empty result
if self.getConfig('slapgrid-version', '') <= '1.4.17':
# old version cannot send EmptyResult
self.__sendResult(PromiseQueueResult(item=TestResult()))
else:
self.__sendResult(PromiseQueueResult())
# Anomaly or Test is disabled on this promise, send empty
self.__sendResult(PromiseQueueResult())
else:
try:
self.sense()
......
slapos/slap/standalone.py
View file @ 1bc59b59
......@@ -59,12 +59,25 @@ from .interface.slap import IRequester
from ..grid.slapgrid import SLAPGRID_PROMISE_FAIL
from .slap import slap
from ..util import dumps, rmtree
from ..util import dumps, rmtree, getPartitionIpv6Addr, getPartitionIpv6Range
from ..grid.svcbackend import getSupervisorRPC
from ..grid.svcbackend import _getSupervisordSocketPath
NETMASK_IPV6_FULL = 'ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff'
NETMASK_IPV4_FULL = '255.255.255.255'
def _parseIPv6(ipv6):
try:
addr, prefixlen = ipv6.split('/')
prefixlen = int(prefixlen)
except ValueError:
addr, prefixlen = ipv6, None
return addr, prefixlen
@zope.interface.implementer(IException)
class SlapOSNodeCommandError(Exception):
"""Exception raised when running a SlapOS Node command failed.
......@@ -205,6 +218,7 @@ class SlapOSConfigWriter(ConfigWriter):
read_only_shared_part_list = '\n '.join( # pylint: disable=unused-variable; used in format()
standalone_slapos._shared_part_list)
partition_forward_configuration = '\n'.join(self._getPartitionForwardConfiguration())
has_ipv6_range = ('false', 'true')[standalone_slapos._partitions_have_ipv6_range]
with open(path, 'w') as f:
f.write(
textwrap.dedent(
......@@ -232,6 +246,7 @@ class SlapOSConfigWriter(ConfigWriter):
create_tap = false
create_tun = false
computer_xml = {standalone_slapos._slapos_xml}
partition_has_ipv6_range = {has_ipv6_range}
[slapproxy]
host = {standalone_slapos._server_ip}
......@@ -287,9 +302,7 @@ class SlapformatDefinitionWriter(ConfigWriter):
"""
def writeConfig(self, path):
ipv4 = self._standalone_slapos._ipv4_address
ipv6 = self._standalone_slapos._ipv6_address
ipv4_cidr = ipv4 + '/255.255.255.255' if ipv4 else ''
ipv6_cidr = ipv6 + '/64' if ipv6 else ''
ipv4_cidr = '%s/%s' % (ipv4, NETMASK_IPV4_FULL) if ipv4 else ''
user = pwd.getpwuid(os.getuid()).pw_name
partition_base_name = self._standalone_slapos._partition_base_name
with open(path, 'w') as f:
......@@ -299,12 +312,24 @@ class SlapformatDefinitionWriter(ConfigWriter):
[computer]
address = {ipv4_cidr}\n
""").format(**locals()))
ipv6 = self._standalone_slapos._ipv6_address
for i in range(self._standalone_slapos._partition_count):
ipv6_single, ipv6_range = self._standalone_slapos._getPartitionIpv6(i)
if ipv6_single:
ipv6_single_cidr = '%s/%s' % (ipv6_single, NETMASK_IPV6_FULL)
else:
ipv6_single_cidr = ''
if ipv6_range:
ipv6_range_cidr = '%(addr)s/%(prefixlen)s' % ipv6_range
ipv6_range_config_line = 'ipv6_range = ' + ipv6_range_cidr
else:
ipv6_range_config_line = ''
f.write(
textwrap.dedent(
"""
[partition_{i}]
address = {ipv6_cidr} {ipv4_cidr}
address = {ipv6_single_cidr} {ipv4_cidr}
{ipv6_range_config_line}
pathname = {partition_base_name}{i}
user = {user}
network_interface =\n
......@@ -415,7 +440,13 @@ class StandaloneSlapOS(object):
self._partition_base_name = 'slappart'
self._ipv4_address = None
self._ipv6_address = None
self._ipv6_range_prefixlen = None
self._partitions_have_ipv6_range = False
# NOTE: Using Standalone's own slapos (slapos.cli.entry) instead
# is not that easy because in test nodes standalone is often run
# with gpython (pygolang), and gpython currently doesn't support
# buildout
self._slapos_bin = slapos_bin
self._slapos_commands = {
......@@ -594,8 +625,12 @@ class StandaloneSlapOS(object):
partition_base_name="slappart"):
"""Creates `partition_count` partitions.
All partitions have the same `ipv4_address` and `ipv6_address` and
use the current system user.
All partitions have the same `ipv4_address` and use the current system
user.
`ipv6_address` can be a single address (in this case all partitions have
the same address) or a range in the form IPV6/CIDR (in this case each
partition has a subrange).
When calling this a second time with a lower `partition_count` or with
different `partition_base_name` will delete existing partitions.
......@@ -628,17 +663,19 @@ class StandaloneSlapOS(object):
if not (os.path.exists(partition_path)):
os.mkdir(partition_path)
os.chmod(partition_path, 0o750)
ipv6_addr, ipv6_range = self._getPartitionIpv6(i)
partition_list.append({
'address_list': [
{
'addr': ipv4_address,
'netmask': '255.255.255.255'
'netmask': NETMASK_IPV4_FULL
},
{
'addr': ipv6_address,
'netmask': 'ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff'
},
'addr': ipv6_addr,
'netmask': NETMASK_IPV6_FULL
}
],
'ipv6_range' : ipv6_range,
'path': partition_path,
'reference': partition_reference,
'tap': {
......@@ -665,7 +702,7 @@ class StandaloneSlapOS(object):
self.computer.updateConfiguration(
dumps({
'address': ipv4_address,
'netmask': '255.255.255.255',
'netmask': NETMASK_IPV4_FULL,
'partition_list': partition_list,
'reference': self._computer_id,
'instance_root': self._instance_root,
......@@ -694,11 +731,33 @@ class StandaloneSlapOS(object):
self._partition_count = partition_count
self._partition_base_name = partition_base_name
self._ipv4_address = ipv4_address
self._ipv6_address = ipv6_address
self._ipv6_address, prefixlen = _parseIPv6(ipv6_address)
self._ipv6_range_prefixlen = prefixlen
self._partitions_have_ipv6_range = bool(prefixlen) and prefixlen < 112
if old_partition_count != partition_count:
SlapOSConfigWriter(self).writeConfig(self._slapos_config)
SlapformatDefinitionWriter(self).writeConfig(self._slapformat_definition)
# remove slapos xml configuration in case of ip changes
try:
os.unlink(self._slapos_xml)
except OSError as e:
if e.errno != errno.ENOENT:
raise
# run slapos format --now
command = (
self._slapos_bin, 'node', 'format',
'--now',
'--cfg', self._slapos_config)
self._logger.debug("Running %s", command)
try:
output = subprocess.check_output(command, stderr=subprocess.STDOUT)
self._logger.info(output)
except subprocess.CalledProcessError as e:
self._logger.error(e.output)
raise
def supply(self, software_url, computer_guid=None, state="available"):
"""Supply a software, see ISupply.supply
......@@ -949,3 +1008,17 @@ class StandaloneSlapOS(object):
return
time.sleep(i * .01)
raise RuntimeError("SlapOS not started")
def _getPartitionIpv6(self, i):
# returns (single_ipv6_address, ipv6_range) for a partition
# ipv6_address can be either a range or a single IPv6 address (with no /)
prefixlen = self._ipv6_range_prefixlen
if prefixlen is None:
return self._ipv6_address, None
ipv6_range = {'addr': self._ipv6_address, 'prefixlen': prefixlen}
ipv6_single = getPartitionIpv6Addr(ipv6_range, i)['addr']
if self._partitions_have_ipv6_range:
ipv6_partition_range = getPartitionIpv6Range(ipv6_range, i, 16)
return ipv6_single, ipv6_partition_range
else:
return ipv6_single, None
slapos/tests/test_cli.py
View file @ 1bc59b59
......@@ -442,16 +442,13 @@ class TestCliBoot(CliMixin):
patch('slapos.cli.boot.ConfigCommand.config_path', return_value=slapos_conf.name), \
patch(
'slapos.cli.boot.netifaces.ifaddresses',
return_value={socket.AF_INET6: ({'addr': '2000::1'},),},) as ifaddresses,\
patch('slapos.cli.boot._ping_hostname', return_value=1) as _ping_hostname:
return_value={socket.AF_INET6: ({'addr': '2000::1'},),},) as ifaddresses:
app.run(('node', 'boot'))
# boot command runs as root
check_root_user.assert_called_once()
# it waits for interface to have an IPv6 address
ifaddresses.assert_called_once_with('interface_name_from_config')
# then ping master hostname to wait for connectivity
_ping_hostname.assert_called_once_with('slap.vifib.com')
# then format and bang
SlapOSApp().run.assert_any_call(['node', 'format', '--now', '--verbose'])
SlapOSApp().run.assert_any_call(['node', 'bang', '-m', 'Reboot'])
......
slapos/tests/test_standalone.py
View file @ 1bc59b59
......@@ -25,6 +25,7 @@
#
##############################################################################
import json
import unittest
import mock
import os
......@@ -40,6 +41,7 @@ import multiprocessing
from contextlib import closing
from six.moves.configparser import ConfigParser
import netaddr
import psutil
from slapos.slap.standalone import StandaloneSlapOS
......@@ -79,100 +81,186 @@ class TestSlapOSStandaloneSetup(unittest.TestCase):
def setUp(self):
checkPortIsFree()
def test_format(self):
def setupSimpleStandalone(self):
working_dir = tempfile.mkdtemp(prefix=__name__)
self.addCleanup(slapos.util.rmtree, working_dir)
standalone = StandaloneSlapOS(
working_dir, SLAPOS_TEST_IPV4, SLAPOS_TEST_PORT)
self.addCleanup(standalone.stop)
return standalone
@staticmethod
def getInstancePath(standalone, *segments):
return os.path.join(standalone.instance_directory, *segments)
def assertExists(self, path):
self.assertTrue(os.path.exists(path))
def assertNotExists(self, path):
self.assertFalse(os.path.exists(path))
@classmethod
def getJsonResourceList(cls, standalone):
return [
cls.getJson(
cls.getInstancePath(
standalone, 'slappart%d' % i, '.slapos-resource'))
for i in range(standalone._partition_count)]
@staticmethod
def getJson(path):
with open(path) as f:
return json.load(f)
def test_format(self):
standalone = self.setupSimpleStandalone()
standalone.format(3, SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV6)
self.assertExists(standalone.software_directory)
self.assertExists(standalone.instance_directory)
self.assertExists(
os.path.join(standalone.instance_directory, 'slappart0'))
self.assertExists(
os.path.join(standalone.instance_directory, 'slappart1'))
self.assertExists(
os.path.join(standalone.instance_directory, 'slappart2'))
for i in range(3):
self.assertExists(
self.getInstancePath(standalone, 'slappart%d' % i, '.slapos-resource'))
def test_format_ipv6_big_range(self):
standalone = self.setupSimpleStandalone()
prefixlen = 96
slapos_fake_ipv6_range = '%s/%d' % (SLAPOS_TEST_IPV6, prefixlen)
standalone.format(3, SLAPOS_TEST_IPV4, slapos_fake_ipv6_range)
resource_list = self.getJsonResourceList(standalone)
for i, resource in enumerate(resource_list):
resource_prefixlen = int(resource['ipv6_range']['network'].split('/')[1])
self.assertEqual(resource_prefixlen, prefixlen + 16)
self.assertTrue(netaddr.valid_ipv6(resource['address_list'][0]['addr']))
for other_resource in resource_list[i + 1:]:
self.assertNotEqual(
resource['ipv6_range']['addr'],
other_resource['ipv6_range']['addr'])
self.assertNotEqual(
resource['address_list'][0]['addr'],
other_resource['address_list'][0]['addr'])
def test_format_ipv6_small_range(self):
standalone = self.setupSimpleStandalone()
prefixlen = 112
slapos_fake_ipv6_range = '%s/%d' % (SLAPOS_TEST_IPV6, prefixlen)
addr0 = str(netaddr.IPNetwork(slapos_fake_ipv6_range).network)
standalone.format(3, SLAPOS_TEST_IPV4, slapos_fake_ipv6_range)
resource_list = self.getJsonResourceList(standalone)
for i, resource in enumerate(resource_list):
self.assertFalse(resource['ipv6_range'])
self.assertTrue(netaddr.valid_ipv6(resource['address_list'][0]['addr']))
for other_resource in resource_list[i + 1:]:
self.assertNotEqual(
resource['address_list'][0]['addr'],
other_resource['address_list'][0]['addr'])
self.assertNotEqual(
resource['address_list'][0]['addr'],
addr0)
def test_format_ipv6_very_small_range(self):
standalone = self.setupSimpleStandalone()
prefixlen = 126
slapos_fake_ipv6_range = '%s/%d' % (SLAPOS_TEST_IPV6, prefixlen)
addr0 = str(netaddr.IPNetwork(slapos_fake_ipv6_range).network)
standalone.format(8, SLAPOS_TEST_IPV4, slapos_fake_ipv6_range)
resource_list = self.getJsonResourceList(standalone)
for i, resource in enumerate(resource_list):
self.assertFalse(resource['ipv6_range'])
self.assertTrue(netaddr.valid_ipv6(resource['address_list'][0]['addr']))
for j, other_resource in enumerate(resource_list[i + 1:]):
self.assertNotEqual(
resource['address_list'][0]['addr'],
addr0)
if j % 2 == 1:
self.assertEqual(
resource['address_list'][0]['addr'],
other_resource['address_list'][0]['addr'])
else:
self.assertNotEqual(
resource['address_list'][0]['addr'],
other_resource['address_list'][0]['addr'])
def test_format_ipv6_slapsh_128_range(self):
standalone = self.setupSimpleStandalone()
prefixlen = 128
slapos_fake_ipv6_range = '%s/%d' % (SLAPOS_TEST_IPV6, prefixlen)
standalone.format(3, SLAPOS_TEST_IPV4, slapos_fake_ipv6_range)
resource_list = self.getJsonResourceList(standalone)
for i, resource in enumerate(resource_list):
self.assertFalse(resource['ipv6_range'])
self.assertTrue(netaddr.valid_ipv6(resource['address_list'][0]['addr']))
self.assertEqual(
resource['address_list'][0]['addr'],
SLAPOS_TEST_IPV6)
self.assertTrue(os.path.exists(standalone.software_directory))
self.assertTrue(os.path.exists(standalone.instance_directory))
self.assertTrue(
os.path.exists(
os.path.join(standalone.instance_directory, 'slappart0')))
self.assertTrue(
os.path.exists(
os.path.join(standalone.instance_directory, 'slappart1')))
self.assertTrue(
os.path.exists(
os.path.join(standalone.instance_directory, 'slappart2')))
def test_format_ipv6_no_range(self):
standalone = self.setupSimpleStandalone()
standalone.format(3, SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV6)
resource_list = self.getJsonResourceList(standalone)
for i, resource in enumerate(resource_list):
self.assertFalse(resource['ipv6_range'])
self.assertTrue(netaddr.valid_ipv6(resource['address_list'][0]['addr']))
for other_resource in resource_list[i + 1:]:
self.assertEqual(
resource['address_list'][0]['addr'],
SLAPOS_TEST_IPV6)
def test_reformat_less_partitions(self):
working_dir = tempfile.mkdtemp(prefix=__name__)
self.addCleanup(slapos.util.rmtree, working_dir)
standalone = StandaloneSlapOS(
working_dir, SLAPOS_TEST_IPV4, SLAPOS_TEST_PORT)
self.addCleanup(standalone.stop)
standalone = self.setupSimpleStandalone()
standalone.format(2, SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV6)
standalone.format(1, SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV6)
self.assertFalse(
os.path.exists(
os.path.join(standalone.instance_directory, 'slappart1')))
self.assertNotExists(
os.path.join(standalone.instance_directory, 'slappart1'))
self.assertEqual(
['slappart0'],
[cp.getId() for cp in standalone.computer.getComputerPartitionList()])
def test_reformat_less_chmod_files(self):
working_dir = tempfile.mkdtemp(prefix=__name__)
self.addCleanup(slapos.util.rmtree, working_dir)
standalone = StandaloneSlapOS(
working_dir, SLAPOS_TEST_IPV4, SLAPOS_TEST_PORT)
self.addCleanup(standalone.stop)
standalone = self.setupSimpleStandalone()
standalone.format(2, SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV6)
# removing this directory should not be a problem
chmoded_dir_path = os.path.join(standalone.instance_directory, 'slappart1', 'directory')
os.mkdir(chmoded_dir_path)
os.chmod(chmoded_dir_path, 0o000)
standalone.format(1, SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV6)
self.assertFalse(os.path.exists(chmoded_dir_path))
self.assertNotExists(chmoded_dir_path)
self.assertEqual(
['slappart0'],
[cp.getId() for cp in standalone.computer.getComputerPartitionList()])
def test_reformat_different_base_name(self):
working_dir = tempfile.mkdtemp(prefix=__name__)
self.addCleanup(slapos.util.rmtree, working_dir)
standalone = StandaloneSlapOS(
working_dir, SLAPOS_TEST_IPV4, SLAPOS_TEST_PORT)
self.addCleanup(standalone.stop)
standalone = self.setupSimpleStandalone()
standalone.format(
1, SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV6, partition_base_name="a")
self.assertTrue(
os.path.exists(os.path.join(standalone.instance_directory, 'a0')))
self.assertExists(os.path.join(standalone.instance_directory, 'a0'))
standalone.format(
1, SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV6, partition_base_name="b")
self.assertFalse(
os.path.exists(os.path.join(standalone.instance_directory, 'a0')))
self.assertTrue(
os.path.exists(os.path.join(standalone.instance_directory, 'b0')))
self.assertNotExists(os.path.join(standalone.instance_directory, 'a0'))
self.assertExists(os.path.join(standalone.instance_directory, 'b0'))
self.assertEqual(
['b0'],
[cp.getId() for cp in standalone.computer.getComputerPartitionList()])
def test_reformat_refuse_deleting_running_partition(self):
working_dir = tempfile.mkdtemp(prefix=__name__)
self.addCleanup(slapos.util.rmtree, working_dir)
standalone = StandaloneSlapOS(
working_dir, SLAPOS_TEST_IPV4, SLAPOS_TEST_PORT)
self.addCleanup(standalone.stop)
standalone = self.setupSimpleStandalone()
standalone.format(1, SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV6)
with mock.patch("slapos.slap.ComputerPartition.getState", return_value="busy"),\
self.assertRaisesRegex(ValueError, "Cannot reformat to remove busy partition at .*slappart0"):
standalone.format(0, SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV6)
def test_slapos_node_format(self):
working_dir = tempfile.mkdtemp(prefix=__name__)
self.addCleanup(slapos.util.rmtree, working_dir)
standalone = StandaloneSlapOS(
working_dir, SLAPOS_TEST_IPV4, SLAPOS_TEST_PORT)
self.addCleanup(standalone.stop)
self.assertTrue(os.path.exists(standalone.instance_directory))
standalone = self.setupSimpleStandalone()
self.assertExists(standalone.instance_directory)
format_command = (standalone._slapos_wrapper, 'node', 'format', '--now')
glob_pattern = os.path.join(standalone.instance_directory, 'slappart*')
self.assertFalse(glob.glob(glob_pattern))
self.assertTrue(subprocess.call(format_command))
self.assertTrue(subprocess.call(format_command)) # non-zero exitcode
self.assertFalse(glob.glob(glob_pattern))
for partition_count in (3, 2):
standalone.format(partition_count, SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV6)
......@@ -314,10 +402,14 @@ class SlapOSStandaloneTestCase(unittest.TestCase):
'SLAPOS_TEST_SHARED_PART_LIST', '').split(os.pathsep) if p
],
)
if self._auto_stop_standalone:
self.addCleanup(self.standalone.stop)
self.addCleanup(self.stopStandalone)
self.standalone.format(1, SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV6)
def stopStandalone(self):
if self._auto_stop_standalone:
self.standalone.stop()
self._auto_stop_standalone = False
class TestSlapOSStandaloneLogFile(SlapOSStandaloneTestCase):
def test_log_files(self):
......@@ -448,8 +540,6 @@ class TestSlapOSStandaloneSoftware(SlapOSStandaloneTestCase):
class TestSlapOSStandaloneInstance(SlapOSStandaloneTestCase):
_auto_stop_standalone = False # we stop explicitly
def test_request_instance(self):
with tempfile.NamedTemporaryFile(suffix="-%s.cfg" % self.id()) as f:
# This is a minimal / super fast buildout that's compatible with slapos.
......@@ -564,5 +654,5 @@ class TestSlapOSStandaloneInstance(SlapOSStandaloneTestCase):
if p['statename'] == 'RUNNING'
])
self.assertEqual(set([True]), set([p.is_running() for p in process_list]))
self.standalone.stop()
self.stopStandalone()
self.assertEqual(set([False]), set([p.is_running() for p in process_list]))
slapos/util.py
View file @ 1bc59b59
......@@ -38,6 +38,7 @@ import socket
import sqlite3
import struct
import subprocess
import sys
import warnings
import jsonschema
......@@ -201,7 +202,7 @@ def ipv6FromBin(ip, suffix=''):
if suffix_len > 0:
ip += suffix.rjust(suffix_len, '0')
elif suffix_len:
sys.exit("Prefix exceeds 128 bits")
sys.exit("Prefix %s exceeds 128 bits by %d bit" % (ip, -suffix_len))
return socket.inet_ntop(socket.AF_INET6,
struct.pack('>QQ', int(ip[:64], 2), int(ip[64:], 2)))
......@@ -214,11 +215,23 @@ def getPartitionIpv6Addr(ipv6_range, partition_index):
}
returns the IPv6 addr
addr::(partition_index+2) (address 1 is is used by re6st)
If the range is too small, wrap around
"""
addr = ipv6_range['addr']
prefixlen = ipv6_range['prefixlen']
prefix = binFromIpv6(addr)[:prefixlen]
return dict(addr=ipv6FromBin(prefix + bin(partition_index+2)[2:].zfill(128 - prefixlen)), prefixlen=prefixlen)
remaining = 128 - prefixlen
suffix = bin(partition_index+2)[2:]
if len(suffix) > remaining:
if remaining >= 2:
# skip reserved addresses 0 and 1
suffix = bin((partition_index % ((1 << remaining) - 2)) + 2)[2:]
else:
# truncate, we have no other addresses than 0 and 1
suffix = suffix[len(suffix) - remaining:]
suffix = suffix.zfill(remaining)
bits = prefix + suffix
return dict(addr=ipv6FromBin(bits), prefixlen=prefixlen)
def getIpv6RangeFactory(k, s):
"""
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7