connect to ip option

docs/user/advanced.rst

@@ -284,6 +284,16 @@ You can combine a recipe with additional command line options, for example::
 
 In case of repeated options, command line options have precedence over options included in the recipe.
 
+Connect to an specific host
+---------------------------------------
+
+The --ip option can be used to connect to a specific host and port instead of the URL's host and port::
+
+    wfuzz -z range,1-1 --ip 127.0.0.1 http://www.google.com/anything/FUZZ
+
+This useful, for example, to test if a reverse proxy can be manipulated into misrouting requests to a destination of our choice.
+
+
 Scan Mode: Ignore Errors and Exceptions
 ---------------------------------------
 

src/wfuzz/fuzzobjects.py

@@ -3,6 +3,7 @@
 import re
 import itertools
 import operator
+import pycurl
 
 # Python 2 and 3
 import sys
@@ -126,6 +127,7 @@ def __init__(self):
         self._allvars = None
         self.wf_fuzz_methods = None
         self.wf_retries = 0
+        self.wf_ip = None
 
         self.headers.request = {"User-Agent": Facade().sett.get("connection", "user-agent")}
 
@@ -294,7 +296,12 @@ def perform(self):
         return Facade().http_pool.perform(res)
 
     def to_http_object(self, c):
-        return Request.to_pycurl_object(c, self._request)
+        pycurl_c = Request.to_pycurl_object(c, self._request)
+
+        if self.wf_ip:
+            pycurl_c.setopt(pycurl.CONNECT_TO, ["::{}:{}".format(self.wf_ip['ip'], self.wf_ip['port'])])
+
+        return pycurl_c
 
     def from_http_object(self, c, bh, bb):
         return self._request.response_from_conn_object(c, bh, bb)
@@ -337,6 +344,9 @@ def update_from_options(self, options):
         if options['postdata'] is not None:
             self.params.post = options['postdata']
 
+        if options['connect_to_ip']:
+            self.wf_ip = options['connect_to_ip']
+
         if options['method']:
             self.method = options['method']
             self.wf_fuzz_methods = options['method']
@@ -355,6 +365,7 @@ def from_copy(self):
         newreq.wf_proxy = self.wf_proxy
         newreq.wf_allvars = self.wf_allvars
         newreq.wf_fuzz_methods = self.wf_fuzz_methods
+        newreq.wf_ip = self.wf_ip
 
         newreq.headers.request = self.headers.request
         newreq.params.post = self.params.post

src/wfuzz/options.py

@@ -84,6 +84,7 @@ def _defaults(self):
             allvars=None,
             script="",
             script_args={},
+            connect_to_ip=None,
 
             # this is equivalent to payloads but in a different format
             dictio=None,

src/wfuzz/ui/console/clparser.py

@@ -161,7 +161,7 @@ def show_plugin_ext_help(self, registrant, category="$all$"):
     def parse_cl(self):
         # Usage and command line help
         try:
-            opts, args = getopt.getopt(self.argv[1:], "hLAZX:vcb:e:R:d:z:r:f:t:w:V:H:m:f:o:s:p:w:u:", ['filter-help', 'AAA', 'AA', 'slice=', 'zP=', 'oF=', 'recipe=', 'dump-recipe=', 'req-delay=', 'conn-delay=', 'sc=', 'sh=', 'sl=', 'sw=', 'ss=', 'hc=', 'hh=', 'hl=', 'hw=', 'hs=', 'ntlm=', 'basic=', 'digest=', 'follow', 'script-help=', 'script=', 'script-args=', 'prefilter=', 'filter=', 'interact', 'help', 'version', 'dry-run', 'prev'])
+            opts, args = getopt.getopt(self.argv[1:], "hLAZX:vcb:e:R:d:z:r:f:t:w:V:H:m:f:o:s:p:w:u:", ['ip=', 'filter-help', 'AAA', 'AA', 'slice=', 'zP=', 'oF=', 'recipe=', 'dump-recipe=', 'req-delay=', 'conn-delay=', 'sc=', 'sh=', 'sl=', 'sw=', 'ss=', 'hc=', 'hh=', 'hl=', 'hw=', 'hs=', 'ntlm=', 'basic=', 'digest=', 'follow', 'script-help=', 'script=', 'script-args=', 'prefilter=', 'filter=', 'interact', 'help', 'version', 'dry-run', 'prev'])
             optsd = defaultdict(list)
 
             payload_cache = {}
@@ -437,6 +437,16 @@ def _parse_seed(self, url, optsd, options):
         if "--follow" in optsd or "-L" in optsd:
             options['follow'] = True
 
+        if "--ip" in optsd:
+            splitted = optsd["--ip"][0].partition(":")
+            if not splitted[0]:
+                raise FuzzExceptBadOptions("An IP must be specified")
+
+            options["connect_to_ip"] = {
+                "ip": splitted[0],
+                "port": splitted[2] if splitted[2] else "80"
+            }
+
         if "-d" in optsd:
             options['postdata'] = optsd["-d"][0]
 

src/wfuzz/ui/console/common.py

