"isb", "generate_strings_from_regex", "get_alphabet_from_mask", "handle_error", "i2s", "is_native",

"list_categories", "list_encodings", "list_macros", "lookup", "maketrans", "os", "rank", "re", "register",

"remove", "reset", "s2i", "search", "stopfunc", "BytesIO", "_input", "_stripl", "CodecMacro",

'a': printable,

'b': "".join(chr(i) for i in range(256)),

'd': digits,

'h': digits + "abcdef",

'H': digits + "ABCDEF",

'l': ascii_lowercase,

'p': punctuation,

's': " ",

'u': ascii_uppercase,

"""Macro details when looking up the codec registry. """

def __new__(cls, name):

self = tuple.__new__(cls)

self.name = name

# get from personal macros first

try:

self.codecs = PERS_MACROS[name]

except KeyError:

try:

self.codecs = MACROS[name]

except KeyError:

raise LookupError("unknown macro: %s" % name)

if not isinstance(self.codecs, (tuple, list)):

raise ValueError("bad macro list: %s" % str(self.codecs))

self.codecs = [lookup(e, False) for e in self.codecs] # lookup(e, False)

self.parameters = {'name': name, 'category': "macro"} # ^ means that macros won't be nestable

# test examples to check that the chain of encodings works

for action, examples in (self.codecs[0].parameters.get('examples', {}) or {'enc-dec(': ["T3st str!"]}).items():

if re.match(r"enc(-dec)?\(", action):

for e in (examples.keys() if action.startswith("enc(") else examples or []):

rd = re.match(r"\@random(?:\{(\d+(?:,(\d+))*?)\})?$", e)

if rd:

for n in (rd.group(1) or "512").split(","):

self.encode("".join(chr(randint(0, 255)) for i in range(int(n))))

continue

self.encode(e)

class Codec:

decode = self.decode

encode = self.encode

class IncrementalEncoder(codecs.IncrementalEncoder):

def encode(self, input, final=False):

return b(self.encode(input, self.errors)[0])

self.incrementalencoder = IncrementalEncoder

class IncrementalDecoder(codecs.IncrementalDecoder):

def decode(self, input, final=False):

return ensure_str(self.decode(input, self.errors)[0])

self.incrementaldecoder = IncrementalDecoder

class StreamWriter(Codec, codecs.StreamWriter):

charbuffertype = bytes

self.streamwriter = StreamWriter

class StreamReader(Codec, codecs.StreamReader):

charbuffertype = bytes

self.streamreader = StreamReader

return self

def decode(self, input, error="strict"):

""" Decode with each codec in reverse order. """

for ci in self.codecs[::-1]:

input, l = ci.decode(input, error)

return input, l

def encode(self, input, error="strict"):

""" Encode with each codec. """

for ci in self.codecs:

input, l = ci.encode(input, error)

return input, l

def __repr__(self):

def __init__(self, name, func):

self.__name = name

self.__func = func

update_wrapper(self, func)

def __call__(self, *args, **kwargs):

return self.__func(*args, **kwargs)

def __repr__(self):

""" Stdin pipe read function. """

try:

with open(0, 'rb') as f:

for l in f:

yield l

except TypeError:

for l in sys.stdin:

# handle input file or stdin

c = b("")

if infile:

with open(infile, 'rb') as f:

c = f.read()

else:

for line in __stdin_pipe():

c += line

if st_crlf:

s = s.replace(b"\r\n", b"") if isb(s) else s.replace("\r\n", "")

if st_lines:

s = s.replace(b"\n", b"") if isb(s) else s.replace("\n", "")

