def test_do_req_through_proxy(self):
        resp_body = self.proxy_opener.open(get_moth_http()).read()

        # Basic check
        self.assertTrue(len(resp_body) > 0)

        # Get response using the proxy
        proxy_resp = self.proxy_opener.open(get_moth_http())
        # Get it without any proxy
        direct_resp = urllib2.urlopen(get_moth_http())

        # Must be equal
        self.assertEqual(direct_resp.read(), proxy_resp.read())

        # Have to remove the Date header because in some cases they differ
        # because one request was sent in second X and the other in X+1, which
        # makes the test fail
        direct_resp_headers = dict(direct_resp.info())
        proxy_resp_headers = dict(proxy_resp.info())

        # Make sure that a change in the seconds returned in date doesn't break
        # the test
        del direct_resp_headers['date']
        del proxy_resp_headers['date']

        del direct_resp_headers['transfer-encoding']
        del proxy_resp_headers['content-length']

        self.assertEqual(direct_resp_headers, proxy_resp_headers)

    def test_cache(self):
        url = URL(get_moth_http())
        http_response = self.uri_opener.GET(url)
        self.assertIn(self.MOTH_MESSAGE, http_response.body)

        url = URL(get_moth_http())
        http_response = self.uri_opener.GET(url)
        self.assertIn(self.MOTH_MESSAGE, http_response.body)

    def test_qs_params(self):
        url = URL(get_moth_http('/audit/xss/simple_xss.py?text=123456abc'))
        http_response = self.uri_opener.GET(url, cache=False)
        self.assertIn('123456abc', http_response.body)

        url = URL(get_moth_http('/audit/xss/simple_xss.py?text=root:x:0'))
        http_response = self.uri_opener.GET(url, cache=False)
        self.assertIn('root:x:0', http_response.body)

 def test_blacklist_handler_pass(self):
     """Verify that the blacklist handler works as expected"""
     opener = urllib2.build_opener(BlacklistHandler)
     
     request = urllib2.Request(get_moth_http())
     request.url_object = URL(get_moth_http())
     response = opener.open(request)
     
     self.assertEqual(response.code, 200)

 def test_history_access(self):
     self.count_plugin.loops = 1
     self.w3afcore.start()
     
     history_item = HistoryItem() 
     self.assertTrue(history_item.load(1))
     self.assertEqual(history_item.id, 1)
     self.assertEqual(history_item.get_request().get_uri().url_string,
                      get_moth_http())
     self.assertEqual(history_item.get_response().get_uri().url_string,
                      get_moth_http())

 def test_proxy_req_ok(self):
     """Test if self._proxy.stop() works as expected. Note that the check
     content is the same as the previous check, but it might be that this
     check fails because of some error in start() or stop() which is run
     during setUp and tearDown."""
     # Get response using the proxy
     proxy_resp = self.proxy_opener.open(get_moth_http()).read()
     # Get it the other way
     resp = urllib2.urlopen(get_moth_http()).read()
     # They must be very similar
     self.assertEqual(resp, proxy_resp)

 def test_request_trapped_send(self):
     def send_request(proxy_opener, result_queue):
         response = proxy_opener.open(get_moth_http())
         result_queue.put(response)
     
     self._proxy.set_trap(True)
     
     result_queue = Queue.Queue()
     send_thread = threading.Thread(target=send_request, args=(self.proxy_opener,
                                                               result_queue))
     send_thread.start()
     time.sleep(0.5)
     
     request = self._proxy.get_trapped_request()
     
     self.assertEqual(request.get_url().url_string, get_moth_http())
     self.assertEqual(request.get_method(), 'GET')
     
     self._proxy.send_raw_request(request, request.dump_request_head(),
                                  request.get_data())
     
     response = result_queue.get()
     
     self.assertEqual(response.code, 200)
     self.assertIn(self.MOTH_MESSAGE, response.read())

    def test_found_exploit_blind_sqli_form_GET(self):
        """
        Reproduce bug https://github.com/andresriancho/w3af/issues/262
        "it appears that you have provided tainted parameter values"
        """
        target = get_moth_http('/audit/blind_sqli/blind_where_integer_form_get.py')
        cfg = self._run_configs['blind_sqli']
        self._scan(target, cfg['plugins'])

        # Assert the general results
        vulns = self.kb.get('blind_sqli', 'blind_sqli')

        self.assertEquals(1, len(vulns))
        vuln = vulns[0]

        self.assertEquals("Blind SQL injection vulnerability", vuln.get_name())
        self.assertEquals('q', vuln.get_mutant().get_token_name())
        self.assertEquals('blind_where_integer_form_get.py',
                          vuln.get_url().get_file_name())

        vuln_to_exploit_id = vuln.get_id()

        #
        #   Execute the exploit.
        #
        plugin = self.w3afcore.plugins.get_plugin_inst('attack', 'sqlmap')

        #   Assert success
        self.assertTrue(plugin.can_exploit(vuln_to_exploit_id))
        exploit_result = plugin.exploit(vuln_to_exploit_id)
        self.assertEqual(len(exploit_result), 1, exploit_result)

def get_test_profile(profile=FAST_TEST_PROFILE):
    moth = get_moth_http('/')

    target_url = PROFILE_URL.replace('http://127.0.0.1:8000/', moth)
    profile = profile.replace('http://127.0.0.1:8000/', moth)

    return profile, target_url

 def send_request(proxy_opener, result_queue):
     try:
         proxy_opener.open(get_moth_http())
     except urllib2.HTTPError, he:
         # Catch the 403 from the local proxy when the user
         # drops the HTTP request.
         result_queue.put(he)

 def send(uri_opener, output):
     url = URL(get_moth_http())
     try:
         http_response = uri_opener.GET(url)
         output.put(http_response)
     except:
         output.put(None)

    def test_deflate(self):
        url = URL(get_moth_http('/core/deflate/deflate.html'))
        res = self.uri_opener.GET(url, cache=False)
        headers = res.get_headers()
        content_encoding, _ = headers.iget('content-encoding', '')

        self.assertIn('deflate', content_encoding)
        self.assertIn('View HTTP response headers.', res.get_body())

 def prepare_script(self):
     fhandler = tempfile.NamedTemporaryFile(prefix='spider_long-',
                                            suffix='.w3af',
                                            dir=tempfile.tempdir,
                                            delete=False)
     fhandler.write(file(self.SCRIPT).read() % {'moth': get_moth_http()})
     fhandler.close()
     return fhandler.name

 def test_basic(self):
     url = URL(get_moth_http())
     http_response = self.uri_opener.GET(url, cache=False)
     
     self.assertIn(self.MOTH_MESSAGE, http_response.body)
     
     self.assertGreaterEqual(http_response.id, 1)
     self.assertNotEqual(http_response.id, None)

    def test_post(self):
        url = URL(get_moth_http('/audit/xss/simple_xss_form.py'))

        data = URLEncodedForm()
        data['text'] = ['123456abc']

        http_response = self.uri_opener.POST(url, data, cache=False)
        self.assertIn('123456abc', http_response.body)

 def test_gzip(self):
     url = URL(get_moth_http('/core/gzip/gzip.html'))
     res = self.uri_opener.GET(url, cache=False)
     headers = res.get_headers()
     content_encoding, _ = headers.iget('content-encoding', '')
     test_res = 'gzip' in content_encoding or \
                'compress' in content_encoding
     self.assertTrue(test_res, content_encoding)

    def test_http_port_specification_via_proxy(self):
        self.assertEqual(self._proxy.total_handled_requests, 0)

        url = URL(get_moth_http())
        http_response = self.uri_opener.GET(url, cache=False)

        self.assertIn(self.MOTH_MESSAGE, http_response.body)
        self.assertEqual(self._proxy.total_handled_requests, 1)

 def test_get_wait_time(self):
     """
     Asserts that all the responses coming out of the extended urllib have a
     get_wait_time different from the default.
     """
     url = URL(get_moth_http())
     http_response = self.uri_opener.GET(url, cache=False)
     self.assertNotEqual(http_response.get_wait_time(), DEFAULT_WAIT_TIME)

    def test_post_special_chars(self):
        url = URL(get_moth_http('/audit/xss/simple_xss_form.py'))
        test_data = u'abc<def>"-á-'

        data = URLEncodedForm()
        data['text'] = [test_data]

        http_response = self.uri_opener.POST(url, data, cache=False)
        self.assertIn(test_data, http_response.body)

 def test_redirect_handler(self):
     """Test the redirect handler using urllib2"""
     redirect_url = URL(get_moth_http('/audit/global_redirect/redirect-header-302.py?url=/'))
     opener = urllib2.build_opener(HTTP30XHandler)
     
     request = urllib2.Request(redirect_url.url_string)
     response = opener.open(request)
     
     self.assertEqual(response.code, FOUND)