neutron/ipam/utils.py

   1 # Copyright 2015 OpenStack LLC.
   2 # All Rights Reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15
  16 import netaddr
  17
  18
  19 def check_subnet_ip(cidr, ip_address):
  20     """Validate that the IP address is on the subnet."""
  21     ip = netaddr.IPAddress(ip_address)
  22     net = netaddr.IPNetwork(cidr)
  23     # Check that the IP is valid on subnet. This cannot be the
  24     # network or the broadcast address (which exists only in IPv4)
  25     return (ip != net.network
  26             and (net.version == 6 or ip != net[-1])
  27             and net.netmask & ip == net.network)
  28
  29
  30 def check_gateway_in_subnet(cidr, gateway):
  31     """Validate that the gateway is on the subnet."""
  32     ip = netaddr.IPAddress(gateway)
  33     if ip.version == 4 or (ip.version == 6 and not ip.is_link_local()):
  34         return check_subnet_ip(cidr, gateway)
  35     return True
  36
  37
  38 def generate_pools(cidr, gateway_ip):
  39     """Create IP allocation pools for a specified subnet
  40
  41     The Neutron API defines a subnet's allocation pools as a list of
  42     IPRange objects for defining the pool range.
  43     """
  44     # Auto allocate the pool around gateway_ip
  45     net = netaddr.IPNetwork(cidr)
  46     ip_version = net.version
  47     first = netaddr.IPAddress(net.first, ip_version)
  48     last = netaddr.IPAddress(net.last, ip_version)
  49     if first == last:
  50         # handle single address subnet case
  51         return [netaddr.IPRange(first, last)]
  52     first_ip = first + 1
  53     # last address is broadcast in v4
  54     last_ip = last - (ip_version == 4)
  55     if first_ip >= last_ip:
  56         # /31 lands here
  57         return []
  58     ipset = netaddr.IPSet(netaddr.IPRange(first_ip, last_ip))
  59     if gateway_ip:
  60         ipset.remove(netaddr.IPAddress(gateway_ip, ip_version))
  61     return list(ipset.iter_ipranges())