""" This adds a new codec to the codecs module setting its encode and/or decode functions, eventually dynamically

remove(ename)

if encode:

if not isinstance(encode, FunctionType):

raise ValueError("Bad 'encode' function")

_set_exc("%sEncodeError" % exc_name(ename)) # create the custom encode exception as a builtin

if decode:

if not isinstance(decode, FunctionType):

raise ValueError("Bad 'decode' function")

_set_exc("%sDecodeError" % exc_name(ename)) # create the custom decode exception as a builtin

if not encode and not decode:

raise ValueError("At least one en/decoding function must be defined")

for exc in kwargs.get('extra_exceptions', []):

_set_exc(exc) # create additional custom exceptions as builtins

glob = currentframe().f_back.f_globals

# search function for the new encoding

@_with_repr(ename)

def getregentry(encoding):

if encoding != ename and not (pattern and re.match(pattern, encoding)):

return

fenc, fdec, name = encode, decode, encoding

# prepare CodecInfo input arguments

if pattern:

m, args, i = re.match(pattern, encoding), [], 1

try:

while True:

try:

g = m.group(i) or ""

if g.isdigit() and not g.startswith("0") and "".join(set(g)) != "01":

g = int(g)

args += [g]

i += 1

except AttributeError:

# this occurs when m is None or there is an error in fenc(g) or fdec(g), meaning no match

if m is not None:

raise

return

except IndexError:

# this occurs while m is not None, but possibly no capture group that gives at least 1 group index ;

# in this case, if fenc/fdec is a decorated function, execute it with no arg

if len(args) == 0:

if fenc and len(getfullargspec(fenc).args) == 1:

fenc = fenc()

if fdec and len(getfullargspec(fdec).args) == 1:

fdec = fdec()

else:

fenc = fenc(*args) if fenc else fenc

fdec = fdec(*args) if fdec else fdec

if fenc:

fenc = fix_inout_formats(fenc)

if fdec:

fdec = fix_inout_formats(fdec)

sl, sc = kwargs.pop('strip_lines', False), kwargs.pop('strip_crlf', False)

if sl or sc:

def _striplines(f):

def __wrapper(input, *a, **kw):

return f(_stripl(input, sc, sl), *a, **kw)

return __wrapper

# this fixes issues with wrapped encoded inputs

fdec = _striplines(fdec)

class Codec(codecs.Codec):

def encode(self, input, errors="strict"):

if fenc is None:

raise NotImplementedError

return fenc(input, errors)

def decode(self, input, errors="strict"):

if fdec is None:

raise NotImplementedError

return fdec(input, errors)

class IncrementalEncoder(codecs.IncrementalEncoder):

def encode(self, input, final=False):

if fenc is None:

raise NotImplementedError

return b(fenc(input, self.errors)[0])

class IncrementalDecoder(codecs.IncrementalDecoder):

def decode(self, input, final=False):

if fdec is None:

raise NotImplementedError

return ensure_str(fdec(input, self.errors)[0])

class StreamWriter(Codec, codecs.StreamWriter):

charbuffertype = bytes

class StreamReader(Codec, codecs.StreamReader):

charbuffertype = bytes

ci = codecs.CodecInfo(

name=name,

encode=Codec().encode,

decode=Codec().decode,

incrementalencoder=IncrementalEncoder,

incrementaldecoder=IncrementalDecoder,

streamwriter=StreamWriter,

streamreader=StreamReader,

_is_text_encoding=text,

)

ci.parameters = kwargs

ci.parameters['name'] = ename

ci.parameters['add_to_codecs'] = add_to_codecs

ci.parameters['pattern'] = pattern

ci.parameters['text'] = text

f = glob.get('__file__', os.path.join("custom", "_"))

cat = f.split(os.path.sep)[-2].rstrip("s")

if cat not in CODECS_CATEGORIES:

CODECS_CATEGORIES.append(cat)

ci.parameters['category'] = kwargs.get('category', cat)

ci.parameters['examples'] = kwargs.get('examples', glob.get('__examples__'))

ci.parameters['guess'] = kwargs.get('guess', glob.get('__guess__', [ename])) or []

ci.parameters['module'] = kwargs.get('module', glob.get('__name__'))

ci.parameters.setdefault("scoring", {})

for attr in ["bonus_func", "entropy", "expansion_factor", "len_charset", "penalty", "printables_rate",

"padding_char", "transitive"]:

a = kwargs.pop(attr, None)

if a is not None:

ci.parameters['scoring'][attr] = a

return ci

getregentry.__name__ = re.sub(r"[\s\-]", "_", ename)

if kwargs.get('aliases'):

getregentry.__aliases__ = list(map(lambda n: re.sub(r"[\s\-]", "_", n), kwargs['aliases']))

getregentry.__pattern__ = pattern

register(getregentry, add_to_codecs)

""" This allows to define a macro, chaining multiple codecs one after the other. This relies on a default set of