@@ -103,6 +103,7 @@
 \t-s N                      : Specify time delay between requests (0 default)
 \t-R depth                  : Recursive path discovery being depth the maximum recursion level.
 \t-L,--follow               : Follow HTTP redirections
+\t--ip host:port            : Specify an IP to connect to instead of the URL's host in the format ip:port
 \t-Z                        : Scan mode (Connection errors will be ignored).
 \t--req-delay N             : Sets the maximum time in seconds the request is allowed to take (CURLOPT_TIMEOUT). Default 90.
 \t--conn-delay N            : Sets the maximum time in seconds the connection phase to the server to take (CURLOPT_CONNECTTIMEOUT). Default 90.

tests/test_acceptance.py

@@ -1,6 +1,7 @@
 #!/usr/bin/python
 # -*- coding: utf-8 -*-
 
+import copy
 import os
 import unittest
 import tempfile
@@ -42,6 +43,7 @@
 ]
 
 testing_tests = [
+    ("test_url_all_url_fuzz2", "FUZZ", [["http://webscantest.com/datastore/search_get_by_name.php?name=Rake"]], dict(), [(200, '/datastore/search_get_by_name.php')], None),
 ]
 
 savedsession_tests = [
@@ -82,6 +84,9 @@
 ]
 
 basic_tests = [
+    # different connect host ip
+    ("test_static_strquery_set_ip", "http://wfuzz.org/FUZZ?var=1&var2=2", [["anything"], ['PUT', 'GET', 'DELETE']], dict(connect_to_ip={'ip': '127.0.0.1', 'port': '9000'}, method='FUZ2Z', filter="content~'url' and content~'http://wfuzz.org'"), [(200, '/anything')] * 3, None),
+
     # encoding tests
     ("test_encode_cookie2_utf8_return", "%s/anything" % HTTPBIN_URL, [["は国"]], dict(cookie=["test=FUZZ"], filter="content~'test=\\\\u00e3\\\\u0081\\\\u00af\\\\u00e5\\\\u009b\\\\u00bd'"), [(200, '/anything')], None),
     ("test_encode_header_utf8_return", "%s/headers" % HTTPBIN_URL, [["は国"]], dict(headers=[("myheader", "FUZZ")], filter="content~'Myheader' and content~'\\\\u00e3\\\\u0081\\\\u00af\\\\u00e5\\\\u009b\\\\u00bd'"), [(200, '/headers')], None),
@@ -263,6 +268,10 @@ def test(self):
                     if proxied_payloads:
                         proxied_payloads = [[payload.replace(original_host, proxied_host) for payload in payloads_list] for payloads_list in proxied_payloads]
 
+                if 'connect_to_ip' in extra_params and extra_params['connect_to_ip']:
+                    extra_params['connect_to_ip']['ip'] = 'httpbin'
+                    extra_params['connect_to_ip']['port'] = '80'
+
             with wfuzz.FuzzSession(url=proxied_url) as s:
                 same_list = [(x.code, x.history.urlparse.path) for x in s.get_payloads(proxied_payloads).fuzz(**extra_params)]
 
@@ -393,12 +402,13 @@ def duplicate_tests_diff_params(test_list, group, next_extra_params, previous_ex
         if group == "_proxy_" and "encode" in test_name:
             continue
 
-        next_extra = dict(list(params.items()) + list(next_extra_params.items()))
+        next_extra = copy.deepcopy(params)
+        next_extra.update(next_extra_params)
         new_test = "%s_%s" % (test_name, group)
 
-        prev_extra = params
+        prev_extra = copy.deepcopy(params)
         if previous_extra_params:
-            prev_extra = dict(list(params.items()) + list(previous_extra_params.items()))
+            prev_extra.update(previous_extra_params)
 
         create_test(new_test, url, payloads, prev_extra, None, next_extra, exception_str)
 

tests/test_clparser.py

@@ -9,3 +9,23 @@ def test_listplugins(self):
             CLParser(['wfuzz', '-e', 'iterators']).parse_cl()
 
         self.assertEqual(cm.exception.code, 0)
+
+    def test_ip_option(self):
+        options = CLParser(['wfuzz', '--ip', '127.0.0.1']).parse_cl()
+
+        self.assertEqual(options.data['connect_to_ip']['ip'], '127.0.0.1')
+        self.assertEqual(options.data['connect_to_ip']['port'], '80')
+
+        options = CLParser(['wfuzz', '--ip', '127.0.0.1:22']).parse_cl()
+
+        self.assertEqual(options.data['connect_to_ip']['ip'], '127.0.0.1')
+        self.assertEqual(options.data['connect_to_ip']['port'], '22')
+
+        options = CLParser(['wfuzz', '--ip', '127.0.0.1:']).parse_cl()
+
+        self.assertEqual(options.data['connect_to_ip']['ip'], '127.0.0.1')
+        self.assertEqual(options.data['connect_to_ip']['port'], '80')
+
+        with self.assertRaises(Exception) as cm:
+            options = CLParser(['wfuzz', '--ip', ':80']).parse_cl()
+        self.assertTrue("An IP must be specified" in str(cm.exception))