from __future__ import absolute_import
import re
+import ipaddress
from distutils.util import strtobool
+import six
from ..awsauth import S3Auth
from ..settings import Settings, Options
from ..rest_client import RestClient, RequestException
-from ..tools import build_url, dict_contains_path, is_valid_ip_address
+from ..tools import build_url, dict_contains_path
from .. import mgr, logger
# Group 1: [
# Group 2: 2001:db8:85a3::8a2e:370:7334
addr = match.group(3) if match.group(3) else match.group(2)
- if not is_valid_ip_address(addr):
+ try:
+ ipaddress.ip_address(six.u(addr))
+ return addr
+ except ValueError:
raise LookupError('Invalid RGW address \'{}\' found'.format(addr))
- return addr
raise LookupError('Failed to determine RGW address')
from ..services.exception import handle_rados_error
from ..controllers import RESTController, ApiController, Controller, \
BaseController, Proxy
-from ..tools import is_valid_ipv6_address, dict_contains_path, \
- RequestLoggingTool
+from ..tools import dict_contains_path, RequestLoggingTool
# pylint: disable=W0613
class TestFunctions(unittest.TestCase):
- def test_is_valid_ipv6_address(self):
- self.assertTrue(is_valid_ipv6_address('::'))
- self.assertTrue(is_valid_ipv6_address('::1'))
- self.assertFalse(is_valid_ipv6_address('127.0.0.1'))
- self.assertFalse(is_valid_ipv6_address('localhost'))
- self.assertTrue(is_valid_ipv6_address('1200:0000:AB00:1234:0000:2552:7777:1313'))
- self.assertFalse(is_valid_ipv6_address('1200::AB00:1234::2552:7777:1313'))
-
def test_dict_contains_path(self):
x = {'a': {'b': {'c': 'foo'}}}
self.assertTrue(dict_contains_path(x, ['a', 'b', 'c']))
import inspect
import json
import functools
+import ipaddress
import collections
from datetime import datetime, timedelta
import fnmatch
import time
import threading
-import socket
+import six
from six.moves import urllib
import cherrypy
self.lock.release()
-def is_valid_ip_address(addr):
- """
- Validate the given IPv4 or IPv6 address.
-
- >>> is_valid_ip_address('2001:0db8::1234')
- True
-
- >>> is_valid_ip_address('192.168.121.1')
- True
-
- >>> is_valid_ip_address('1:::1')
- False
-
- >>> is_valid_ip_address('8.1.0')
- False
-
- >>> is_valid_ip_address('260.1.0.1')
- False
-
- :param addr:
- :type addr: str
- :return: Returns ``True`` if the IP address is valid,
- otherwise ``False``.
- :rtype: bool
- """
- return is_valid_ipv4_address(addr) or is_valid_ipv6_address(addr)
-
-
-def is_valid_ipv4_address(addr):
- """
- Validate the given IPv4 address.
-
- >>> is_valid_ipv4_address('0.0.0.0')
- True
-
- >>> is_valid_ipv4_address('192.168.121.1')
- True
-
- >>> is_valid_ipv4_address('a.b.c.d')
- False
-
- >>> is_valid_ipv4_address('172.1.0.a')
- False
-
- >>> is_valid_ipv4_address('2001:0db8::1234')
- False
-
- >>> is_valid_ipv4_address(None)
- False
-
- >>> is_valid_ipv4_address(123456)
- False
-
- :param addr:
- :type addr: str
- :return: Returns ``True`` if the IPv4 address is valid,
- otherwise ``False``.
- :rtype: bool
- """
- try:
- socket.inet_pton(socket.AF_INET, addr)
- return True
- except (socket.error, TypeError):
- return False
-
-
-def is_valid_ipv6_address(addr):
- """
- Validate the given IPv6 address.
-
- >>> is_valid_ipv6_address('2001:0db8::1234')
- True
-
- >>> is_valid_ipv6_address('fe80::bc6c:66b0:5af8:f44')
- True
-
- >>> is_valid_ipv6_address('192.168.121.1')
- False
-
- >>> is_valid_ipv6_address('a:x::1')
- False
-
- >>> is_valid_ipv6_address('1200:0000:AB00:1234:O000:2552:7777:1313')
- False
-
- >>> is_valid_ipv6_address(None)
- False
-
- >>> is_valid_ipv6_address(123456)
- False
-
- :param addr:
- :type addr: str
- :return: Returns ``True`` if the IPv6 address is valid,
- otherwise ``False``.
- :rtype: bool
- """
- try:
- socket.inet_pton(socket.AF_INET6, addr)
- return True
- except (socket.error, TypeError):
- return False
-
-
def build_url(host, scheme=None, port=None):
"""
Build a valid URL. IPv6 addresses specified in host will be enclosed in brackets
:type port: int
:rtype: str
"""
- netloc = host if not is_valid_ipv6_address(host) else '[{}]'.format(host)
+ try:
+ ipaddress.IPv6Address(six.u(host))
+ netloc = '[{}]'.format(host)
+ except ValueError:
+ netloc = host
if port:
netloc += ':{}'.format(port)
pr = urllib.parse.ParseResult(