global PERS_MACROS

# check for name clash with alreday existing macros and codecs

if mname in MACROS or mname in PERS_MACROS:

raise ValueError("Macro name already exists")

try:

ci = lookup(mname, False)

raise ValueError("Macro name clashes with codec '%s'" % ci.name)

except LookupError:

pass

try:

PERS_MACROS[mname] = encodings

CodecMacro(mname)

with open(PERS_MACROS_FILE, 'w') as f:

json.dump(PERS_MACROS, f, indent=2)

except ValueError:

del PERS_MACROS[mname]

""" This adds a new mapping codec (that is, declarable with a simple character mapping dictionary) to the codecs

outype = outype or intype

if ignore_case not in [None, "encode", "decode", "both"]:

raise ValueError("Bad ignore_case parameter while creating encoding map")

if intype not in [None, "str", "bin", "ord"]:

raise ValueError("Bad input type parameter while creating encoding map")

if outype not in [None, "str", "bin", "ord"]:

raise ValueError("Bad output type parameter while creating encoding map")

def __generic_code(decode=False):

def _wrapper(param):

""" The parameter for wrapping comes from the encoding regex pattern ; e.g.

p = param

if isinstance(encmap, FunctionType):

mapdict = encmap(p)

p = None

else:

mapdict = encmap

if isinstance(mapdict, dict):

smapdict = {k: v for k, v in mapdict.items()}

elif isinstance(mapdict, list) and isinstance(mapdict[0], dict):

smapdict = {k: v for k, v in mapdict[0].items()}

else:

raise ValueError("Bad mapping dictionary or list of mapping dictionaries")

if p is not None:

# case 1: param is empty string

if p == "":

if isinstance(mapdict, list):

smapdict = {k: v for k, v in mapdict[0].items()}

elif isinstance(mapdict, dict):

if '' in mapdict.keys() and isinstance(mapdict[''], dict):

smapdict = {k: v for k, v in mapdict[''].items()}

else:

smapdict = {k: v for k, v in mapdict.items()}

# no 'else' handling a LookupError here ; this case is covered by the first if/elif/else block

# case 2: list or dictionary or dictionary of numbered encodings

elif isinstance(p, int):

# if mapdict is a list, we shall align the parameter (starting from 1) as an index (starting from 0)

if isinstance(mapdict, list):

p -= 1

if isinstance(mapdict, list) and 0 <= p < len(mapdict) or \

isinstance(mapdict, dict) and p in mapdict.keys():

smapdict = {k: v for k, v in mapdict[p].items()}

else:

raise LookupError("Bad parameter for encoding '{}': '{}'".format(ename, p))

# case 3: dictionary of regex-selected encoding mappings

elif isinstance(mapdict, dict) and isinstance(list(mapdict.values())[0], dict):

tmp = None

for r, d in mapdict.items():

if r == '': # this is already handled in case 1 ; anyway, an empty regex always matches, hence

continue # it must be excluded

if re.match(r, p):

tmp = d

break

if tmp is None:

raise LookupError("Bad parameter for encoding '{}': '{}'".format(ename, p))

smapdict = tmp

# case 4: encoding characters translation

else:

# collect base tokens in order of appearance in the mapping dictionary

base_tokens = ""

for _, c in sorted(mapdict.items()):

for t in c:

for st in t:

if st not in base_tokens:

base_tokens += st

if " " not in sep:

base_tokens = base_tokens.replace(" ", "")

if len(p) > 0 and p[0] in "-_" and len(p[1:]) == len(set(p[1:])) == len(base_tokens):

p = p[1:]

if len(p) == len(set(p)) == len(base_tokens):

t = maketrans(base_tokens, p)

for k, v in smapdict.items():

smapdict[k] = [x.translate(t) for x in v] if isinstance(v, list) else v.translate(t)

else:

raise LookupError("Bad parameter for encoding '{}': '{}'".format(ename, p))

if ignore_case is not None:

cases = ["upper", "lower"]

case_d = cases[any(c in str(list(smapdict.values())) for c in "abcdefghijklmnopqrstuvwxyz")]

case_e = cases[any(c in str(list(smapdict.keys())) for c in "abcdefghijklmnopqrstuvwxyz")]

