diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 56796e6..5ede2e0 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/ftw/http.py b/ftw/http.py index a4b91d8..31e6f27 100644 --- a/ftw/http.py +++ b/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,11 +366,11 @@ 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 @@ -389,17 +378,17 @@ def build_request(self): 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 diff --git a/ftw/logchecker.py b/ftw/logchecker.py index c99ad83..3f3d557 100644 --- a/ftw/logchecker.py +++ b/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 diff --git a/ftw/pytest_plugin.py b/ftw/pytest_plugin.py index 710e81f..e141d59 100644 --- a/ftw/pytest_plugin.py +++ b/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): diff --git a/ftw/ruleset.py b/ftw/ruleset.py index 1a865f6..7b250e1 100644 --- a/ftw/ruleset.py +++ b/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) diff --git a/ftw/testrunner.py b/ftw/testrunner.py index 12fbda0..217e6b5 100644 --- a/ftw/testrunner.py +++ b/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 diff --git a/ftw/util.py b/ftw/util.py index ce7fdea..7c4ef09 100644 --- a/ftw/util.py +++ b/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)) diff --git a/requirements.txt b/requirements.txt index 6f40c81..a5d2a37 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,4 +3,3 @@ IPy==0.83 PyYAML==4.2b1 pytest==4.6 python-dateutil==2.6.0 -six==1.14.0 diff --git a/setup.py b/setup.py index 70cd745..9df9859 100644 --- a/setup.py +++ b/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' ])