hhucn · osarier · May 5, 2014 · Jun 5, 2014 · Jun 8, 2014 · Jun 8, 2014
diff --git a/.gitignore b/.gitignore
@@ -1,4 +1,5 @@
 *.py[cod]
+*~
 
 # C extensions
 *.so

diff --git a/test/test_billion_laughs.py b/test/test_billion_laughs.py
@@ -0,0 +1,96 @@
+from __future__ import unicode_literals
+import unittest
+import re
+import xml_parser
+import tutil
+import webvulnscan.attacks.billion_laughs
+import socket
+import time
+
+
+try:
+    from urllib.request import Request, URLError, HTTPError, urlopen
+except ImportError:
+    from urllib2 import Request, URLError, HTTPError, urlopen
+
+
+def get_XML(url):
+    if "http://" not in url:
+        url = "http://" + url
+    headers = {'Accept': 'application/xrds+xml'}
+    req = Request(url, None, headers)
+
+    try:
+        response = urlopen(req)
+    except (HTTPError, URLError):
+        print(
+            'There was a problem with your request!\n')
+        return None
+
+    html = response.read()
+    xrds_loc = re.search(
+        r'<meta\s+http-equiv="x-xrds-Location"\s+content="(?P<value>[^"]*)"\s*', html.decode(), re.IGNORECASE)
+
+    if not xrds_loc:
+        return None
+
+    xrds_url = xrds_loc.group('value')
+
+    try:
+        xrds_doc = urlopen(xrds_url)
+    except (HTTPError, URLError):
+        print(
+            'There was a problem with your request!\n')
+        return None
+
+    return xrds_doc.read()
+
+
+def form_client(vulnerable=False):
+    form = u'''<html><form action="./send" method="post">
+                    <input name="openid" type="text">
+                    <input type="submit" value="submit" >
+                </form></html>'''
+
+    def result(request):
+        if 'openid' not in request.parameters:
+            return u'<html>There is no input!</html>'
+
+        xml = get_XML(request.parameters['openid'])
+
+        if xml is None:
+            return u'<html>XML document could not be found!</html>'
+
+        res = xml_parser.get_xml_length(xml)
+
+        timeout = getattr(request, 'timeout', None)
+        if vulnerable:
+            if not timeout:
+                time.sleep(res / 100000)
+            elif timeout < res:
+                raise socket.timeout()
+
+        return u'<html>%s</html>' % res
+
+    return {
+        '/': form,
+        '/send': result,
+    }
+
+
+class billion_laughs(unittest.TestCase):
+    attack = webvulnscan.attacks.billion_laughs
+
+    @tutil.webtest(False)
+    def test_billion_laughs_static_site():
+        return {
+            '/': u'''<html></html>''',
+        }
+
+    @tutil.webtest(False)
+    def test_billion_laughs_form():
+        return form_client()
+
+    @tutil.webtest(True)
+    def test_billion_laughs_dangerous():
+        return form_client(True)
diff --git a/test/test_utf7_check.py b/test/test_utf7_check.py
@@ -0,0 +1,20 @@
+import unittest
+import tutil
+import webvulnscan.attacks.utf7_check
+
+
+class UTF7_checkTest(unittest.TestCase):
+    attack = webvulnscan.attacks.utf7_check
+
+    @tutil.webtest(False)
+    def test_utf7_check_static_site():
+        return{
+            '/': u'''<html></html>''',
+        }
+
+    @tutil.webtest(True)
+    def test_utf7_check_dangerous():
+        return{'/': (
+            200, b'''<html></html>''',
+            {'Content-Type': 'text/html; charset=UTF-7'}),
+        }
diff --git a/test/xml_parser.py b/test/xml_parser.py
@@ -0,0 +1,192 @@
+import unittest
+import re
+
+
+def get_objects(xml_data, allow_entity_def=True):
+    idx = 0
+    while idx < len(xml_data):
+        c = xml_data[idx]
+        if c == '<' and allow_entity_def:
+            m = re.match(r'''(?x)
+                !ENTITY\s+
+                (?P<name>[a-zA-Z0-9_-]+)\s+
+                "(?P<value>[^"]*)"\s*
+                >
+            ''', xml_data[idx + 1:].decode())
+            if m:
+                value = get_objects(m.group('value'), allow_entity_def=False)
+                yield ('entity', (m.group('name'), value))
+                idx += len(m.group(0)) + 1
+                continue
+
+        if c == '&':
+            endpos = xml_data.find(';', idx + 1)
+            yield ('entity_reference', xml_data[idx + 1:endpos])
+
+            idx = endpos
+
+        else:
+            yield ('text', c)
+
+        idx += 1
+
+
+def get_xml_length(xml_data):
+    object_stream = get_objects(xml_data)
+    return _calc_xml_length(object_stream, {})
+
+
+def _calc_xml_length(object_stream, entities):
+    res = 0
+    for otype, ovalue in object_stream:
+        if otype == 'entity':
+            ename, evalue = ovalue
+            entities[ename] = _calc_xml_length(evalue, entities)
+        elif otype == 'entity_reference':
+            res += entities[ovalue]
+        else:  # text
+            res += len(ovalue)
+    return res
+
+
+def _make_list(gen):
+    gen = list(gen)
+    return [
+        (key, (val[0], _make_list(val[1])))
+        if key == 'entity'
+        else (key, val)
+        for key, val in gen]
+
+if __name__ == '__main__':
+    unittest.main()
+
+
+class XMLParser_test(unittest.TestCase):
+
+    def test_entity_parsing(self):
+        self.assertEqual(
+            _make_list(get_objects(
+                '''<!ENTITY x "y" >''')),
+            [('entity', ('x', [('text', 'y')]))]
+        )
+
+        self.assertEqual(
+            _make_list(get_objects(
+                '''&y;x''')),
+            [('entity_reference', 'y'), ('text', 'x')]
+        )
+
+        self.assertEqual(
+            _make_list(get_objects(
+                '''&ya;x''')),
+            [('entity_reference', 'ya'), ('text', 'x')]
+        )
+
+    def test_empty(self):
+        xml_str = ''''''
+        res = get_xml_length(xml_str)
+        self.assertEqual(res, 0)
+
+    def test_invalid_reference(self):
+        xml_str = '&a;'
+        self.assertRaises(BaseException, get_xml_length, xml_str)
+
+    def test_lol1(self):
+        xml_str = '''
+<!ENTITY lol "lol">
+<!ENTITY lol1 "&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;">
+<test>&lol1;</test>
+'''
+        res = get_xml_length(xml_str)
+        self.assertTrue(res >= 30)
+
+    def test_no_reference(self):
+        xml_str = '''
+<!ENTITY lol "lol">
+<!ENTITY lol1 "&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;">
+<test></test>
+'''
+        res = get_xml_length(xml_str)
+        self.assertTrue(res < 30)
+
+    def test_lol1_multiple(self):
+        xml_str = '''
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE lolz [
+  <!ENTITY lol "lol">
+  <!ENTITY lol1 "&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;">
+]>
+<test>&lol1;&lol1;&lol1;</test>
+'''
+        res = get_xml_length(xml_str)
+        self.assertTrue(res >= 90)
+
+    def test_lol1_missing_semicolon(self):
+        xml_str = '''
+<!ENTITY lol "lol">
+<!ENTITY lol1 "&lol&lol&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;">
+<test>&lol1;</test>
+'''
+        self.assertRaises(BaseException, get_xml_length, xml_str)
+
+    def test_quadratic_blowup(self):
+        xml_str = '''
+<?xml version="1.0"?>
+<!DOCTYPE payload [
+<!ENTITY A "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA">
+]>
+<payload>&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;&A;</payload>
+'''
+        res = get_xml_length(xml_str)
+        self.assertTrue(res >= 2000)
+
+    def test_missing_bracket1(self):
+        xml_str = '''
+<?xml version="1.0"?>
+<!DOCTYPE payload [
+<!ENTITY A "AAAAAAAAAAAAAAAAAAAAAAA"
+]>
+<payload>&A;</payload>
+'''
+        self.assertRaises(BaseException, get_xml_length, xml_str)
+
+    def test_missing_bracket2(self):
+        xml_str = '''
+<?xml version="1.0"?>
+<!DOCTYPE payload [
+!ENTITY A "AAAAAAAAAAAAAAAAAAAAAAA">
+]>
+<payload>&A;</payload>
+'''
+        self.assertRaises(BaseException, get_xml_length, xml_str)
+
+    def test_missing_semicolon(self):
+        xml_str = '''
+<?xml version="1.0"?>
+<!DOCTYPE payload [
+<!ENTITY A "AAAAAAAAAAAAAAAAAAAAAAA">
+]>
+<payload>&A</payload>
+'''
+        self.assertRaises(BaseException, get_xml_length, xml_str)
+
+    def test_missing_slash(self):
+        xml_str = '''
+<?xml version="1.0"?>
+<!DOCTYPE payload [
+<!ENTITY A "AAAAAAAAAAAAAAAAAAAAAAA">
+]>
+<payload>&A;<payload>
+'''
+        res = get_xml_length(xml_str)
+        self.assertTrue(res >= 23)
+
+    def test_missing_quotes(self):
+        xml_str = '''
+<?xml version="1.0"?>
+<!DOCTYPE payload [
+<!ENTITY A AAAAAAAAAAAAAAAAAAAAAAA>
+]>
+<payload>&A;</payload>
+'''
+        self.assertRaises(BaseException, get_xml_length, xml_str)
diff --git a/webvulnscan/__init__.py b/webvulnscan/__init__.py
@@ -37,7 +37,7 @@ def run(options, targets):
         log = Log(verbosity=u'vuln')
     else:
         log = Log()