i = ignore_case

smapdict = {getattr(k, case_e)() if i in ["both", "encode"] else k: \

([getattr(x, case_d)() for x in v] if isinstance(v, list) else getattr(v, case_d)()) \

if i in ["both", "decode"] else v for k, v in smapdict.items()}

if decode:

tmp = {}

# this has a meaning for encoding maps that could have clashes in encoded chars (e.g. Bacon's cipher ;

# I => abaaa but also J => abaaa, with the following, we keep I instead of letting J overwrite it)

for k, v in sorted(smapdict.items()):

if not isinstance(v, list):

v = [v]

for x in v:

if x not in tmp.keys():

tmp[x] = k

smapdict, cs = tmp, reduce(lambda acc, x: acc + x, tmp.keys())

kwargs['strip_lines'], kwargs['strip_crlf'] = "\n" not in set(cs), "\r\n" not in cs

# this allows to avoid an error with Python2 in the "for i, c in enumerate(parts)" loop

if '' not in smapdict.keys():

smapdict[''] = ""

# determine token and result lengths

tmaxlen = max(map(len, smapdict.keys()))

tminlen = max(1, min(map(len, set(smapdict.keys()) - {''})))

l = []

for x in smapdict.values():

getattr(l, ["append", "extend"][isinstance(x, list)])(x)

rminlen = max(1, min(map(len, set(l) - {''})))

# generic encoding/decoding function for map encodings

def code(text, errors="strict"):

icase = ignore_case == "both" or \

decode and ignore_case == "decode" or \

not decode and ignore_case == "encode"

if icase:

case = case_d if decode else case_e

if no_error:

errors = "leave"

text = ensure_str(text)

if not decode:

if intype == "bin":

text = "".join("{:0>8}".format(bin(ord(c))[2:]) for c in text)

elif intype == "ord":

text = "".join(str(ord(c)).zfill(3) for c in text)

r = ""

lsep = "" if decode else sep if len(sep) <= 1 else sep[0]

kind = ["character", "token"][tmaxlen > 1]

error_func = handle_error(ename, errors, lsep, repl_char, rminlen, decode, kind)

# get the value from the mapping dictionary, trying the token with its inverted case if relevant

def __get_value(token, position, case_changed=False):

try:

result = smapdict[token]

except KeyError:

if icase and not case_changed:

token_inv_case = getattr(token, case)()

return __get_value(token_inv_case, position, True)

return error_func(token, position)

if isinstance(result, list):

result = result[0]

return result + lsep

# if a separator is defined, rely on it by splitting the input text

if decode and len(sep) > 0:

for i, c in enumerate(re.split("[" + sep + "]", text)):

r += __get_value(c, i)

# otherwise, move through the text using a cursor for tokenizing it ; this allows defining more complex

# encodings with variable token lengths

else:

cursor, bad = 0, ""

while cursor < len(text):

token = text[cursor:cursor+1]

for l in range(tminlen, tmaxlen + 1):

token = text[cursor:cursor+l]

if token in smapdict.keys() or icase and getattr(token, case)() in smapdict.keys():

r += __get_value(token, cursor)

cursor += l

break

else:

# collect bad chars and only move the cursor one char to the right

bad += text[cursor]

cursor += 1

# if the number of bad chars is the minimum token length, consume it and start a new buffer

if len(bad) == tminlen or errors == "leave":

posn = cursor - len(bad)

r += error_func(bad, posn)

bad = ""

if decode:

if outype in ["bin", "ord"]:

tmp, r = "", r.replace(lsep, "")

step = [3, 8][outype == "bin"]

for i in range(0, len(r), step):

s = r[i:i+step]

try:

tmp += chr(int(s, 2) if outype == "bin" else int(s))

except ValueError:

if len(s) > 0:

tmp += "[" + s + "]"

r = tmp + lsep

return r[:len(r)-len(lsep)], len(b(text))

return code

if re.search(r"\([^(?:)]", kwargs.get('pattern', "")) is None:

# in this case, there is no capturing group for parametrization

return _wrapper(None)

return _wrapper

glob = currentframe().f_back.f_globals

kwargs['category'] = glob['__file__'].split(os.path.sep)[-2].rstrip("s")

