Remove six

.github/workflows/ci.yml

@@ -9,7 +9,7 @@ jobs:
 
     strategy:
       matrix:
-        python-version: [ '2.x', '3.6', '3.7', '3.8', '3.9' ]
+        python-version: [ '3.6', '3.7', '3.8', '3.9' ]
 
     steps:
       - name: Checkout repo

ftw/http.py

@@ -1,6 +1,4 @@
-
-import brotli
-import io
+from io import BytesIO
 import socket
 import ssl
 import errno
@@ -12,30 +10,21 @@
 import base64
 import zlib
 import encodings
+import brotli
 from IPy import IP
-
-from six import BytesIO, PY2, ensure_binary, ensure_str, iteritems, \
-    text_type
-from six.moves import http_cookies
+from http import cookies
 
 from . import errors
+from . import util
 
 
 # Fallback to PROTOCOL_SSLv23 if PROTOCOL_TLS is not available.
 PROTOCOL_TLS = getattr(ssl, "PROTOCOL_TLS", ssl.PROTOCOL_SSLv23)
 
 
-if PY2:
-    reload(sys)  # pragma: no flakes
-    sys.setdefaultencoding('utf8')
-    escape_codec = 'string_escape'
-else:
-    escape_codec = 'unicode_escape'
-
-
 class HttpResponse(object):
     def __init__(self, http_response, user_agent):
-        self.response = ensure_binary(http_response)
+        self.response = util.ensure_binary(http_response)
         # For testing purposes HTTPResponse might be called OOL
         try:
             self.dest_addr = user_agent.request_object.dest_addr
@@ -139,7 +128,7 @@ def check_for_cookie(self, cookie):
                             'function': 'http.HttpResponse.check_for_cookie'
                         })
                 try:
-                    with io.open(psl_path, 'r', encoding='utf-8') as fo:
+                    with open(psl_path, 'r', encoding='utf-8') as fo:
                         for line in fo:
                             if line[:2] == '//' or line[0] == ' ' or \
                                line[0].strip() == '':
@@ -185,7 +174,7 @@ def process_response(self):
         Parses an HTTP response after an HTTP request is sent
         """
         split_response = self.response.split(self.CRLF)
-        response_line = ensure_str(split_response[0])
+        response_line = util.ensure_str(split_response[0])
         response_headers = {}
         response_data = None
         data_line = None
@@ -204,13 +193,13 @@ def process_response(self):
                             'header_rcvd': str(header),
                             'function': 'http.HttpResponse.process_response'
                         })
-                header = ensure_str(header[0]), ensure_str(header[1])
+                header = util.ensure_str(header[0]), util.ensure_str(header[1])
                 response_headers[header[0].lower()] = header[1].lstrip()
         if 'set-cookie' in list(response_headers.keys()):
             try:
-                cookie = http_cookies.SimpleCookie()
+                cookie = cookies.SimpleCookie()
                 cookie.load(response_headers['set-cookie'])
-            except http_cookies.CookieError as err:
+            except cookies.CookieError as err:
                 raise errors.TestError(
                     'Error processing the cookie content into a SimpleCookie',
                     {
@@ -364,9 +353,9 @@ def build_request(self):
             if 'cookie' in list(self.request_object.headers.keys()):
                 # Create a SimpleCookie out of our provided cookie
                 try:
-                    provided_cookie = http_cookies.SimpleCookie()
+                    provided_cookie = cookies.SimpleCookie()
                     provided_cookie.load(self.request_object.headers['cookie'])
-                except http_cookies.CookieError as err:
+                except cookies.CookieError as err:
                     raise errors.TestError(
                         'Error processing the existing cookie into a '
                         'SimpleCookie',
@@ -377,29 +366,29 @@ def build_request(self):
                             'function': 'http.HttpResponse.build_request'
                         })
                 result_cookie = {}
-                for cookie_key, cookie_morsal in iteritems(provided_cookie):
+                for cookie_key, cookie_morsal in list(provided_cookie.items()):
                     result_cookie[cookie_key] = \
                         provided_cookie[cookie_key].value
                 for cookie in available_cookies:
-                    for cookie_key, cookie_morsal in iteritems(cookie):
+                    for cookie_key, cookie_morsal in cookie:
                         if cookie_key in list(result_cookie.keys()):
                             # we don't overwrite a user specified
                             # cookie with a saved one
                             pass
                         else:
                             result_cookie[cookie_key] = \
                                 cookie[cookie_key].value
-                for key, value in iteritems(result_cookie):
-                    cookie_value += (text_type(key) + '=' +
-                                     text_type(value) + '; ')
+                for key, value in list(result_cookie.items()):
+                    cookie_value += (str(key) + '=' +
+                                     str(value) + '; ')
                     # Remove the trailing semicolon
                 cookie_value = cookie_value[:-2]
                 self.request_object.headers['cookie'] = cookie_value
             else:
                 for cookie in available_cookies:
-                    for cookie_key, cookie_morsal in iteritems(cookie):
-                        cookie_value += (text_type(cookie_key) + '=' +
-                                         text_type(cookie_morsal.coded_value) +
+                    for cookie_key, cookie_morsal in list(cookie.items()):
+                        cookie_value += (str(cookie_key) + '=' +
+                                         str(cookie_morsal.coded_value) +
                                          '; ')
                         # Remove the trailing semicolon
                     cookie_value = cookie_value[:-2]
@@ -408,9 +397,9 @@ def build_request(self):
         # Expand out our headers into a string
         headers = ''
         if self.request_object.headers != {}:
-            for hname, hvalue in iteritems(self.request_object.headers):
-                headers += text_type(hname) + ': ' + \
-                    text_type(hvalue) + self.CRLF
+            for hname, hvalue in self.request_object.headers.items():
+                headers += str(hname) + ': ' + \
+                    str(hvalue) + self.CRLF
         request = request.replace('#headers#', headers)
 
         # If we have data append it
@@ -435,18 +424,19 @@ def build_request(self):
                     if choice in possible_choices:
                         encoding = choice
             try:
-                data = self.request_object.data.encode(encoding)
-            except UnicodeEncodeError as err:
+                data_bytes = \
+                    self.request_object.data.encode(encoding, 'strict')
+            except UnicodeError as err:
                 raise errors.TestError(
                     'Error encoding the data with the charset specified',
                     {
                         'msg': str(err),
                         'Content-Type':
                             str(self.request_object.headers['Content-Type']),
-                        'data': text_type(self.request_object.data),
+                        'data': str(self.request_object.data),
                         'function': 'http.HttpResponse.build_request'
                     })
-            request = request.replace('#data#', ensure_str(data))
+            request = request.replace('#data#', util.ensure_str(data_bytes))
         else:
             request = request.replace('#data#', '')
         # If we have a Raw Request we should use that instead
@@ -457,15 +447,15 @@ def build_request(self):
                     {
                         'function': 'http.HttpUA.build_request'
                     })
-            request = ensure_binary(self.request_object.raw_request)
+            request = self.request_object.raw_request.encode('utf-8', 'strict')
             # We do this regardless of magic if you want to send a literal
             # '\' 'r' or 'n' use encoded request.
-            request = request.decode(escape_codec)
+            request = request.decode('unicode_escape')
         if self.request_object.encoded_request is not None:
             request = base64.b64decode(self.request_object.encoded_request)
-            request = request.decode(escape_codec)
+            request = request.decode('unicode_escape')
         # if we have an Encoded request we should use that
-        self.request = ensure_binary(request)
+        self.request = request.encode('utf-8', 'strict')
 
     def get_response(self):
         """
@@ -486,7 +476,7 @@ def get_response(self):
             try:
                 data = self.sock.recv(self.RECEIVE_BYTES)
                 if data:
-                    our_data.append(ensure_binary(data))
+                    our_data.append(util.ensure_binary(data))
                     begin = time.time()
                 else:
                     # Sleep for sometime to indicate a gap

ftw/logchecker.py

@@ -1,9 +1,7 @@
-import abc
-import six
+from abc import ABC, abstractmethod
 
 
-@six.add_metaclass(abc.ABCMeta)
-class LogChecker():
+class LogChecker(ABC):
     """
     LogChecker is an abstract class that integrations with WAFs MUST implement.
     This class is used by the testrunner to test log lines against an expected
@@ -17,7 +15,7 @@ def set_times(self, start, end):
         self.start = start
         self.end = end
 
-    @abc.abstractmethod
+    @abstractmethod
     def get_logs(self):
         """
         MUST be implemented, MUST return an array of strings

ftw/pytest_plugin.py

@@ -3,8 +3,8 @@
 from . import util
 from .ruleset import Test
 
-from six.moves.BaseHTTPServer import HTTPServer
-from six.moves.SimpleHTTPServer import SimpleHTTPRequestHandler
+from http.server import HTTPServer
+from http.server import SimpleHTTPRequestHandler
 
 
 def get_testdata(rulesets):

ftw/ruleset.py

@@ -1,9 +1,9 @@
 import re
 
-from six import ensure_str
-from six.moves.urllib.parse import parse_qsl, unquote, urlencode
+from urllib.parse import parse_qsl, unquote, urlencode
 
 from . import errors
+from . import util
 
 
 class Output(object):
@@ -122,7 +122,7 @@ def __init__(self, raw_request=None,
             if 'Content-Type' in list(headers.keys()):
                 if headers['Content-Type'] == \
                    'application/x-www-form-urlencoded' and stop_magic is False:
-                    if ensure_str(unquote(self.data)) == self.data:
+                    if util.ensure_str(unquote(self.data)) == self.data:
                         query_string = parse_qsl(self.data)
                         if len(query_string) != 0:
                             encoded_args = urlencode(query_string)

ftw/testrunner.py

@@ -4,7 +4,6 @@
 import pytest
 import sqlite3
 
-from six import ensure_str
 
 from . import errors
 from . import http
@@ -57,7 +56,7 @@ def test_response(self, response_object, regex):
                     'response_object': response_object,
                     'function': 'testrunner.TestRunner.test_response'
                 })
-        if regex.search(ensure_str(response_object.response)):
+        if regex.search(util.ensure_str(response_object.response)):
             assert True
         else:
             assert False
@@ -67,7 +66,7 @@ def test_response_str(self, response, regex):
         Checks if the response response contains a regex specified in the
         output stage. It will assert that the regex is present.
         """
-        if regex.search(ensure_str(response)):
+        if regex.search(util.ensure_str(response)):
             assert True
         else:
             assert False

ftw/util.py

@@ -1,9 +1,7 @@
-
-import io
-import yaml
 import os
 import sqlite3
 from glob import glob
+import yaml
 
 from . import ruleset
 
@@ -86,7 +84,7 @@ def extract_yaml(yaml_files):
     loaded_yaml = []
     for yaml_file in yaml_files:
         try:
-            with io.open(yaml_file, encoding='utf-8') as fd:
+            with open(yaml_file, encoding='utf-8') as fd:
                 loaded_yaml.append(yaml.safe_load(fd))
         except IOError as e:
             print('Error reading file', yaml_file)
@@ -98,3 +96,21 @@ def extract_yaml(yaml_files):
             print('General error')
             raise e
     return loaded_yaml
+
+
+def ensure_str(s, encoding='utf-8', errors='strict'):
+    # Optimization: Fast return for the common case.
+    if isinstance(s, str):
+        return s
+    if isinstance(s, bytes):
+        return s.decode(encoding, errors)
+    elif not isinstance(s, (str, bytes)):
+        raise TypeError("not expecting type '%s'" % type(s))
+
+
+def ensure_binary(s, encoding='utf-8', errors='strict'):
+    if isinstance(s, bytes):
+        return s
+    if isinstance(s, str):
+        return s.encode(encoding, errors)
+    raise TypeError("not expecting type '%s'" % type(s))

requirements.txt

@@ -3,4 +3,3 @@ IPy==0.83
 PyYAML==4.2b1
 pytest==4.6
 python-dateutil==2.6.0
-six==1.14.0

setup.py

@@ -25,6 +25,5 @@
           'IPy==0.83',
           'PyYAML==4.2b1',
           'pytest==4.6',
-          'python-dateutil==2.6.0',
-          'six==1.14.0'
+          'python-dateutil==2.6.0'
       ])