-    client = Client(log=log)
+    client = Client(log=log, config=options.bl_config)
 
     if options.import_cookies:
         client.cookie_jar = MozillaCookieJar(options.import_cookies)

diff --git a/webvulnscan/attacks/__init__.py b/webvulnscan/attacks/__init__.py
@@ -6,7 +6,9 @@
 from .clickjack import clickjack
 from .cookiescan import cookiescan
 from .exotic_characters import exotic_characters
+from .utf7_check import utf7_check
+from .billion_laughs import billion_laughs
 
 
 def all_attacks():
-    return [xss, csrf, crlf, breach, clickjack, cookiescan, exotic_characters]
+    return [xss, csrf, crlf, breach, clickjack, cookiescan, exotic_characters, utf7_check, billion_laughs]
diff --git a/webvulnscan/attacks/billion_laughs.py b/webvulnscan/attacks/billion_laughs.py
@@ -0,0 +1,39 @@
+from ..utils import attack
+from ..openID_test_server import Create_server
+import socket
+
+
+def search(page):
+    for form in page.get_forms():
+        yield (form,)
+
+
+@attack(search)
+def billion_laughs(client, log, form):
+    def attack(attack_url):
+        parameters = dict(form.get_parameters(True))
+        for key in parameters:
+            if 'openid' in key.lower():
+                parameters[key] = attack_url
+
+        try:
+            # Necessary because submit buttons usually do not have names which
+            # can cause bugs. So the submit buttons are deleted from the
+            # parameter list.
+            if '' in parameters:
+                del parameters['']
+            form.send(client, parameters, timeout=1)
+        except socket.timeout:
+            return True
+        return False
+
+    with Create_server(client.config) as openid_server:
+
+        if attack(openid_server.benign_url):
+            log('warn', openid_server.benign_url, "Billion Laughs",
+                "Test did not work!")
+            return
+
+        for evil_url in openid_server.evil_urls:
+            if attack(evil_url):
+                log('vuln', evil_url, "Billion Laughs/Quadratic Blowup")
diff --git a/webvulnscan/attacks/utf7_check.py b/webvulnscan/attacks/utf7_check.py
@@ -0,0 +1,22 @@
+from ..utils import attack
+
+
+@attack()
+def utf7_check(client, log, page):
+    if 'Content-Type' not in page.headers:
+        return
+
+    content_type = page.headers['Content-Type']
+
+    if ';' not in content_type:
+        return
+
+    # Use Content-Type to get the charset
+    charset = content_type.split(';')[1].split('=')[1]
+    # If charset is not UTF-7 there is nothing to do
+    if charset != 'utf-7' and charset != 'UTF-7':
+        return
+    else:
+        # If charset is UTF-7 give warning
+        log('vuln', page.url, 'UTF-7',
+            'Website uses UTF-7')