kwargs['examples'] = kwargs.get('examples', glob.get('__examples__'))

kwargs['encmap'] = encmap

kwargs['repl_char'] = repl_char

kwargs['sep'] = sep

kwargs['ignore_case'] = ignore_case

kwargs['no_error'] = no_error

kwargs['intype'] = intype

kwargs['outype'] = outype

kwargs['module'] = glob.get('__name__')

try:

if isinstance(encmap, dict):

smapdict = {k: v for k, v in encmap.items()}

elif isinstance(encmap, list) and isinstance(encmap[0], dict):

smapdict = {k: v for k, v in encmap[0].items()}

kwargs['repl_minlen'] = i = max(1, min(map(len, set(smapdict.values()) - {''})))

kwargs['repl_minlen_b'] = max(1, min(map(len, map(b, set(smapdict.values()) - {''}))))

except:

pass

""" Use the search function to get the matching encodings and provide examples of valid encoding names. """

e = []

for name in search(encoding):

for search_function in __codecs_registry:

n = search_function.__name__

if name in [n, n.replace("_", "-")]:

temp = []

for s in generate_strings_from_regex(search_function.__pattern__, yield_max=16*number):

temp.append(s)

random.shuffle(temp)

i = 0

while i < min(number, len(temp)):

if not temp[i].isdigit():

try:

lookup(temp[i], False)

e.append(temp[i])

except LookupError:

pass

i += 1

for alias, codec in ALIASES.items():

if name == codec:

if codec not in e:

e.append(codec)

if not alias.isdigit():

e.append(alias)

random.shuffle(e)

""" Get a list of all codec categories. """

c = CODECS_CATEGORIES

root = os.path.dirname(__file__)

for d in os.listdir(root):

if os.path.isdir(os.path.join(root, d)) and not d.startswith("__"):

c.append(d.rstrip("s"))

# particular category, hardcoded from base/_base.py

c += ["base-generic"]

""" Get a list of all codecs. """

# if "non-native" is in the input list, extend the list with the whole categories but "native"

categories, exclude = list(categories), []

for c in categories[:]:

if c == "non-native":

for c in CODECS_CATEGORIES:

if c == "native" or c in categories:

continue

categories.append(c)

categories.remove("non-native")

if c.startswith("~"):

exclude.append(c[1:])

categories.remove(c)

try:

categories.remove(c[1:])

except ValueError:

pass

# now, filter codecs according to the input list of categories

enc = []

if (len(categories) == 0 or "native" in categories) and "native" not in exclude:

for a in set(ALIASES.values()):

try:

ci = __orig_lookup(a)

except LookupError:

continue

if lookup(a) is ci:

enc.append(ci.name)

for search_function in CODECS_OVERWRITTEN + __codecs_registry:

name = search_function.__name__.replace("_", "-")

p = search_function.__pattern__

ci = search_function(name) if p is None else search_function(generate_string_from_regex(p))

c = "other" if ci is None else ci.parameters['category']

if (len(categories) == 0 or c in categories) and c not in exclude:

enc.append(name)

for category in categories:

if category not in CODECS_CATEGORIES:

raise ValueError("Category '%s' does not exist" % category)

""" Remove all search functions matching the input encoding name from codext's local registry or any macro with the

global __codecs_registry, MACROS, PERS_MACROS

tbr = []

for search_function in __codecs_registry:

if search_function(name) is not None:

tbr.append(search_function)

for search_function in tbr:

__codecs_registry.remove(search_function)

try:

del MACROS[name]

except KeyError:

pass

try:

del PERS_MACROS[name]

with open(PERS_MACROS_FILE, 'w') as f:

json.dump(PERS_MACROS, f, indent=2)

except KeyError:

pass

for s in ["En", "De"]:

try:

delattr(builtins, "%s%scodeError" % (name.capitalize(), s))

except AttributeError:

""" Reset codext's local registry of search functions and macros. """

global __codecs_registry, CODECS_REGISTRY, MACROS, PERS_MACROS

clear()

d = os.path.dirname(__file__)

for pkg in sorted(os.listdir(d)):

if pkg.startswith("_") or not os.path.isdir(os.path.join(d, pkg)):

continue

reload(import_module("codext." + pkg))

# backup codext's registry

