Ticket #1713: match-ipv6.patch

turbogears/identity/conditions.py

@@ -1 +1 @@
-import cherrypy
-import socket
-import struct
-import threading
-import types
-import turbogears
-from turbogears.identity.exceptions import *
-from turbogears.identity import *
-from turbogears.decorator import weak_signature_decorator
+from turbogears.util import match_ip
-
+
-class Predicate(object):
-    '''
-    Generic base class for testing true or false for a condition.
@@ -177 +176 @@
-        return ""
-
-
-def _match_ip(cidr, ip):
-    if not '/' in cidr:
-        return cidr == ip
-
-    else:
-        try:
-            b,m = cidr.split('/')
-            shift = 32 - int(m)
-            a1 = struct.unpack('!L', socket.inet_aton(b))[0] >> shift
-            a2 = struct.unpack('!L', socket.inet_aton(ip))[0] >> shift
-            return a1 == a2
-
-        except:
-            return False
-
-
-class from_host(Predicate, IdentityPredicateHelper):
-    '''
-    Predicate for checking whether the visitor's host is an allowed host.
@@ -209 +192 @@
-        Match the visitor's host against the criteria.
-        '''
-        ip = _remoteHost()
-_
+
-            return True
-
-        self.append_error_message(errors)

turbogears/tests/test_util.py

@@ -90 +90 @@
-        == 'Chip & Chap'
-    assert util.fixentities('<&quot;&lt;&copy;&gt;&quot;>') \
-        == '<&quot;&lt;&#169;&gt;&quot;>'
+
+
+def test_match_ip():
+    m = util.match_ip
+    assert m('1.2.3.4', '1.2.3.4')
+    assert not m('1.2.3.4', '1.2.3.5')
+    assert not m('4.3.2.1', '5.3.2.1')
+    assert m('1.2.3.4', '::ffff:102:304')
+    assert not m('1.2.3.4', '::102:304')
+    assert m('127.0.0.1', '::ffff:127.0.0.1')
+    assert m('127.0.0.1', '::ffff:7f00:0001')
+    assert m('::ffff:127.0.0.1', '127.0.0.1')
+    assert m('192.168.42.76/32', '192.168.42.76')
+    assert m('192.168.42.77/32', '192.168.42.77')
+    assert not m('192.168.42.76/32', '192.168.42.77')
+    assert not m('192.168.42.77/32', '192.168.42.76')
+    assert m('192.168.42.76/31', '192.168.42.76')
+    assert m('192.168.42.76/31', '192.168.42.77')
+    assert m('224.0.0.0/3', '224.1.2.3')
+    assert m('224.0.0.0/3', '255.255.255.255')
+    assert not m('224.0.0.0/3', '192.0.0.0')
+    assert not m('192.168.42.76/31', '192.168.42.73')
+    assert m('1:2:3::', '1:02:0003:0:0::0')
+    assert m('1:2:3::/48', '1:2:3:4::')
+    assert not m('1:2:3::/48', '1:2:4:::')
+    assert m('1:2:3::8', '1:2:3:0:0:0:0:8')
+    assert m('2001:0db8:85a3:08d3:1319:8a2e:0370:7334',
+        '2001:db8:85a3:8d3:1319:8a2e:3.112.115.52')
+    assert m('2001:db8:85a3:8d3:1300::/72',
+        '2001:0db8:85a3:08d3:1319:8a2e:0370:7334')
+    assert m('2001:db8:85a3:8d3:1300::/72',
+        '2001:0db8:85a3:08d3:1399:8a2e:0370:7334')
+    assert not m('2001:db8:85a3:8d3:1300::/72',
+        '2001:0db8:85a3:08d3:1219:8a2e:0370:7334')

turbogears/util.py

@@ -6 +6 @@
-import logging
-import warnings
-import htmlentitydefs
+import socket
+import struct
-from inspect import getargspec, getargvalues
-from itertools import izip, islice, chain, imap
-from operator import isSequenceType
@@ -539 +541 @@
-    return re.sub("&(\w+);?", repl, htmltext)
-
-
+if hasattr(socket, 'inet_pton') and hasattr(socket, 'AF_INET6'):
+
+    def inet6_aton(addr):
+        """Convert IP6 standard hex notation to IP6 address."""
+        return socket.inet_pton(socket.AF_INET6, addr)
+
+else: # Windows etc.
+
+    import string
+    _inet6_chars = string.hexdigits + ':.'
+
+    def inet6_aton(addr):
+        """Convert IPv6 standard hex notation to IPv6 address.
+
+        Inspired by http://twistedmatrix.com/trac/.
+
+        """
+        faulty = addr.lstrip(_inet6_chars)
+        if faulty:
+            raise ValueError("Illegal character '%c' in IPv6 address" % faulty[0])
+        parts = addr.split(':')
+        elided = parts.count('')
+        extenso = '.' in parts[-1] and 7 or 8
+        if len(parts) > extenso or elided > 3:
+            raise ValueError("Syntactically invalid IPv6 address")
+        if elided == 3:
+            return '\x00' * 16
+        if elided:
+            zeros = ['0'] * (extenso - len(parts) + elided)
+            if addr.startswith('::'):
+                parts[:2] = zeros
+            elif addr.endswith('::'):
+                parts[-2:] = zeros
+            else:
+                idx = parts.index('')
+                parts[idx:idx+1] = zeros
+        if len(parts) != extenso:
+            raise ValueError("Syntactically invalid IPv6 address")
+        if extenso == 7:
+            ipv4 = parts.pop()
+            if ipv4.count('.') != 3:
+                raise ValueError("Syntactically invalid IPv6 address")
+            parts = [int(x, 16) for x in parts]
+            return struct.pack('!6H', *parts) + socket.inet_aton(ipv4)
+        else:
+            parts = [int(x, 16) for x in parts]
+            return struct.pack('!8H', *parts)
+
+
+def inet_aton(addr):
+    """Convert IPv4 or IPv6 notation to IPv6 address."""
+    if ':' in addr:
+        return inet6_aton(addr)
+    else:
+        return struct.pack('!QL', 0, 0xffff) + socket.inet_aton(addr)
+
+
+def _inet_prefix(addr, masked):
+    """Remove the number of masked bits from the IPV6 address."""
+    hi, lo = struct.unpack("!QQ", addr)
+    return (hi << 64 | lo) >> masked
+
+
+def match_ip(cidr, ip):
+    """Check whether IP address matches CIDR IP address block."""
+    if '/' in cidr:
+        cidr, prefix = cidr.split('/', 1)
+        masked = (':' in cidr and 128 or 32) - int(prefix)
+    else:
+        masked = None
+    cidr = inet_aton(cidr)
+    ip = inet_aton(ip)
+    if masked:
+        cidr = _inet_prefix(cidr, masked)
+        ip = _inet_prefix(ip, masked)
+    return ip == cidr
+
+
-__all__ = ["Bunch", "DictObj", "DictWrapper", "Enum", "setlike",
-           "get_package_name", "get_model", "load_project_config",
-           "ensure_sequence", "has_arg", "to_kw", "from_kw",
@@ -548 +628 @@
-           "flatten_sequence", "load_class", "parse_http_accept_header",
-           "to_unicode", "to_utf8", "quote_cookie", "unquote_cookie",
-           "get_template_encoding_default", "find_precision",
-
+match_ip", "