diff --git a/libmproxy/netstring.py b/libmproxy/netstring.py index a9b71a329..a7b7a5372 100644 --- a/libmproxy/netstring.py +++ b/libmproxy/netstring.py @@ -10,13 +10,6 @@ def header(data): return str(len(data))+":" -def encode(data): - if not isinstance(data, str): - raise ValueError("data should be of type 'str'") - - return "%i:%s," % (len(data), data) - - class FileEncoder(object): def __init__(self, file_out): """" @@ -39,43 +32,16 @@ class FileEncoder(object): return self -def netstrings_to_file(file_out, data_container): - """ - Writes a container of netstrings to a file. - - file_out -- A writeable file-like object - data_container -- An iterable of strings - """ - write = file_out.write - for s in data_container: - if not isinstance(s, str): - raise ValueError("data should be of type 'str'") - write(header(s)) - write(s) - write(',') - - -def encode_netstrings(data_container): - """ - Encodes a number of strings as sequence of netstrings. - - data_container -- An iterable of strings to be encoded - """ - return "".join(encode(s) for s in data_container) - - class DecoderError(Exception): - ( - PRECEDING_ZERO_IN_SIZE, - MAX_SIZE_REACHED, - ILLEGAL_DIGIT_IN_SIZE, - ILLEGAL_DIGIT - ) = range(4) + PRECEDING_ZERO_IN_SIZE = 0 + MAX_SIZE_REACHED = 1 + ILLEGAL_DIGIT_IN_SIZE = 2 + ILLEGAL_DIGIT = 3 error_text = { - PRECEDING_ZERO_IN_SIZE:"PRECEDING_ZERO_IN_SIZE", - MAX_SIZE_REACHED:"MAX_SIZE_REACHED", - ILLEGAL_DIGIT_IN_SIZE:"ILLEGAL_DIGIT_IN_SIZE", - ILLEGAL_DIGIT:"ILLEGAL_DIGIT" + PRECEDING_ZERO_IN_SIZE: "PRECEDING_ZERO_IN_SIZE", + MAX_SIZE_REACHED: "MAX_SIZE_REACHED", + ILLEGAL_DIGIT_IN_SIZE: "ILLEGAL_DIGIT_IN_SIZE", + ILLEGAL_DIGIT: "ILLEGAL_DIGIT" } def __init__(self, code, text): Exception.__init__(self) @@ -109,34 +75,6 @@ class Decoder(object): self.data_out = StringIO() self.yield_data = "" - def __str__(self): - if self.data_size is None: - bytes = len(self.size_string) - else: - bytes = self.data_out.tell() - return ""%bytes - - def peek_buffer(self): - """ - Returns any bytes not used by decoder. - """ - return self.data_out.getvalue() - - def reset(self): - """ - Resets decoder to initial state, and discards any cached stream data. - """ - self.data_pos = 0 - self.string_start = 0 - self.expecting_terminator = False - self.size_string = "" - self.data_size = None - self.remaining_bytes = 0 - self.yield_data = "" - - self.data_out.reset() - self.data_out.truncate() - def feed(self, data): """ A generator that yields 0 or more strings from the given data. @@ -200,19 +138,6 @@ class Decoder(object): self.expecting_terminator = True -def decode(data): - """ - Decodes netstrings and returns a tuple containing a - list of strings, and any remaining data. - - data -- A string containing netstring data - """ - decoder = Decoder() - netstrings = list(decoder.feed(data)) - remaining = data[decoder.string_start:] - return netstrings, remaining - - def decode_file(file_in, buffer_size=1024): """ Generates 0 or more strings from a netstring file. diff --git a/libmproxy/utils.py b/libmproxy/utils.py index ab861c55e..afef8e638 100644 --- a/libmproxy/utils.py +++ b/libmproxy/utils.py @@ -342,8 +342,10 @@ def dummy_ca(path): stdout=subprocess.PIPE, stdin=subprocess.PIPE ) + # begin nocover if ret: return False + # end nocover else: return True @@ -382,8 +384,7 @@ def dummy_cert(certdir, ca, commonname): stdout=subprocess.PIPE, stdin=subprocess.PIPE ) - if ret: - return None + if ret: return None cmd = [ "openssl", "x509", @@ -402,8 +403,7 @@ def dummy_cert(certdir, ca, commonname): stdout=subprocess.PIPE, stdin=subprocess.PIPE ) - if ret: - return None + if ret: return None else: # Create a new selfsigned certificate + key cmd = [ @@ -424,8 +424,7 @@ def dummy_cert(certdir, ca, commonname): stdout=subprocess.PIPE, stdin=subprocess.PIPE ) - if ret: - return None + if ret: return None return certpath @@ -437,5 +436,3 @@ def mkdir_p(path): pass else: raise - - diff --git a/test/test_netstring.py b/test/test_netstring.py index efcfd83f1..5146d1507 100644 --- a/test/test_netstring.py +++ b/test/test_netstring.py @@ -16,13 +16,6 @@ class uNetstring(libpry.AutoTree): for test, result in tests: assert netstring.header(test) == result - def test_encode(self): - tests = [ ("netstring", "9:netstring,"), - ("Will McGugan", "12:Will McGugan,"), - ("", "0:,") ] - for test, result in tests: - assert netstring.encode(test) == result - def test_file_encoder(self): file_out = StringIO() data = self.test_data.split()