if CODECS_REGISTRY is None:

CODECS_REGISTRY = __codecs_registry[:]

# restore codext's registry

else:

__codecs_registry = CODECS_REGISTRY[:]

# restore codext's embedded set of macros

with open(os.path.join(os.path.dirname(__file__), "macros.json")) as f:

MACROS = json.load(f)

# reload personal set of macros

PERS_MACROS = {}

if os.path.exists(PERS_MACROS_FILE):

with open(PERS_MACROS_FILE) as f:

""" Non-crashing bytes conversion function. """

if PY3:

try:

return s.encode("latin-1")

except:

pass

try:

return s.encode("utf-8")

except:

pass

""" Similar to six.ensure_str. Adapted here to avoid messing up with six version errors. """

if not PY3 and isinstance(s, text_type):

return s.encode(encoding, errors)

elif PY3 and isinstance(s, binary_type):

try:

return s.decode(encoding, errors)

except:

return s.decode("latin-1")

""" This decorator ensures that the first output of f will have the same text format as the first input (str or

@wraps(f)

def _wrapper(*args, **kwargs):

a0 = args[0]

a0_isb = isb(a0)

a0 = ensure_str(a0) if iss(a0) or a0_isb else a0

r = f(a0, *args[1:], **kwargs)

# special case: input is in bytes ; ensure that the returned length is this of the bytes, not this processed by

# the decode/encode function

if isinstance(r, (tuple, list)) and isinstance(r[1], int) and a0_isb:

r = tuple([list(r)[0]] + [len(args[0])] + list(r)[2:])

return (fix(r[0], args[0]), ) + r[1:] if isinstance(r, (tuple, list)) else fix(r, args[0])

""" This function generates an alphabet from the given mask. The style used is similar to Hashcat ; group keys are

i, alphabet = 0, ""

while i < len(mask):

c = mask[i]

if c == "?" and i < len(mask) - 1 and mask[i+1] in MASKS.keys():

for c in MASKS[mask[i+1]]:

if c not in alphabet:

alphabet += c

i += 1

elif c not in alphabet:

alphabet += c

i += 1

""" This shortcut function allows to handle error modes given some tuning parameters.

exc = "%s%scodeError" % (exc_name(ename), ["En", "De"][decode])

def _handle_error(token, position, output="", eename=None):

""" This handles an encoding/decoding error according to the selected handling mode.

if errors == "strict":

msg = "'%s' codec can't %scode %s '%s' in %s %d"

token = token[:7] + "..." if len(token) > 10 else token

err = getattr(builtins, exc)(msg % (eename or ename, ["en", "de"][decode], kind, token, item, position))

err.output = output

err.__cause__ = err

raise err

elif errors == "leave":

return token + sep

elif errors == "replace":

return repl_char * repl_minlen + sep

elif errors == "ignore":

return ""

else:

raise ValueError("Unsupported error handling '{}'".format(errors))

""" Custom encode function relying on the hooked lookup function. """

n, m = 1, re.search(r"\[(\d+)\]$", encoding)

if m:

n = int(m.group(1))

encoding = re.sub(r"\[(\d+)\]$", "", encoding)

ci = lookup(encoding)

for i in range(n):

obj = ci.encode(obj, errors)[0]

""" Hooked lookup function for searching first for codecs in the local registry of this module. """

# first, try to match the given encoding with codecs' search functions

for search_function in CODECS_OVERWRITTEN + __codecs_registry:

codecinfo = search_function(encoding)

if codecinfo is not None:

return codecinfo

# then, if a codec name was given, generate an encoding name from its pattern and get the CodecInfo

for search_function in CODECS_OVERWRITTEN + __codecs_registry:

if search_function.__name__.replace("_", "-") == encoding or \

encoding in getattr(search_function, "__aliases__", []):

codecinfo = search_function(generate_string_from_regex(search_function.__pattern__))

if codecinfo is not None:

return codecinfo

# finally, get a CodecInfo with the original lookup function and refine it with a dictionary of parameters

try:

ci = __orig_lookup(encoding)

ci.parameters = {'category': "native", 'module': "codecs", 'name': ALIASES.get(ci.name, ci.name)}

return ci

except LookupError:

if not macro:

raise

try: