177 lines
4.8 KiB
Python
177 lines
4.8 KiB
Python
# Copyright 2010 United States Government as represented by the
|
|
# Administrator of the National Aeronautics and Space Administration.
|
|
# Copyright 2011 Justin Santa Barbara
|
|
# Copyright (c) 2012 NTT DOCOMO, INC.
|
|
# All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
"""Utilities and helper functions."""
|
|
|
|
|
|
import netaddr
|
|
import re
|
|
import six
|
|
import uuid
|
|
|
|
from oslo.config import cfg
|
|
|
|
from cerberus.common import exception
|
|
from cerberus.openstack.common.gettextutils import _ # noqa
|
|
from cerberus.openstack.common import log as logging
|
|
|
|
CONF = cfg.CONF
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
class LazyPluggable(object):
|
|
"""A pluggable backend loaded lazily based on some value."""
|
|
|
|
def __init__(self, pivot, config_group=None, **backends):
|
|
self.__backends = backends
|
|
self.__pivot = pivot
|
|
self.__backend = None
|
|
self.__config_group = config_group
|
|
|
|
def __get_backend(self):
|
|
if not self.__backend:
|
|
if self.__config_group is None:
|
|
backend_name = CONF[self.__pivot]
|
|
else:
|
|
backend_name = CONF[self.__config_group][self.__pivot]
|
|
if backend_name not in self.__backends:
|
|
msg = _('Invalid backend: %s') % backend_name
|
|
raise exception.CerberusException(msg)
|
|
|
|
backend = self.__backends[backend_name]
|
|
if isinstance(backend, tuple):
|
|
name = backend[0]
|
|
fromlist = backend[1]
|
|
else:
|
|
name = backend
|
|
fromlist = backend
|
|
|
|
self.__backend = __import__(name, None, None, fromlist)
|
|
return self.__backend
|
|
|
|
def __getattr__(self, key):
|
|
backend = self.__get_backend()
|
|
return getattr(backend, key)
|
|
|
|
|
|
def is_valid_ipv4(address):
|
|
"""Verify that address represents a valid IPv4 address."""
|
|
try:
|
|
return netaddr.valid_ipv4(address)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def is_valid_ipv6(address):
|
|
try:
|
|
return netaddr.valid_ipv6(address)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def is_valid_ipv6_cidr(address):
|
|
try:
|
|
str(netaddr.IPNetwork(address, version=6).cidr)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def get_shortened_ipv6(address):
|
|
addr = netaddr.IPAddress(address, version=6)
|
|
return str(addr.ipv6())
|
|
|
|
|
|
def get_shortened_ipv6_cidr(address):
|
|
net = netaddr.IPNetwork(address, version=6)
|
|
return str(net.cidr)
|
|
|
|
|
|
def is_valid_cidr(address):
|
|
"""Check if the provided ipv4 or ipv6 address is a valid CIDR address."""
|
|
try:
|
|
# Validate the correct CIDR Address
|
|
netaddr.IPNetwork(address)
|
|
except netaddr.core.AddrFormatError:
|
|
return False
|
|
except UnboundLocalError:
|
|
# NOTE(MotoKen): work around bug in netaddr 0.7.5 (see detail in
|
|
# https://github.com/drkjam/netaddr/issues/2)
|
|
return False
|
|
|
|
# Prior validation partially verify /xx part
|
|
# Verify it here
|
|
ip_segment = address.split('/')
|
|
|
|
if (len(ip_segment) <= 1 or
|
|
ip_segment[1] == ''):
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def get_ip_version(network):
|
|
"""Returns the IP version of a network (IPv4 or IPv6).
|
|
|
|
:raises: AddrFormatError if invalid network.
|
|
"""
|
|
if netaddr.IPNetwork(network).version == 6:
|
|
return "IPv6"
|
|
elif netaddr.IPNetwork(network).version == 4:
|
|
return "IPv4"
|
|
|
|
|
|
def convert_to_list_dict(lst, label):
|
|
"""Convert a value or list into a list of dicts."""
|
|
if not lst:
|
|
return None
|
|
if not isinstance(lst, list):
|
|
lst = [lst]
|
|
return [{label: x} for x in lst]
|
|
|
|
|
|
def sanitize_hostname(hostname):
|
|
"""Return a hostname which conforms to RFC-952 and RFC-1123 specs."""
|
|
if isinstance(hostname, six.text_type):
|
|
hostname = hostname.encode('latin-1', 'ignore')
|
|
|
|
hostname = re.sub('[ _]', '-', hostname)
|
|
hostname = re.sub('[^\w.-]+', '', hostname)
|
|
hostname = hostname.lower()
|
|
hostname = hostname.strip('.-')
|
|
|
|
return hostname
|
|
|
|
|
|
def generate_uuid():
|
|
return str(uuid.uuid4())
|
|
|
|
|
|
def is_uuid_like(val):
|
|
"""Returns validation of a value as a UUID.
|
|
|
|
For our purposes, a UUID is a canonical form string:
|
|
aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa
|
|
|
|
"""
|
|
try:
|
|
return str(uuid.UUID(val)) == val
|
|
except (TypeError, ValueError, AttributeError):
|
|
return False
|