def test_can_be_pickled(self):
        # Pickle a URL object that contains a cache
        u = URL('http://www.w3af.com/')
        domain_path = u.get_domain_path()

        cPickle.dumps(u)
        cPickle.dumps(domain_path)

 def test_default_proto(self):
     """
     http is the default protocol, we can provide URLs with no proto
     """
     u = URL('w3af.com')
     self.assertEqual(u.get_domain(), 'w3af.com')
     self.assertEqual(u.get_protocol(), 'http')

 def test_url_join_case03(self):
     u = URL('http://w3af.com/def/jkl/')
     self.assertEqual(u.url_join('/def/abc.html').url_string,
                      u'http://w3af.com/def/abc.html')
     
     self.assertEqual(u.url_join('def/abc.html').url_string,
                      u'http://w3af.com/def/jkl/def/abc.html')

 def test_simplest_url(self):
     u = URL('http://w3af.com/foo/bar.txt')
     
     self.assertEqual(u.path, '/foo/bar.txt')
     self.assertEqual(u.scheme, 'http')
     self.assertEqual(u.get_file_name(), 'bar.txt')
     self.assertEqual(u.get_extension(), 'txt')

 def test_websocket_secure_proto(self):
     """
     We can also parse and handle ws and wss protocols
     """
     u = URL('wss://w3af.com')
     self.assertEqual(u.get_domain(), 'w3af.com')
     self.assertEqual(u.get_protocol(), 'wss')

    def from_httplib_resp(cls, httplibresp, original_url=None):
        """
        Factory function. Build a HTTPResponse object from a
        httplib.HTTPResponse instance
    
        :param httplibresp: httplib.HTTPResponse instance
        :param original_url: Optional 'url_object' instance.
    
        :return: A HTTPResponse instance
        """
        resp = httplibresp
        code, msg, hdrs, body = (resp.code, resp.msg, resp.info(), resp.read())
        hdrs = Headers(hdrs.items())

        if original_url:
            url_inst = URL(resp.geturl(), original_url.encoding)
            url_inst = url_inst.url_decode()
        else:
            url_inst = original_url = URL(resp.geturl())

        httplib_time = DEFAULT_WAIT_TIME
        if hasattr(httplibresp, 'get_wait_time'):
            # This is defined in the keep alive http response object
            httplib_time = httplibresp.get_wait_time()

        if isinstance(resp, urllib2.HTTPError):
            # This is possible because in errors.py I do:
            # err = urllib2.HTTPError(req.get_full_url(), code, msg, hdrs, resp)
            charset = getattr(resp.fp, 'encoding', None)
        else:
            # The encoding attribute is only set on CachedResponse instances
            charset = getattr(resp, 'encoding', None)
        
        return cls(code, body, hdrs, url_inst, original_url,
                   msg, charset=charset, time=httplib_time)

 def test_url_join_case01(self):
     u = URL('http://w3af.com/foo.bar')
     self.assertEqual(u.url_join('abc.html').url_string,
                      u'http://w3af.com/abc.html')
     
     self.assertEqual(u.url_join('/abc.html').url_string,
                      u'http://w3af.com/abc.html')

 def test_encode_decode(self):
     """Encode and Decode should be able to run one on the result of the
     other and return the original"""
     original = URL(u'https://w3af.com:443/file.asp?id=1%202')
     encoded = original.url_encode()
     decoded = URL(encoded).url_decode()
     self.assertEqual(original, decoded)

    def test_from_url_keep_form(self):
        o = URL('http://w3af.com/foo/bar.txt')
        o.querystring = URLEncodedForm()

        u = URL.from_URL(o)
        self.assertIsInstance(u.querystring, URLEncodedForm)
        self.assertIsNot(u.querystring, o.querystring)
        self.assertEqual(u.querystring, o.querystring)

    def http_request(self, req):
        url_instance = URL(req.get_full_url())
        url_instance.set_param(self._url_parameter)

        new_request = HTTPRequest(url_instance, headers=req.headers,
                                  origin_req_host=req.get_origin_req_host(),
                                  unverifiable=req.is_unverifiable(),
                                  retries=req.retries_left)
        return new_request

    def do_follow_redirect(self, req, fp, code, msg, headers):

        # Check if we can redirect according to the RFC
        if not self.redirect_allowed_by_rfc(req, code):
            raise self.create_error_from_parts(req, code, msg, headers, fp)

        # Some servers (incorrectly) return multiple Location headers
        # (so probably same goes for URI). Use first header.
        if LOCATION in headers:
            new_url_raw = headers.getheaders(LOCATION)[0]
        elif URI in headers:
            new_url_raw = headers.getheaders(URI)[0]
        else:
            raise self.create_error_from_parts(req, code, msg, headers, fp)

        # Calculate the target URL
        try:
            current_url = URL(req.get_full_url())
            new_url_str = current_url.url_join(new_url_raw).url_string
            new_url_obj = current_url.url_join(new_url_raw)
        except ValueError:
            raise self.create_error_from_parts(req, code, msg, headers, fp)

        # For security reasons we do not allow redirects to protocols
        # other than HTTP or HTTPS
        new_url_lower = new_url_str.lower()
        if not (new_url_lower.startswith('http://') or
                new_url_lower.startswith('https://')):
            raise self.create_error_from_parts(req, code, msg, headers, fp)

        # XXX Probably want to forget about the state of the current
        # request, although that might interact poorly with other
        # handlers that also use handler-specific request attributes
        new_request = self.create_redirect_request(req, fp, code, msg,
                                                   headers, new_url_str,
                                                   new_url_obj)

        # loop detection
        # .redirect_dict has a key url if url was previously visited.
        if hasattr(req, 'redirect_dict'):
            visited = new_request.redirect_dict = req.redirect_dict
            if (visited.get(new_url_str, 0) >= self.max_repeats or
                len(visited) >= self.max_redirections):
                raise self.create_error_from_parts(req, code, msg, headers, fp)
        else:
            visited = new_request.redirect_dict = req.redirect_dict = {}

        visited[new_url_str] = visited.get(new_url_str, 0) + 1

        # Don't close the fp until we are sure that we won't use it
        # with HTTPError.
        fp.read()
        fp.close()

        return self.parent.open(new_request, timeout=req.timeout)

        def get_response(self, http_request, uri, response_headers):
            uri = URL(uri)

            try:
                callback = uri.get_querystring()['callback'][0]
            except KeyError:
                callback = 'default'

            body = '%s({})' % callback
            response_headers['Content-Type'] = 'application/javascript'

            return self.status, response_headers, body

 def test_from_url(self):
     o = URL('http://w3af.com/foo/bar.txt')
     u = URL.from_URL(o)
     
     self.assertEqual(u.path, '/foo/bar.txt')
     self.assertEqual(u.scheme, 'http')
     self.assertEqual(u.get_file_name(), 'bar.txt')
     self.assertEqual(u.get_extension(), 'txt')
     
     o = URL('w3af.com')
     u = URL.from_URL(o)
     self.assertEqual(u.get_domain(), 'w3af.com')
     self.assertEqual(u.get_protocol(), 'http')

    def test_memoized(self):
        u = URL('http://www.w3af.com/')
        self.assertEqual(u._cache, dict())

        url = u.uri2url()
        self.assertNotEqual(u._cache, dict())
        self.assertIn(url, u._cache.values())

        second_url = u.uri2url()
        self.assertIs(url, second_url)

        self.assertIsInstance(url, URL)
        self.assertIsInstance(second_url, URL)

 def test_set_params(self):
     u = URL('http://w3af.com/;id=1')
     u.set_param('file=2')
     
     self.assertEqual(u.get_params_string(), 'file=2')
     
     u = URL('http://w3af.com/xyz.txt;id=1?file=2')
     u.set_param('file=3')
     
     self.assertEqual(u.get_params_string(), 'file=3')
     self.assertEqual(u.get_path_qs(), '/xyz.txt;file=3?file=2')

    def test_phishtank_match_last_url(self):
        phishtank_inst = self.w3afcore.plugins.get_plugin_inst('crawl',
                                                               'phishtank')

        vuln_url = URL(self.get_last_vulnerable_url())
        phishtank_inst.crawl(FuzzableRequest(vuln_url))

        vulns = self.kb.get('phishtank', 'phishtank')

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

        self.assertEqual(vuln.get_name(), 'Phishing scam')
        self.assertEqual(vuln.get_severity(), MEDIUM)
        self.assertEqual(vuln.get_url().get_domain(), vuln_url.get_domain())

    def test_redirect_uri_relative(self):
        ws = web_spider()
        body = ''
        url = URL('http://www.w3af.org')
        redir_url = '/redir'
        headers = Headers([('content-type', 'text/html'),
                           ('uri', redir_url)])
        resp = HTTPResponse(200, body, headers, url, url)

        gen = ws._headers_url_generator(resp, None)

        extracted_data = [i for i in gen]
        expected_data = [(url.url_join(redir_url), None, resp, False)]

        self.assertEqual(extracted_data, expected_data)

    def met_search(self, query):
        """
        Query a Public Key Server.

        This method is based from the pks.py file from the massive enumeration
        toolset, coded by pdp and released under GPL v2.
        """
        url = URL(u'http://pgp.mit.edu:11371/pks/lookup')
        url.querystring = [(u'op', [u'index']), (u'search', [query])]

        try:
            response = self._uri_opener.GET(url, headers=self._headers,
                                            cache=True, grep=False)
        except HTTPRequestException:
            # Very naive exception handling for the case where we can't reach
            # the PKS server (it's down, blocking us, bad internet connection)
            return []

        content = response.get_body()

        content = re.sub('(<.*?>|&lt;|&gt;)', '', content)

        results = []
        accounts = []

        for line in content.split('\n')[2:]:
            if not line.strip():
                continue

            tokens = line.split()

            if len(tokens) >= 5:
                email = tokens[-1]
                name = ' '.join(tokens[3:-1])

                if SGMLParser.EMAIL_RE.match(email):

                    account = email.split('@')[0]
                    domain = email.split('@')[1]

                    if domain == query:
                        if account not in accounts:
                            accounts.append(account)
                            
                            pksr = PKSResult(name, account, domain, response.id)
                            results.append(pksr)

        return results

    def get_response(self, http_request, uri, response_headers):
        """
        Overwrite the mock response with one simple objective: add a delay
        which depends on the length of the redos parameter.
        """
        response_headers.update({'status': self.status})
        response_headers.update(self.headers)

        uri = URL(uri)
        qs = uri.get_querystring()
        redos_param = qs.get('redos')[0]

        delay = len(redos_param) / 13.0
        time.sleep(delay)

        return self.status, response_headers, self.body

 def test_from_parts(self):
     u = URL.from_parts('http', 'w3af.com', '/foo/bar.txt', None, 'a=b',
                        'frag')
     
     self.assertEqual(u.path, '/foo/bar.txt')
     self.assertEqual(u.scheme, 'http')
     self.assertEqual(u.get_file_name(), 'bar.txt')
     self.assertEqual(u.get_extension(), 'txt')