File: noxfile.py

"""
The file for Nox
"""
from typing import Any

import nox
from nox.sessions import Session

locations = "ciphey/", "tests/", "docs/"
nox.options.sessions = ["tests"]
package = "ciphey"


def install_with_constraints(session: Session, *args: str, **kwargs: Any) -> None:
    """Install packages constrained by Poetry's lock file.
    This function is a wrapper for nox.sessions.Session.install. It
    invokes pip to install packages inside of the session's virtualenv.
    Additionally, pip is passed a constraints file generated from
    Poetry's lock file, to ensure that the packages are pinned to the
    versions specified in poetry.lock. This allows you to manage the
    packages as Poetry development dependencies.
    Arguments:
        session: The Session object.
        args: Command-line arguments for pip.
        kwargs: Additional keyword arguments for Session.install.
    """
    session.run(
        "poetry",
        "export",
        "--dev",
        "--format=requirements.txt",
        "--output=requirements.txt",
        external=True,
    )
    session.install("--constraint=requirements.txt", *args, **kwargs)


# noxfile.py
@nox.session
def black(session):
    args = session.posargs or locations
    session.install("black")
    session.run("black", *args)


@nox.session(python="3.8")
def coverage(session: Session) -> None:
    """Upload coverage data."""
    install_with_constraints(session, "coverage[toml]", "codecov")
    session.run("pip3", "install", "cipheydists")
    session.run("coverage", "xml", "--fail-under=0")
    session.run("codecov", *session.posargs)


# noxfile.py
@nox.session
def docs(session: Session) -> None:
    """Build the documentation."""
    install_with_constraints(session, "sphinx")
    session.run("sphinx-build", "docs", "docs/_build")


@nox.session
def tests(session):
    session.run("pip3", "install", "cipheydists")
    session.run("poetry", "install", external=True)
    session.run("poetry", "run", "pytest", "--cov=ciphey")



File: tools/freq_analysis.py

import json
import sys

import cipheycore

data = sys.stdin.read()

analysis = cipheycore.analyse_string(data)

print(json.dumps({i: j / len(data) for i, j in analysis.freqs.items()}))



File: ciphey/mathsHelper.py

"""
 ██████╗██╗██████╗ ██╗  ██╗███████╗██╗   ██╗
██╔════╝██║██╔══██╗██║  ██║██╔════╝╚██╗ ██╔╝
██║     ██║██████╔╝███████║█████╗   ╚████╔╝
██║     ██║██╔═══╝ ██╔══██║██╔══╝    ╚██╔╝
╚██████╗██║██║     ██║  ██║███████╗   ██║
© Brandon Skerritt
Github: brandonskerritt

Class to provide helper functions for mathematics
(oh, not entirely mathematics either. Some NLP stuff and sorting dicts. It's just a helper class
)
"""

from collections import OrderedDict
from string import punctuation
from typing import Optional

import logging
from rich.logging import RichHandler


class mathsHelper:
    """Class to provide helper functions for mathematics and other small things"""

    def __init__(self):
        # ETAOIN is the most popular letters in order
        self.ETAOIN = "ETAOINSHRDLCUMWFGYPBVKJXQZ"
        self.LETTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"

    @staticmethod
    def gcd(a, b) -> int:
        """Greatest common divisor.

        The Greatest Common Divisor of a and b using Euclid's Algorithm.

        Args:
            a -> num 1
            b -> num 2

        Returns:
            Returns  GCD(a, b)

        """
        # Return
        while a != 0:
            a, b = b % a, a
        return b

    @staticmethod
    def mod_inv(a: int, m: int) -> Optional[int]:
        """
        Returns the modular inverse of a mod m, or None if it does not exist.

        The modular inverse of a is the number a_inv that satisfies the equation
        a_inv * a mod m === 1 mod m

        Note: This is a naive implementation, and runtime may be improved in several ways.
        For instance by checking if m is prime to perform a different calculation,
        or by using the extended euclidean algorithm.
        """
        for i in range(1, m):
            if (m * i + 1) % a == 0:
                return (m * i + 1) // a
        return None

    @staticmethod
    def percentage(part: float, whole: float) -> float:
        """Returns percentage.

        Just a normal algorithm to return the percent.

        Args:
            part -> part of the whole number
            whole -> the whole number

        Returns:
            Returns the percentage of part to whole.

        """
        if part <= 0 or whole <= 0:
            return 0
        # works with percentages
        return 100 * float(part) / float(whole)

    def sort_prob_table(self, prob_table: dict) -> dict:
        """Sorts the probability table.

        Sorts a dictionary of dictionaries (and all the sub-dictionaries).

        Args:
            prob_table -> The probability table returned by the neural network to sort.

        Returns:
            Returns the prob_table, but sorted.

        """
        # for each object: prob table in dictionary
        max_overall: int = 0
        max_dict_pair: dict = {}
        highest_key = None
        empty_dict: dict = {}
        # sorts the prob table before we find max, and converts it to order dicts
        for key, value in prob_table.items():
            prob_table[key] = self.new_sort(value)
            prob_table[key] = dict(prob_table[key])

        # gets maximum key then sets it to the front
        counter_max: int = 0
        counter_prob: int = len(prob_table)
        while counter_max < counter_prob:
            max_overall = 0
            highest_key = None
            logging.debug(
                f"Running while loop in sort_prob_table, counterMax is {counter_max}"
            )
            for key, value in prob_table.items():
                logging.debug(f"Sorting {key}")
                maxLocal = 0
                # for each item in that table
                for key2, value2 in value.items():
                    logging.debug(
                        f"Running key2 {key2}, value2 {value2} for loop for {value.items()}"
                    )
                    maxLocal = maxLocal + value2
                    logging.debug(
                        f"MaxLocal is {maxLocal} and maxOverall is {max_overall}"
                    )
                    if maxLocal > max_overall:
                        logging.debug(f"New max local found {maxLocal}")
                        # because the dict doesn't reset
                        max_dict_pair = {}
                        max_overall = maxLocal
                        # so eventually, we get the maximum dict pairing?
                        max_dict_pair[key] = value
                        highest_key = key
                        logging.debug(f"Highest key is {highest_key}")
                # removes the highest key from the prob table
            logging.debug(
                f"Prob table is {prob_table} and highest key is {highest_key}"
            )
            logging.debug(f"Removing {prob_table[highest_key]}")
            del prob_table[highest_key]
            logging.debug(f"Prob table after deletion is {prob_table}")
            counter_max += 1
            empty_dict = {**empty_dict, **max_dict_pair}

        # returns the max dict (at the start) with the prob table
        # this way, it should always work on most likely first.
        logging.debug(
            f"The prob table is {prob_table} and the maxDictPair is {max_dict_pair}"
        )
        logging.debug(f"The new sorted prob table is {empty_dict}")
        return empty_dict

    @staticmethod
    def new_sort(new_dict: dict) -> dict:
        """Uses OrderedDict to sort a dictionary.

        I think it's faster than my implementation.

        Args:
            new_dict -> the dictionary to sort

        Returns:
            Returns the dict, but sorted.

        """
        # (f"d is {d}")
        logging.debug(f"The old dictionary before new_sort() is {new_dict}")
        sorted_i = OrderedDict(
            sorted(new_dict.items(), key=lambda x: x[1], reverse=True)
        )
        logging.debug(f"The dictionary after new_sort() is {sorted_i}")
        # sortedI = sort_dictionary(x)
        return sorted_i

    @staticmethod
    def is_ascii(s: str) -> bool:
        """Returns the boolean value if is_ascii is an ascii char.

        Does what it says on the tree. Stolen from
        https://stackoverflow.com/questions/196345/how-to-check-if-a-string-in-python-is-in-ascii

        Args:
            s -> the char to check.

        Returns:
            Returns the boolean of the char.

        """

        return bool(lambda s: len(s) == len(s.encode()))

    @staticmethod
    def strip_punctuation(text: str) -> str:
        """Strips punctuation from a given string.

        Uses string.punctuation.

        Args:
            text -> the text to strip punctuation from.

        Returns:
            Returns string without punctuation.
        """
        text: str = (str(text).translate(str.maketrans("", "", punctuation))).strip(
            "\n"
        )
        return text



File: ciphey/__init__.py

from . import basemods, common, iface
from .ciphey import decrypt



File: ciphey/common.py

"""Some useful adapters"""
from typing import Any


def id_lambda(value: Any):
    """
    A function used in dynamic class generation that abstracts away a constant return value (like in getName)
    """
    return lambda *args: value


def fix_case(target: str, base: str) -> str:
    """Returns the lower-case string target with the case of base"""
    ret = "".join(
        [
            target[i].upper() if base[i].isupper() else target[i]
            for i in range(len(target))
        ]
    )
    return "".join(
        [
            target[i].upper() if base[i].isupper() else target[i]
            for i in range(len(target))
        ]
    )



File: ciphey/__main__.py

#! /usr/bin/env python3

"""
Ciphey: https://github.com/Ciphey/Ciphey
"""

import platform
import sys

if __name__ == "__main__":
    major = sys.version_info[0]
    minor = sys.version_info[1]

    python_version = (
        str(sys.version_info[0])
        + "."
        + str(sys.version_info[1])
        + "."
        + str(sys.version_info[2])
    )

    if major != 3 or minor < 6:
        print(
            f"Ciphey requires Python 3.6+, you are using {python_version}. Please install a higher Python version. https://www.python.org/downloads/"
        )
        print(
            "Alternatively, visit our Discord and use the Ciphey bot in #bots http://discord.skerritt.blog"
        )
        sys.exit(1)
    if platform.system() == "Windows":
        if minor > 8:
            print(
                "Ciphey does not currently support Python 3.9 on Windows. Please use the Discord bot at http://discord.skerritt.blog"
            )
            sys.exit(1)

        if sys.maxsize > 2 ** 32 is False:
            print(
                "You are using Python 32 bit and Windows, Ciphey does not support this. Please upgrade to Python 64-bit here https://www.python.org/downloads/"
            )
            sys.exit(1)
    from .ciphey import main

    main()



File: ciphey/ciphey.py

"""
 ██████╗██╗██████╗ ██╗  ██╗███████╗██╗   ██╗
██╔════╝██║██╔══██╗██║  ██║██╔════╝╚██╗ ██╔╝
██║     ██║██████╔╝███████║█████╗   ╚████╔╝
██║     ██║██╔═══╝ ██╔══██║██╔══╝    ╚██╔╝
╚██████╗██║██║     ██║  ██║███████╗   ██║
https://github.com/Ciphey
https://github.com/Ciphey/Ciphey/wiki

The cycle goes:
main -> argparsing (if needed) -> call_encryption -> new Ciphey object -> decrypt() -> produceProbTable ->
one_level_of_decryption -> decrypt_normal
"""
import os
import warnings
from typing import Any, Optional, Union

import click
from appdirs import AppDirs
import logging
from rich.logging import RichHandler
from rich.console import Console

from . import iface

warnings.filterwarnings("ignore")

console = Console()


def decrypt(config: iface.Config, ctext: Any) -> Union[str, bytes]:
    """A simple alias for searching a ctext and makes the answer pretty"""
    res: Optional[iface.SearchResult] = config.objs["searcher"].search(ctext)
    if res is None:
        return "Failed to crack"
    if config.verbosity < 0:
        return res.path[-1].result.value
    else:
        return iface.pretty_search_results(res)


def get_name(ctx, param, value):
    # reads from stdin if value was not supplied
    if not value and not click.get_text_stream("stdin").isatty():
        click.get_text_stream("stdin").read().strip()
        return click.get_text_stream("stdin").read().strip()
    else:
        return value


def print_help(ctx):
    # prints help menu
    # if no arguments are passed
    click.echo(ctx.get_help())
    ctx.exit()


@click.command()
@click.option(
    "-t",
    "--text",
    help="The ciphertext you want to decrypt.",
    type=str,
)
@click.option(
    "-q", "--quiet", help="Decrease verbosity", type=int, count=True, default=None
)
@click.option(
    "-g",
    "--greppable",
    help="Only print the answer (useful for grep)",
    type=bool,
    is_flag=True,
    default=None,
)
@click.option("-v", "--verbose", count=True, type=int)
@click.option("-C", "--checker", help="Use the given checker", default=None)
@click.option(
    "-c",
    "--config",
    help="Uses the given config file. Defaults to appdirs.user_config_dir('ciphey', 'ciphey')/'config.yml'",
)
@click.option("-w", "--wordlist", help="Uses the given wordlist")
@click.option(
    "-p",
    "--param",
    help="Passes a parameter to the language checker",
    multiple=True,
)
@click.option(
    "-l",
    "--list-params",
    help="List the parameters of the selected module",
    type=bool,
)
@click.option(
    "--searcher",
    help="Select the searching algorithm to use",
)
# HARLAN TODO XXX
# I switched this to a boolean flag system
# https://click.palletsprojects.com/en/7.x/options/#boolean-flags
# True for bytes input, False for str
@click.option(
    "-b",
    "--bytes",
    help="Forces ciphey to use binary mode for the input",
    is_flag=True,
    default=None,
)
@click.option(
    "--default-dist",
    help="Sets the default character/byte distribution",
    type=str,
    default=None,
)
@click.option(
    "-m",
    "--module",
    help="Adds a module from the given path",
    type=click.Path(),
    multiple=True,
)
@click.option(
    "-A",
    "--appdirs",
    help="Print the location of where Ciphey wants the settings file to be",
    type=bool,
    is_flag=True,
)
@click.option("-f", "--file", type=click.File("rb"), required=False)
@click.argument("text_stdin", callback=get_name, required=False)
def main(**kwargs):
    """Ciphey - Automated Decryption Tool

    Documentation:
    https://github.com/Ciphey/Ciphey/wiki\n
    Discord (support here, we're online most of the day):
    http://discord.skerritt.blog\n
    GitHub:
    https://github.com/ciphey/ciphey\n

    Ciphey is an automated decryption tool using smart artificial intelligence and natural language processing. Input encrypted text, get the decrypted text back.

    Examples:\n
        Basic Usage: ciphey -t "aGVsbG8gbXkgbmFtZSBpcyBiZWU="

    """

    """Function to deal with arguments. Either calls with args or not. Makes Pytest work.

    It gets the arguments in the function definition using locals()
    if withArgs is True, that means this is being called with command line args
    so go to arg_parsing() to get those args
    we then update locals() with the new command line args and remove "withArgs"
    This function then calls call_encryption(**result) which passes our dict of args
    to the function as its own arguments using dict unpacking.
        Returns:
            The output of the decryption.
    """

    # if user wants to know where appdirs is
    # print and exit
    if "appdirs" in kwargs and kwargs["appdirs"]:
        dirs = AppDirs("Ciphey", "Ciphey")
        path_to_config = dirs.user_config_dir
        print(
            f"The settings.yml file should be at {os.path.join(path_to_config, 'settings.yml')}"
        )
        return None

    # Now we create the config object
    config = iface.Config()

    # Load the settings file into the config
    load_msg: str
    cfg_arg = kwargs["config"]
    if cfg_arg is None:
        # Make sure that the config dir actually exists
        os.makedirs(iface.Config.get_default_dir(), exist_ok=True)
        config.load_file(create=True)
        load_msg = f"Opened config file at {os.path.join(iface.Config.get_default_dir(), 'config.yml')}"
    else:
        config.load_file(cfg_arg)
        load_msg = f"Opened config file at {cfg_arg}"

    # Load the verbosity, so that we can start logging
    verbosity = kwargs["verbose"]
    quiet = kwargs["quiet"]
    if verbosity is None:
        if quiet is not None:
            verbosity = -quiet
    elif quiet is not None:
        verbosity -= quiet
    if kwargs["greppable"] is not None:
        verbosity -= 999
    # Use the existing value as a base
    config.verbosity += verbosity
    config.update_log_level(config.verbosity)
    logging.info(load_msg)
    logging.debug(f"Got cmdline args {kwargs}")

    # Now we load the modules
    module_arg = kwargs["module"]
    if module_arg is not None:
        config.modules += list(module_arg)

    # We need to load formats BEFORE we instantiate objects
    if kwargs["bytes"] is not None:
        config.update_format("bytes")

    # Next, load the objects
    params = kwargs["param"]
    if params is not None:
        for i in params:
            key, value = i.split("=", 1)
            parent, name = key.split(".", 1)
            config.update_param(parent, name, value)
    config.update("checker", kwargs["checker"])
    config.update("searcher", kwargs["searcher"])
    config.update("default_dist", kwargs["default_dist"])

    config.complete_config()

    logging.debug(f"Command line opts: {kwargs}")
    logging.debug(f"Config finalised: {config}")

    # Finally, we load the plaintext
    if kwargs["text"] is None:
        if kwargs["file"] is not None:
            kwargs["text"] = kwargs["file"].read()
        elif kwargs["text_stdin"] is not None:
            kwargs["text"] = kwargs["text_stdin"]
        else:
            # else print help menu
            print("[bold red]Error. No inputs were given to Ciphey. [bold red]")

            @click.pass_context
            def all_procedure(ctx):
                print_help(ctx)

            all_procedure()

            return None

    if issubclass(config.objs["format"], type(kwargs["text"])):
        pass
    elif config.objs["format"] == str and isinstance(kwargs["text"], bytes):
        kwargs["text"] = kwargs["text"].decode("utf-8")
    elif config.objs["format"] == bytes and isinstance(kwargs["text"], str):
        kwargs["text"] = kwargs["text"].encode("utf-8")
    else:
        raise TypeError(f"Cannot load type {config.format} from {type(kwargs['text'])}")

    result: Optional[str]

    # if debug or quiet mode is on, run without spinner
    if config.verbosity != 0:
        result = decrypt(config, kwargs["text"])
    else:
        # else, run with spinner if verbosity is 0
        with console.status("[bold green]Thinking...", spinner="moon") as status:
            config.set_spinner(status)
            result = decrypt(config, kwargs["text"])
    if result is None:
        result = "Could not find any solutions."

    console.print(result)



File: ciphey/basemods/__init__.py

from . import Checkers, Crackers, Decoders, Resources, Searchers



File: ciphey/basemods/Checkers/ezcheck.py

from typing import Dict, List, Optional

from ciphey.iface import Checker, Config, ParamSpec, T, registry

from .brandon import Brandon
from .format import JsonChecker
from .human import HumanChecker
from .quadgrams import Quadgrams
from .regex import RegexList
from .what import What


@registry.register
class EzCheck(Checker[str]):
    """
    This object is effectively a prebuilt quorum (with requirement 1) of common patterns, followed by a human check
    """

    def check(self, text: str) -> Optional[str]:
        for checker in self.checkers:
            res = checker.check(text)
            if (
                res is not None
                and (self.decider is None or self.decider.check(text)) is not None
            ):
                return res
        return None

    def getExpectedRuntime(self, text: T) -> float:
        return sum(
            i.getExpectedRuntime(text) for i in self.checkers
        ) + self.decider.getExpectedRuntime(text)

    def __init__(self, config: Config):
        super().__init__(config)

        self.checkers: List[Checker[str]] = []
        # Disable human checker for automated systems
        if config.verbosity >= 0:
            self.decider = config(HumanChecker)
        else:
            self.decider = None

        # We need to modify the config for each of the objects

        # First PyWhat, as it's the fastest
        self.checkers.append(config(What))

        # Next, the json checker
        self.checkers.append(config(JsonChecker))

        # Second to last, the quadgrams checker
        self.checkers.append(config(Quadgrams))

        # Finally, the Brandon checker, as it is the slowest
        self.checkers.append(config(Brandon))

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        pass



File: ciphey/basemods/Checkers/what.py

from typing import Dict, Optional

from ciphey.iface import Checker, Config, ParamSpec, T, registry
import logging
from rich.logging import RichHandler
from pywhat import identifier
from rich.console import Console

console = Console()


@registry.register
class What(Checker[str]):

    """
    Uses PyWhat to determine plaintext with regexes
    https://github.com/bee-san/pyWhat
    """

    def check(self, ctext: T) -> Optional[str]:
        logging.debug("Trying PyWhat checker")
        returned_regexes = self.id.identify(ctext)
        if returned_regexes["Regexes"]:
            matched_regex = returned_regexes["Regexes"]['text'][0]["Regex Pattern"]

            ret = f'The plaintext is a [yellow]{matched_regex["Name"]}[/yellow]'
            human = (
                f'\nI think the plaintext is a [yellow]{matched_regex["Name"]}[/yellow]'
            )

            if "Description" in matched_regex and matched_regex["Description"]:
                s = matched_regex["Description"]
                # lowercases first letter so it doesn't look weird
                s = f", which is {s[0].lower() + s[1:]}\n"
                ret += s
                human += s

            # if URL is attached, include that too.
            if "URL" in matched_regex and matched_regex["URL"]:
                link = matched_regex["URL"] + ctext.replace(" ", "")
                ret += f"\nClick here to view in browser [#CAE4F1][link={link}]{link}[/link][/#CAE4F1]\n"

            # If greppable mode is on, don't print this
            if self.config.verbosity >= 0:
                # Print with full stop
                console.print(human)
            return ret
        return None

    def getExpectedRuntime(self, text: T) -> float:
        # TODO: actually bench this
        return 2e-7 * len(text)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    def __init__(self, config: Config):
        super().__init__(config)
        self.config = config
        self.id = identifier.Identifier()



File: ciphey/basemods/Checkers/__init__.py

from . import any, brandon, ezcheck, format, human, quadgrams, quorum, regex, what



File: ciphey/basemods/Checkers/format.py

import json
from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Checker, Config, ParamSpec, T, registry


@registry.register
class JsonChecker(Checker[str]):

    """
    This object is effectively a prebuilt quorum (with requirement 1) of common patterns
    """

    def check(self, text: T) -> Optional[str]:
        logging.debug("Trying json checker")

        # https://github.com/Ciphey/Ciphey/issues/389
        if text.isdigit():
            return None

        try:
            json.loads(text)
            return ""
        except ValueError:
            return None

    def getExpectedRuntime(self, text: T) -> float:
        # TODO: actually bench this
        return 1e-7 * len(text)  # From benchmarks I found online

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        pass



File: ciphey/basemods/Checkers/quorum.py

from typing import Dict, Generic, Optional

from ciphey.iface import Checker, Config, ParamSpec, T, _registry


class Quorum(Generic[T], Checker[T]):
    def check(self, text: T) -> Optional[str]:
        left = self._params().k
        results = []
        for checker in self.checkers:
            results.append(checker.check(text))
            if results[-1] is None:
                continue
            left -= 1
            # Early return check
            if left == 0:
                return str(results)

    def __init__(self, config: Config):
        super().__init__(config)

        if self._params().k is None:
            k = len(self._params()["checker"])
        # These checks need to be separate, to make sure that we do not have zero members
        if self._params().k == 0 or self._params().k > len(self._params()["checker"]):
            raise IndexError(
                "k must be between 0 and the number of checkers (inclusive)"
            )

        self.checkers = []
        for i in self._params()["checker"]:
            # This enforces type consistency
            self.checkers.append(_registry.get_named(i, Checker[T]))

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "checker": ParamSpec(
                req=True, desc="The checkers to be used for analysis", list=True
            ),
            "k": ParamSpec(
                req=False,
                desc="The minimum quorum size. Defaults to the number of checkers",
            ),
        }



File: ciphey/basemods/Checkers/human.py

from typing import Dict, Optional

from ciphey.iface import Checker, Config, ParamSpec, registry
from rich.console import Console
from rich.markup import escape

console = Console()


@registry.register
class HumanChecker(Checker[str]):

    """
    Uses the person's decision to determine plaintext
    """

    def check(self, ctext: str) -> Optional[str]:
        with self._config().pause_spinner_handle():
            response = console.input(
                f"Possible plaintext: [blue bold]{escape(ctext.__repr__())}[/blue bold] ([green]y[/green]/[red]N[/red]): "
            )
        if response == "y":
            return ""
        elif response in ("n", ""):
            return None
        else:
            return self.check(ctext)

    def getExpectedRuntime(self, text: str) -> float:
        return 1  # About a second

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    def __init__(self, config: Config):
        super().__init__(config)



File: ciphey/basemods/Checkers/brandon.py

"""
 ██████╗██╗██████╗ ██╗  ██╗███████╗██╗   ██╗
██╔════╝██║██╔══██╗██║  ██║██╔════╝╚██╗ ██╔╝
██║     ██║██████╔╝███████║█████╗   ╚████╔╝
██║     ██║██╔═══╝ ██╔══██║██╔══╝    ╚██╔╝
╚██████╗██║██║     ██║  ██║███████╗   ██║
© Brandon Skerritt
Github: brandonskerritt

Class to determine whether something is English or not.
1. Calculate the Chi Squared score of a sentence
2. If the score is significantly lower than the average score, it _might_ be English
    2.1. If the score _might_ be English, then take the text and compare it to the sorted dictionary
    in O(n log n) time.
    It creates a percentage of "How much of this text is in the dictionary?"
    The dictionary contains:
        * 20,000 most common US words
        * 10,000 most common UK words (there's no repetition between the two)
        * The top 10,000 passwords
    If the word "Looks like" English (chi-squared) and if it contains English words, we can conclude it is
    very likely English. The alternative is doing the dictionary thing but with an entire 479k word dictionary (slower)
    2.2. If the score is not English, but we haven't tested enough to create an average, then test it against
     the dictionary

Things to optimise:
* We only run the dictionary if it's 20% smaller than the average for chi squared
* We consider it "English" if 45% of the text matches the dictionary
* We run the dictionary if there is less than 10 total chisquared test

How to add a language:
* Download your desired dictionary. Try to make it the most popular words, for example. Place this file into this
 folder with languagename.txt
As an example, this comes built in with english.txt
Find the statistical frequency of each letter in that language.
For English, we have:
self.languages = {
    "English":
    [0.0855, 0.0160, 0.0316, 0.0387, 0.1210,0.0218, 0.0209, 0.0496, 0.0733, 0.0022,0.0081, 0.0421, 0.0253, 0.0717,
    0.0747,0.0207, 0.0010, 0.0633, 0.0673, 0.0894,0.0268, 0.0106, 0.0183, 0.0019, 0.0172,0.0011]
}
In chisquared.py
To add your language, do:
self.languages = {
    "English":
    [0.0855, 0.0160, 0.0316, 0.0387, 0.1210,0.0218, 0.0209, 0.0496, 0.0733, 0.0022,0.0081, 0.0421, 0.0253, 0.0717,
    0.0747,0.0207, 0.0010, 0.0633, 0.0673, 0.0894,0.0268, 0.0106, 0.0183, 0.0019, 0.0172,0.0011]
    "German": [0.0973]
}
In alphabetical order
And you're.... Done! Make sure the name of the two match up
"""
import sys
from math import ceil
from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Checker, Config, ParamSpec, T, registry

sys.path.append("..")
try:
    import mathsHelper as mh
except ModuleNotFoundError:
    import ciphey.mathsHelper as mh


@registry.register
class Brandon(Checker[str]):
    """
    Class designed to confirm whether something is **language** based on how many words of **language** appears
    Call confirmLanguage(text, language)
    * text: the text you want to confirm
    * language: the language you want to confirm

    Find out what language it is by using chisquared.py, the highest chisquared score is the language
    languageThreshold = 45
    if a string is 45% **language** words, then it's confirmed to be english
    """

    def getExpectedRuntime(self, text: T) -> float:
        # TODO: actually work this out
        # TODO its 0.2 seconds on average
        return 1e-4  # 100 µs

    wordlist: set

    def clean_text(self, text: str) -> set:
        """Cleans the text ready to be checked

        Strips punctuation, makes it lower case, turns it into a set separated by spaces, removes duplicate words

        Args:
            text -> The text we use to perform analysis on

        Returns:
            text -> the text as a list, now cleaned

        """
        # makes the text unique words and readable
        text = text.lower()
        text = self.mh.strip_punctuation(text)
        text = text.split(" ")
        text = filter(lambda x: len(x) > 2, text)
        text = set(text)
        return text

    def checker(self, text: str, threshold: float, text_length: int, var: set) -> bool:
        """Given text determine if it passes checker

        The checker uses the variable passed to it. I.E. Stopwords list, 1k words, dictionary

        Args:
            text -> The text to check
            threshold -> at what point do we return True? The percentage of text that is in var before we return True
            text_length -> the length of the text
            var -> the variable we are checking against. Stopwords list, 1k words list, dictionary list.
        Returns:
            boolean -> True for it passes the test, False for it fails the test."""
        if text is None:
            logging.debug("Checker's text is None, so returning False")
            return False
        if var is None:
            logging.debug("Checker's input var is None, so returning False")
            return False

        percent = ceil(text_length * threshold)
        logging.debug(f"Checker's chunks are size {percent}")
        meet_threshold = 0
        location = 0
        end = percent

        if text_length <= 0:
            return False

        while location <= text_length:
            # chunks the text, so only gets THRESHOLD chunks of text at a time
            text = list(text)
            to_analyse = text[location:end]
            logging.debug(f"To analyse is {to_analyse}")
            for word in to_analyse:
                # if word is a stopword, + 1 to the counter
                if word in var:
                    logging.debug(
                        f"{word} is in var, which means I am +=1 to the meet_threshold which is {meet_threshold}"
                    )
                    meet_threshold += 1
                meet_threshold_percent = meet_threshold / text_length
                if meet_threshold_percent >= threshold:
                    logging.debug(
                        f"Returning true since the percentage is {meet_threshold / text_length} and the threshold is {threshold}"
                    )
                    # if we meet the threshold, return True
                    # otherwise, go over again until we do
                    # We do this in the for loop because if we're at 24% and THRESHOLD is 25
                    # we don't want to wait THRESHOLD to return true, we want to return True ASAP
                    return True
            location = end
            end = end + percent
        logging.debug(
            f"The language proportion {meet_threshold_percent} is under the threshold {threshold}"
        )
        return False

    def __init__(self, config: Config):
        # Suppresses warning
        super().__init__(config)
        self.mh = mh.mathsHelper()

        phases = config.get_resource(self._params()["phases"])

        self.thresholds_phase1 = phases["1"]
        self.thresholds_phase2 = phases["2"]
        self.top1000Words = config.get_resource(self._params().get("top1000"))
        self.wordlist = config.get_resource(self._params()["wordlist"])
        self.stopwords = config.get_resource(self._params().get("stopwords"))

        self.len_phase1 = len(self.thresholds_phase1)
        self.len_phase2 = len(self.thresholds_phase2)

    def check(self, text: str) -> Optional[str]:
        """Checks to see if the text is in English

        Performs a decryption, but mainly parses the internal data packet and prints useful information.

        Args:
            text -> The text we use to perform analysis on

        Returns:
            bool -> True if the text is English, False otherwise.

        """
        logging.debug(f'In Language Checker with "{text}"')
        text = self.clean_text(text)
        logging.debug(f'Text split to "{text}"')
        if text == "":
            logging.debug("Returning None from Brandon as the text cleaned is none.")
            return None

        length_text = len(text)

        what_to_use = {}

        # this code decides what checker / threshold to use
        # if text is over or equal to maximum size, just use the maximum possible checker
        what_to_use = self.calculateWhatChecker(
            length_text, self.thresholds_phase1.keys()
        )
        logging.debug(self.thresholds_phase1)
        what_to_use = self.thresholds_phase1[str(what_to_use)]
        # def checker(self, text: str, threshold: float, text_length: int, var: set) -> bool:
        if "check" in what_to_use:
            # perform check 1k words
            result = self.checker(
                text, what_to_use["check"], length_text, self.top1000Words
            )
        elif "stop" in what_to_use:
            # perform stopwords
            result = self.checker(
                text, what_to_use["stop"], length_text, self.stopwords
            )
        elif "dict" in what_to_use:
            result = self.checker(text, what_to_use["dict"], length_text, self.wordlist)
            # If result is None, no point doing it again in phase2
            if not result:
                return None
        else:
            logging.info(f"It is neither stop or check, but instead {what_to_use}")

        # return False if phase 1 fails
        if not result:
            return None
        else:
            what_to_use = self.calculateWhatChecker(
                length_text, self.thresholds_phase2.keys()
            )
            what_to_use = self.thresholds_phase2[str(what_to_use)]
            result = self.checker(text, what_to_use["dict"], length_text, self.wordlist)
        return "" if result else None

    def calculateWhatChecker(self, length_text, key):
        """Calculates what threshold / checker to use

        If the length of the text is over the maximum sentence length, use the last checker / threshold
        Otherwise, traverse the keys backwards until we find a key range that does not fit.
        So we traverse backwards and see if the sentence length is between current - 1 and current
        In this way, we find the absolute lowest checker / percentage threshold.
        We traverse backwards because if the text is longer than the max sentence length, we already know.
        In total, the keys are only 5 items long or so. It is not expensive to move backwards, nor is it expensive to move forwards.

        Args:
            length_text -> The length of the text
            key -> What key we want to use. I.E. Phase1 keys, Phase2 keys.
        Returns:
            what_to_use -> the key of the lowest checker."""

        _keys = list(key)
        _keys = list(map(int, _keys))
        if length_text >= int(_keys[-1]):
            what_to_use = list(key)[_keys.index(_keys[-1])]
        else:
            # this algorithm finds the smallest possible fit for the text
            for counter, i in reversed(list(enumerate(_keys))):
                #  [0, 110, 150]
                if i <= length_text:
                    what_to_use = i
        return what_to_use

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "top1000": ParamSpec(
                desc="A wordlist of the top 1000 words",
                req=False,
                default="cipheydists::list::english1000",
            ),
            "wordlist": ParamSpec(
                desc="A wordlist of all the words",
                req=False,
                default="cipheydists::list::english",
            ),
            "stopwords": ParamSpec(
                desc="A wordlist of StopWords",
                req=False,
                default="cipheydists::list::englishStopWords",
            ),
            "threshold": ParamSpec(
                desc="The minimum proportion (between 0 and 1) that must be in the dictionary",
                req=False,
                default=0.45,
            ),
            "phases": ParamSpec(
                desc="Language-specific phase thresholds",
                req=False,
                default="cipheydists::brandon::english",
            ),
        }



File: ciphey/basemods/Checkers/any.py

from typing import Dict, Optional

from ciphey.iface import Config, ParamSpec, PolymorphicChecker, registry


@registry.register
class Any(PolymorphicChecker):
    """Should only be used for debugging, frankly"""

    def getExpectedRuntime(self, text) -> float:
        return 0  # TODO: actually calculate this

    def __init__(self, config: Config):
        super().__init__(config)

    def check(self, text: str) -> Optional[str]:
        return ""

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        pass



File: ciphey/basemods/Checkers/quadgrams.py

import logging
import re
from math import log10
from typing import Dict, Optional

from ciphey.iface import Checker, Config, ParamSpec, T, Translation, registry
from rich.logging import RichHandler


@registry.register
class Quadgrams(Checker[str]):

    """
    Uses Quadgrams to determine plaintext
    """

    def check(self, ctext: T) -> Optional[str]:
        logging.debug("Trying Quadgrams checker")
        # Capitalize and remove everything that's not a letter
        ctext = re.sub("[^A-Z]", "", ctext.upper())
        quadgrams = self.QUADGRAMS_DICT
        quadgrams_sum = sum(quadgrams.values())
        score = 0
        for key in quadgrams.keys():
            quadgrams[key] = float(quadgrams[key]) / quadgrams_sum
        floor = log10(0.01 / quadgrams_sum)
        for i in range(len(ctext) - 4 + 1):
            # Get all quadgrams from ctext and check if they're in the dict
            # If yes then add the score of those quadgrams to the total score
            if ctext[i : i + 4] in quadgrams:
                score += quadgrams[ctext[i : i + 4]]
            else:
                score += floor
        if len(ctext) > 0:
            score = score / len(ctext)
        logging.info(f"Quadgrams is {score}")
        # The default threshold was found to work the best from lots of testing
        if score > self.threshold:
            return ""
        return None

    def getExpectedRuntime(self, text: T) -> float:
        # TODO: actually bench this
        return 2e-7 * len(text)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "dict": ParamSpec(
                desc="The quadgrams dictionary to use",
                req=False,
                default="cipheydists::dist::quadgrams",
            ),
            "score": ParamSpec(
                desc="The score threshold to use",
                req=False,
                default=0.00011,
            ),
        }

    def __init__(self, config: Config):
        super().__init__(config)
        self.QUADGRAMS_DICT = config.get_resource(self._params()["dict"], Translation)
        self.threshold = float(self._params()["score"])



File: ciphey/basemods/Checkers/gtest.py

from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Checker, Config, ParamSpec, T, registry


@registry.register
class GTestChecker(Checker[str]):

    """
    G-test of fitness, similar to Chi squared.
    """

    def check(self, text: T) -> Optional[str]:
        logging.debug("Trying entropy checker")
        pass

    def getExpectedRuntime(self, text: T) -> float:
        # TODO: actually bench this
        return 4e-7 * len(text)

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        pass



File: ciphey/basemods/Checkers/entropy.py

from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Checker, Config, ParamSpec, T, registry


@registry.register
class Entropy(Checker[str]):

    """
    Uses entropy to determine plaintext
    """

    def check(self, text: T) -> Optional[str]:
        logging.debug("Trying entropy checker")
        pass

    def getExpectedRuntime(self, text: T) -> float:
        # TODO: actually bench this
        # Uses benchmark from Discord
        return 2e-7 * len(text)

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        pass



File: ciphey/basemods/Checkers/regex.py

import re
from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Checker, Config, ParamSpec, T, registry


@registry.register
class Regex(Checker[str]):
    def getExpectedRuntime(self, text: T) -> float:
        return 1e-5  # TODO: actually calculate this

    def __init__(self, config: Config):
        super().__init__(config)
        self.regexes = list(map(re.compile, self._params()["regex"]))
        logging.debug(f"There are {len(self.regexes)} regexes")

    def check(self, text: str) -> Optional[str]:
        for regex in self.regexes:
            logging.debug(f"Trying regex {regex} on {text}")
            res = regex.search(text)
            logging.debug(f"Results: {res}")
            if res:
                return f"Passed with regex {regex}. Want to contribute to Ciphey? Submit your regex here to allow Ciphey to automatically get this next time https://github.com/bee-san/pyWhat/wiki/Adding-your-own-Regex\n"

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "regex": ParamSpec(
                req=True,
                desc="The regex that must be matched (in a substring)",
                list=True,
            )
        }


@registry.register
class RegexList(Checker[str]):
    def getExpectedRuntime(self, text: T) -> float:
        return 1e-5  # TODO: actually calculate this

    def __init__(self, config: Config):
        super().__init__(config)
        self.regexes = []
        for i in self._params()["resource"]:
            self.regexes += [re.compile(regex) for regex in config.get_resource(i)]
        logging.debug(f"There are {len(self.regexes)} regexes")

    def check(self, text: str) -> Optional[str]:
        for regex in self.regexes:
            logging.debug(f"Trying regex {regex} on {text}")
            res = regex.search(text)
            logging.debug(f"Results: {res}")
            if res:
                return f"passed with regex {regex}"

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "resource": ParamSpec(
                req=True,
                desc="A list of regexes that could be matched",
                list=True,
            )
        }



File: ciphey/basemods/Resources/files.py

import csv
import json
from functools import lru_cache
from typing import Dict, Generic, Optional, Set

from ciphey.iface import (
    Config,
    Distribution,
    ParamSpec,
    ResourceLoader,
    T,
    WordList,
    registry,
)


# We can use a generic resource loader here, as we can instantiate it later
@registry.register_multi(WordList, Distribution)
class Json(ResourceLoader):
    def whatResources(self) -> T:
        return self._names

    @lru_cache()
    def getResource(self, name: str) -> T:
        prefix, name = name.split("::", 1)
        return {"wordlist": (lambda js: {js}), "dist": (lambda js: js)}[prefix](
            json.load(open(self._paths[int(name) - 1]))
        )

    @staticmethod
    def getName() -> str:
        return "json"

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {"path": ParamSpec(req=True, desc="The path to a JSON file", list=True)}

    def __init__(self, config: Config):
        super().__init__(config)
        self._paths = self._params()["path"]
        self._names = set(range(1, len(self._paths)))


# We can use a generic resource loader here, as we can instantiate it later
@registry.register_multi(WordList, Distribution)
class Csv(Generic[T], ResourceLoader[T]):
    def whatResources(self) -> Set[str]:
        return self._names

    @lru_cache()
    def getResource(self, name: str) -> T:
        prefix, name = name.split("::", 1)
        return {
            "wordlist": (lambda reader: {i[0] for i in reader}),
            "dist": (lambda reader: {i[0]: float(i[1]) for i in reader}),
        }[prefix](csv.reader(open(self._paths[int(name) - 1])))

    @staticmethod
    def getName() -> str:
        return "csv"

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {"path": ParamSpec(req=True, desc="The path to a CSV file", list=True)}

    def __init__(self, config: Config):
        super().__init__(config)
        self._paths = self._params()["path"]
        self._names = set(range(1, len(self._paths)))



File: ciphey/basemods/Resources/__init__.py

from . import cipheydists, files



File: ciphey/basemods/Resources/cipheydists.py

from functools import lru_cache
from typing import Any, Dict, Optional, Set

import cipheydists
import logging

from ciphey.iface import (
    Config,
    Distribution,
    ParamSpec,
    ResourceLoader,
    Translation,
    WordList,
    registry,
)


@registry.register_multi(WordList, Distribution, Translation)
class CipheyDists(ResourceLoader):
    # _wordlists: Set[str] = frozenset({"english", "english1000", "englishStopWords"})
    # _brandons: Set[str] = frozenset({"english"})
    # _dists: Set[str] = frozenset({"twist"})
    # _translates: Set[str] = frozenset({"morse"})
    _getters = {
        "list": cipheydists.get_list,
        "dist": cipheydists.get_dist,
        "brandon": cipheydists.get_brandon,
        "translate": cipheydists.get_translate,
    }

    def whatResources(self) -> Optional[Set[str]]:
        pass

    @lru_cache()
    def getResource(self, name: str) -> Any:
        logging.debug(f"Loading cipheydists resource {name}")
        prefix, name = name.split("::", 1)
        return self._getters[prefix](name)

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None



File: ciphey/basemods/Searchers/__init__.py

from . import ausearch



File: ciphey/basemods/Searchers/astar.py

import cipheycore


class Node:
    """
    A node has a value associated with it
    Calculated from the heuristic
    """

    def __init__(
        self,
        config,
        h: float = None,
        edges: (any, float) = None,
        ctext: str = None,
    ):
        self.weight = h
        # Edges is a list of other nodes it can connect to
        self.edges = edges
        self.ctext = ctext
        self.h = h
        self.path = []
        self.information_content = config.cache.get_or_update(
            self.text,
            "cipheycore::info_content",
            lambda: cipheycore.info_content(self.ctext),
        )

    def __le__(self, node2):
        # if self is less than other
        return self.x <= node2.x

    def __lt__(self, node2):
        return self.x < node2.x

    def append_edge(self, edge):
        self.edges.append(edge)

    def get_edges(self):
        return self.edges


class Graph:
    # example of adjacency list (or rather map)
    # adjacency_list = {
    # 'A': [('B', 1), ('C', 3), ('D', 7)],
    # 'B': [('D', 5)],
    # 'C': [('D', 12)]
    # }

    def __init__(self, adjacency_list):
        """
        adjacency list: basically the graph
        """
        self.adjacency_list = adjacency_list
        self.original_input = cipheycore.info_content(input)

    def get_neighbors(self, v):
        try:
            return self.adjacency_list[v]
        except KeyError:
            # If we have exhausted the adjacency list
            return []

    # heuristic function with equal values for all nodes
    def heuristic(self, n: Node):
        return n.info_content / self.original_input

    def a_star_algorithm(self, start_node: Node, stop_node: Node):
        # TODO store the graph as an attribute
        # open_list is a list of nodes which have been visited, but who's neighbors
        # haven't all been inspected, starts off with the start node
        # closed_list is a list of nodes which have been visited
        # and who's neighbors have been inspected
        open_list = set([start_node])
        closed_list = set([])

        # g contains current distances from start_node to all other nodes
        # the default value (if it's not found in the map) is +infinity
        g = {}

        g[start_node] = 0

        # parents contains an adjacency map of all nodes
        parents = {}
        parents[start_node] = start_node

        while len(open_list) > 0:
            print(f"The open list is {open_list}")
            n = None

            # find a node with the lowest value of f() - evaluation function
            for v in open_list:
                # TODO if v == decoder, run the decoder
                print(f"The for loop node v is {v}")
                if n is None or g[v] + self.h(v) < g[n] + self.h(n):
                    n = v
                    print(f"The value of n is {n}")

            if n is None:
                print("Path does not exist!")
                return None

            # if the current node is the stop_node
            # then we begin reconstructin the path from it to the start_node
            # NOTE Uncomment this for an exit condition
            # TODO Make it exit if decrypter returns True
            # TODO We need to append the decryption methods to each node
            # So when we reconstruct the path we can reconstruct the decryptions
            # used
            if n == stop_node:
                print("n is the stop node, we are stopping!")
                reconst_path = []

                while parents[n] != n:
                    reconst_path.append(n)
                    n = parents[n]

                reconst_path.append(start_node)

                reconst_path.reverse()

                print("Path found: {}".format(reconst_path))
                return reconst_path

            print(n)
            for (m, weight) in self.get_neighbors(n):
                print(f"And the iteration is ({m}, {weight})")
                # if the current node isn't in both open_list and closed_list
                # add it to open_list and note n as it's parent
                if m not in open_list and m not in closed_list:
                    open_list.add(m)
                    parents[m] = n
                    g[m] = g[n] + weight

                # otherwise, check if it's quicker to first visit n, then m
                # and if it is, update parent data and g data
                # and if the node was in the closed_list, move it to open_list
                else:
                    if g[m] > g[n] + weight:
                        g[m] = g[n] + weight
                        parents[m] = n

                        if m in closed_list:
                            closed_list.remove(m)
                            open_list.add(m)

            # remove n from the open_list, and add it to closed_list
            # because all of his neighbors were inspected
            # open_list.remove(node)
            # closed_list.add(node)

            open_list.remove(n)
            closed_list.add(n)
            print("\n")

        print("Path does not exist!")
        return None


adjacency_list = {
    "A": [("B", 1), ("C", 3), ("D", 7)],
    "B": [("D", 5)],
    "C": [("D", 12)],
}
A = Node(1)
B = Node(7)
C = Node(9)
D = Node(16)

A.edges = [(B, 1), (C, 3), (D, 7)]
B.edges = [(D, 5)]
C.edges = [(D, 12)]

# TODO use a dictionary comprehension to make this
adjacency_list = {
    A: A.edges,
    B: B.edges,
    C: C.edges,
}
graph1 = Graph(adjacency_list)
graph1.a_star_algorithm(A, D)

"""
Maybe after it
"""



File: ciphey/basemods/Searchers/imperfection.py

import heapq


class Imperfection:
    """The graph is a Node: [List of nodes]
    Where each item in the list of nodes can also have a node with a list of nodes

    The result is that we can keep track of edges, while also keeping it small

    To calculate current, we push the entire graph to A*

    And it calculates the next node to choose, as well as increasing the size
    of the graph with values

    We're using a heap, meaning the element at [0] is always the smallest element

    So we choose that and return it.


    The current A* implementation has an end, we simply do not let it end as LC will make it
    end far before it reaches Searcher again.

    Current is the start position, so if we say we always start at the start of the graph it'll
    go through the entire graph

    graph = {
            Node: [
                {Node :
                {
                    node
                    }
                }
                ]
            }

    For encodings we just do them straight out

    The last value of parents from abstract
    """

    """

   graph = {'A': ['B', 'C'],
             'B': ['C', 'D'],
             'C': ['D'],
             'D': ['C'],
             'E': ['F'],
             'F': ['C']}"""

    def __init__(self):
        None

    def findBestNode(self, nodes):
        """Finds the best decryption module"""
        return next(iter(nodes))

    # def aStar(self, graph, current, end):
    #     """The A* search algorithm

    #     We're using heaps to find the minimum element (the one that will be the next current)
    #     Heaps are like sets with O(1) lookup time, but maintain the lowest element as [0]
    #     Sets insert in O(1), heaps in O(log N).

    #     https://stackoverflow.com/questions/4159331/python-speed-up-an-a-star-pathfinding-algorithm

    #     Current appears to be the list of all new tiles we can reach from current location

    #     End is the end node, that won't actually run bc LC will make it return before it hits aSTar function
    #     so tbh I'll just make it infinite unless something else forces a return

    #     The graph is the actual data structure used. According to StackOverflow, it looks like this:

    #     graph = {'A': ['B', 'C'],
    #              'B': ['C', 'D'],
    #              'C': ['D'],
    #              'D': ['C'],
    #              'E': ['F'],
    #              'F': ['C']}

    #     """

    #     # Runs decodings first

    #     openSet = set()
    #     openHeap = []
    #     closedSet = set()

    #     def retracePath(c):
    #         # Retraces a path back to the start
    #         path = [c]
    #         while c.parent is not None:
    #             c = c.parent
    #             path.append(c)
    #         path.reverse()
    #         return path

    #     # Adds the current location (start) to the heap and set
    #     openSet.add(current)
    #     openHeap.append((0, current))

    #     # while openSet contains items
    #     while openSet:
    #         # TODO change openSet  to a heap?
    #         # gets the 2nd element from the first element of the heap
    #         # so the heap is (0, current)
    #         # which means we pop current
    #         # this makes me think that current isn't the first?
    #         current = heapq.heappop(openHeap)[1]
    #         # We don't actually want to end, so I'm commenting this:
    #         # XXX
    #         if current == end:
    #             return retracePath(current)
    #         # Removes it from todo and into done i think
    #         # closedSet appears to be the set of things we have done
    #         openSet.remove(current)
    #         closedSet.add(current)

    #         """
    #         Okay so our graph looks like this:
    #         graph = {
    #                 Node: [
    #                     {Node :
    #                     {
    #                         node
    #                         }
    #                     }
    #                     ]
    #                 }
    #         graph[current] **SHOULD** be the list of nodes which contains dictionaries of nodes

    #         """
    #         for tile in graph[current]:
    #             # ClosedSet appears to be the list of visited nodes
    #             # TODO place this as a class attribute
    #             if tile not in closedSet:
    #                 # This is the heuristic
    #                 # TODO expected_time/probability + k * heuristic, for some experimentally determined value of k
    #                 tile.H = (abs(end.x - tile.x) + abs(end.y - tile.y)) * 10

    #                 # if tile is not in the openSet, add it and then pop it from the heap
    #                 if tile not in openSet:
    #                     openSet.add(tile)
    #                     heapq.heappush(openHeap, (tile.H, tile))
    #                 #  I have no idea where this code is called lol
    #                 tile.parent = current

    #     # This returns Nothing
    #     # I need to modify it so it finds the best item from Current
    #     # So basically, return item 0 of openHeap
    #     # return openHeap[0]
    #     # Since the [0] item is always minimum
    #     return []
    def aStar(self, graph, current, end):
        print(f"The graph is {graph}\nCurrent is {current}\n and End is {end}")
        openSet = set()
        openHeap = []
        closedSet = set()

        def retracePath(c):
            print("Calling retrace path")
            path = [c]
            while c.parent is not None:
                c = c.parent
                path.append(c)
            path.reverse()
            return path

        print("\n")

        openSet.add(current)
        openHeap.append((0, current))
        while openSet:
            print(f"Openset is {openSet}")
            print(f"OpenHeap is {openHeap}")
            print(f"ClosedSet is {closedSet}")
            print(f"Current is {current}")
            print(f"I am popping {openHeap} with the first element")
            current = heapq.heappop(openHeap)[1]
            print(f"Current is now {current}")
            print(f"Graph current is {graph[current]}")
            if current == end:
                return retracePath(current)
            openSet.remove(current)
            closedSet.add(current)
            for tile in graph[current]:
                if tile not in closedSet:
                    tile.H = (abs(end.x - tile.x) + abs(end.y - tile.y)) * 10
                    tile.H = 1
                    if tile not in openSet:
                        openSet.add(tile)
                        heapq.heappush(openHeap, (tile.H, tile))
                    tile.parent = current
            print("\n")
        return []


class Node:
    """
    A node has a value associated with it
    Calculated from the heuristic
    """

    def __init__(self, h):
        self.h = h
        self.x = self.h
        self.y = 0.6

    def __le__(self, node2):
        # if self is less than other
        return self.x <= node2.x

    def __lt__(self, node2):
        return self.x < node2.x


if __name__ == "__main__":
    obj = Imperfection()
    graph = {
        "A": ["B", "C"],
        "B": ["C", "D"],
        "C": ["D"],
        "D": ["C"],
        "E": ["F"],
        "F": ["C"],
    }
    # Makes the graph
    y = Node(0.5)
    x = Node(0.3)
    p = Node(0.7)
    q = Node(0.9)
    graph = {y: [x, p], p: q}

    print(obj.aStar(graph, y, q))



File: ciphey/basemods/Searchers/perfection.py

from typing import Dict, Optional, Set

from ciphey.iface import Config, ParamSpec, registry

from .ausearch import AuSearch, Node


@registry.register
class Perfection(AuSearch):
    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    def findBestNode(self, nodes: Set[Node]) -> Node:
        return next(iter(nodes))

    def __init__(self, config: Config):
        super().__init__(config)



File: ciphey/basemods/Searchers/ausearch.py

import bisect
import distutils
import math
from copy import copy
from dataclasses import dataclass
from functools import lru_cache
from typing import Any, Dict, Generic, List, Optional, TypeVar, Union

import logging
from rich.logging import RichHandler

from ciphey.iface import (
    Checker,
    Config,
    Cracker,
    CrackInfo,
    CrackResult,
    Decoder,
    ParamSpec,
    Searcher,
    SearchLevel,
    SearchResult,
    T,
    registry,
)

"""
We are using a tree structure here, because that makes searching and tracing back easier
As such, when we encounter another possible parent, we remove that edge
"""


class DuplicateNode(Exception):
    pass


@dataclass
class AuSearchSuccessful(Exception):
    target: "Node"
    info: str


@dataclass
class Node:
    # The root has no parent edge
    level: SearchLevel
    parent: Optional["Edge"] = None
    depth: int = 0

    @staticmethod
    def decoding(
        config: Config, route: Union[Cracker, Decoder], result: Any, source: "Node"
    ) -> "Node":
        if not config.cache.mark_ctext(result):
            raise DuplicateNode()

        checker: Checker = config.objs["checker"]
        ret = Node(
            parent=None,
            level=SearchLevel(
                name=type(route).__name__.lower(), result=CrackResult(value=result)
            ),
            depth=source.depth + 1,
        )
        edge = Edge(source=source, route=route, dest=ret)
        ret.parent = edge
        check_res = checker(result)
        if check_res is not None:
            raise AuSearchSuccessful(target=ret, info=check_res)
        return ret

    @staticmethod
    def cracker(config: Config, edge_template: "Edge", result: CrackResult) -> "Node":
        if not config.cache.mark_ctext(result.value):
            raise DuplicateNode()

        checker: Checker = config.objs["checker"]
        # Edges do not directly contain containers, so this is fine
        edge = copy(edge_template)
        ret = Node(
            parent=edge,
            level=SearchLevel(name=type(edge.route).__name__.lower(), result=result),
            depth=edge.source.depth + 1,
        )
        edge.dest = ret
        check_res = checker(result.value)
        if check_res is not None:
            raise AuSearchSuccessful(target=ret, info=check_res)
        return ret

    @staticmethod
    def root(config: Config, ctext: Any):
        if not config.cache.mark_ctext(ctext):
            raise DuplicateNode()

        return Node(parent=None, level=SearchLevel.input(ctext))

    def get_path(self):
        if self.parent is None:
            return [self.level]
        return self.parent.source.get_path() + [self.level]


@dataclass
class AusearchEdge:
    # TODO: This is just CrackInfo with failure probability added...
    success_probability: float
    failure_probability: float
    success_time: float
    failure_time: float

    def __init__(self, success_probability, success_time, failure_time):
        self.success_probability = success_probability
        self.failure_probability = 1.0 - success_probability
        self.success_time = success_time
        self.failure_time = failure_time


@dataclass
class AusearchResult:
    weight: float
    index: int


def calculate_score(info: CrackInfo):
    return info.success_likelihood / \
        (info.success_runtime * info.success_likelihood + info.failure_runtime * (1-info.success_likelihood))


@dataclass
class Edge:
    source: Node
    route: Union[Cracker, Decoder]
    dest: Optional[Node] = None
    # Info is not filled in for Decoders
    score: Optional[float] = None


PriorityType = TypeVar("PriorityType")


class PriorityWorkQueue(Generic[PriorityType, T]):
    _sorted_priorities: List[PriorityType]
    _queues: Dict[Any, List[T]]

    def add_work(self, priority: PriorityType, work: List[T]) -> None:
        logging.debug(f"""Adding work at depth {priority}""")

        idx = bisect.bisect_left(self._sorted_priorities, priority)
        if (
            idx == len(self._sorted_priorities)
            or self._sorted_priorities[idx] != priority
        ):
            self._sorted_priorities.insert(idx, priority)
        self._queues.setdefault(priority, []).extend(work)

    def get_work(self) -> T:
        best_priority = self._sorted_priorities[0]
        target = self._queues[best_priority]
        ret = target.pop(0)
        if len(target) == 0:
            self._sorted_priorities.pop()
        return ret

    def get_work_chunk(self) -> List[T]:
        """Returns the best work for now"""
        if len(self._sorted_priorities) == 0:
            return []
        best_priority = self._sorted_priorities.pop(0)
        return self._queues.pop(best_priority)

    def empty(self):
        return len(self._sorted_priorities) == 0

    def __init__(self):
        self._sorted_priorities = []
        self._queues = {}


@registry.register
class AuSearch(Searcher):
    # Deeper paths get done later
    work: PriorityWorkQueue[int, Edge]

    @staticmethod
    def get_crackers_for(t: type):
        return registry[Cracker[t]]

    @lru_cache()  # To save extra sorting
    def get_decoders_for(self, t: type):
        ret = registry[Decoder[t]]
        ret.sort(key=lambda x: x.priority(), reverse=True)
        return ret

    # def expand(self, edge: Edge) -> List[Edge]:
    #     """Evaluates the destination of the given, and adds its child edges to the pool"""
    #     edge.dest = Node(parent=edge, level=edge.route(edge.source.level.result.value))

    def expand_crackers(self, node: Node) -> None:
        if node.depth >= self.max_cipher_depth:
            return

        res = node.level.result.value
        additional_work = []

        for i in self.get_crackers_for(type(res)):
            inst = self._config()(i)
            info = inst.getInfo(res)
            if info.success_likelihood < self.p_threshold:
                continue
            additional_work.append(
                Edge(source=node, route=inst, score=calculate_score(inst.getInfo(res)))
            )

        priority = min(node.depth, self.priority_cap)
        if self.invert_priority:
            priority = -priority

        self.work.add_work(priority, additional_work)

    def expand_decodings(self, node: Node) -> None:
        val = node.level.result.value

        for decoder in self.get_decoders_for(type(val)):
            inst = self._config()(decoder)
            res = inst(val)
            if res is None:
                continue
            try:
                new_node = Node.decoding(
                    config=self._config(), route=inst, result=res, source=node
                )
            except DuplicateNode:
                continue

            logging.debug("Nesting encodings")
            self.recursive_expand(new_node, False)

    def recursive_expand(self, node: Node, nested: bool = True) -> None:
        if node.depth >= self.max_depth:
            return

        logging.debug(f"Expanding depth {node.depth}")

        self.expand_decodings(node)

        # Doing this last allows us to catch simple nested encodings faster
        if not nested or self.enable_nested:
            self.expand_crackers(node)

    def search(self, ctext: Any) -> Optional[SearchResult]:
        logging.debug(
            f"""Beginning AuSearch with {"inverted" if self.invert_priority else "normal"} priority"""
        )

        try:
            root = Node.root(self._config(), ctext)
        except DuplicateNode:
            return None

        check_res = self._config().objs["checker"](ctext)
        if check_res is not None:
            return SearchResult(check_res=check_res, path=[root.level])

        try:
            self.recursive_expand(root, False)

            while True:
                if self.work.empty():
                    break
                # Get the highest level result
                chunk = self.work.get_work_chunk()
                chunk.sort(key=lambda i: i.score)
                # Work through all of this level's results
                while len(chunk) != 0:
                    logging.debug(f"{len(chunk)} remaining on this level")
                    # TODO Cyclic uses some tricky C++ here
                    # I know because it's sorted the one at the back (the anti-weight)
                    # is the most likely

                    edge: Edge = chunk.pop(-1)

                    # Expand the node
                    res = edge.route(edge.source.level.result.value)
                    if res is None:
                        continue
                    for i in res:
                        try:
                            node = Node.cracker(
                                config=self._config(), edge_template=edge, result=i
                            )
                            self.recursive_expand(node)
                        except DuplicateNode:
                            continue

        except AuSearchSuccessful as e:
            logging.info("AuSearch succeeded")
            return SearchResult(path=e.target.get_path(), check_res=e.info)

        logging.info("AuSearch failed")

    def __init__(self, config: Config):
        super().__init__(config)
        self._checker: Checker = config.objs["checker"]
        self.work = PriorityWorkQueue()  # Has to be defined here because of sharing
        self.invert_priority = bool(
            distutils.util.strtobool(self._params()["invert_priority"])
        )
        self.priority_cap = int(self._params()["priority_cap"])
        self.enable_nested = bool(
            distutils.util.strtobool(self._params()["enable_nested"])
        )
        self.max_cipher_depth = int(self._params()["max_cipher_depth"])
        if self.max_cipher_depth == 0:
            self.max_cipher_depth = math.inf
        self.max_depth = int(self._params()["max_depth"])
        if self.max_depth == 0:
            self.max_depth = math.inf
        self.p_threshold = float(self._params()["p_threshold"])

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "enable_nested": ParamSpec(
                req=False,
                desc="Enables nested ciphers. "
                "Incredibly slow, and not guaranteed to terminate",
                default="False",
            ),
            "invert_priority": ParamSpec(
                req=False,
                desc="Causes more complex encodings to be looked at first. "
                "Good for deeply buried encodings.",
                default="False",
            ),
            "max_cipher_depth": ParamSpec(
                req=False,
                desc="The depth at which we stop trying to crack ciphers. "
                "Set to 0 to disable",
                default="0",
            ),
            "max_depth": ParamSpec(
                req=False,
                desc="The depth at which we give up. " 
                     "Set to 0 to disable",
                default="0",
            ),
            "priority_cap": ParamSpec(
                req=False,
                desc="Sets the maximum depth before we give up ordering items.",
                default="2",
            ),
            "p_threshold": ParamSpec(
                req=False,
                desc="Will skip any crackers which have less than this likelihood of succeeding. "
                     "Set to 0 to disable",
                default="0.01",
            ),
        }



File: ciphey/basemods/Decoders/hexadecimal.py

from typing import Dict, Optional

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Hexadecimal(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Hexadecimal decoding
        """
        ctext_decoded = ""
        try:
            ctext_decoded = bytearray.fromhex(ctext).decode("utf-8")
            return ctext_decoded
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        return 0.015

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "hexadecimal"



File: ciphey/basemods/Decoders/gzip.py

import zlib
from typing import Dict, Optional

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Gzip(Decoder[bytes]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Gzip decoding
        """
        try:
            return zlib.decompress(ctext, 16 + zlib.MAX_WBITS).decode("utf-8")
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        # Not expected to show up often, but also very fast to check.
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "gzip"



File: ciphey/basemods/Decoders/binary.py

import re
from typing import Dict, List, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Binary(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        try:
            ctext = re.sub(r"[^\S \n]", " ", ctext, flags=re.UNICODE)
            ctext = ctext.replace("\n", " ")

            existing_split = self.try_split(ctext.split(" "))
            if existing_split is not None:
                return existing_split

            # Now we try our own grouping

            # Remove final bit of whitespace
            ctext = ctext.replace(" ", "")
            # Split into bytes, and test
            return self.try_split([ctext[i : i + 8] for i in range(0, len(ctext), 8)])
        # Catch bad octal chars
        except ValueError:
            return None

    def try_split(self, split_text: List[str]):
        ret = []

        for i in split_text:
            if len(i) == 0:
                continue
            val = int(i, 2)
            if val > 255 or val < 0:
                return None
            ret.append(val)

        if len(ret) != 0:
            ret = bytes(ret)
            logging.info(f"binary successful, returning {ret.__repr__()}")
            return ret

    @staticmethod
    def priority() -> float:
        return 0.3

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "binary"



File: ciphey/basemods/Decoders/a1z26.py

import re
from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class A1z26(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs A1Z26 decoding
        """
        logging.debug("Attempting A1Z26")
        ctext_converted = []
        ctext_split = re.split(r"[ ,;:\-\n]", ctext)
        delimiters = set(sorted(re.sub(r"[^ ,;:\-\n]", "", ctext)))
        ctext_num = re.sub(r"[,;:\-\s]", "", ctext)
        ctext_decoded = ""
        if ctext_num.isnumeric() is False:
            logging.debug("Failed to decode A1Z26 due to non numeric character(s)")
            return None
        try:
            for i in ctext_split:
                val = int(i)
                if val > 26 or val < 1:
                    logging.debug(
                        f"Failed to decode A1Z26 due to invalid number '{val}'"
                    )
                    return None
                val2 = int(i) + 96
                ctext_converted.append(chr(val2))
            ctext_decoded = "".join(ctext_converted)
            logging.info(
                f"A1Z26 successful, returning '{ctext_decoded}' with delimiter(s) {delimiters}"
            )
            return ctext_decoded
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "a1z26"



File: ciphey/basemods/Decoders/galactic.py

from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Decoder, ParamSpec, T, Translation, U, registry


@registry.register
class Galactic(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Takes a string written in the 'Standard Galactic Alphabet'
        (aka Minecraft Enchanting Table Symbols) and translates it to ASCII text.
        """
        logging.debug("Attempting Standard Galactic Alphabet decoder")

        # To avoid complications, only move forward with the decoding if we can
        # reasonably assume that the input string is written in the galactic alphabet
        galactic_matches = 0
        for symbol in self.GALACTIC_DICT.keys():
            # These symbols are assumed to be frequent enough in regular
            # text to be skipped when counting the matches. All others are counted.
            if symbol in ctext and symbol not in ["!", "|"]:
                galactic_matches += 1
            else:
                continue
        if galactic_matches == 0:
            logging.debug(
                "No matching galactic alphabet letters found. Skipping galactic decoder"
            )
            return None
        logging.debug(f"{galactic_matches} galactic alphabet letters found. ")

        result = ""
        # Take out the problematic characters consisting of multiple symbols
        ctext = (
            ctext.replace("||", "|")
            .replace("/", "")
            .replace("¡", "")
            .replace(" ̣ ", "")
            .replace("̇", "x")
        )
        logging.debug(f"Modified string is {ctext}")

        for letter in ctext:
            if letter in self.GALACTIC_DICT.keys():
                # Match every letter of the input to its galactic counterpoint
                result += self.GALACTIC_DICT[letter]
            else:
                # If the current character is not in the defined alphabet,
                # just accept it as-is (useful for numbers, punctuation, etc.)
                result += letter
        # Remove the trailing space (appearing as a leading space)
        # from the x that results from the diacritic replacement
        result = result.replace("x ", "x")
        logging.debug(f"Decoded string is {result}")
        return result

    @staticmethod
    def priority() -> float:
        # Not expected to show up often, but also very fast to check.
        return 0.01

    def __init__(self, config: Config):
        super().__init__(config)
        self.GALACTIC_DICT = config.get_resource(self._params()["dict"], Translation)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "dict": ParamSpec(
                desc="The galactic alphabet dictionary to use",
                req=False,
                default="cipheydists::translate::galactic",
            )
        }

    @staticmethod
    def getTarget() -> str:
        return "galactic"



File: ciphey/basemods/Decoders/base58_ripple.py

from typing import Dict, Optional

import base58

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Base58_ripple(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Base58 (Ripple) decoding
        """
        try:
            return base58.b58decode(ctext, alphabet=base58.RIPPLE_ALPHABET).decode(
                "utf-8"
            )
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        # Not expected to show up often, but also very fast to check.
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "base58_ripple"



File: ciphey/basemods/Decoders/uuencode.py

from binascii import a2b_uu
from codecs import decode
from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Uuencode(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        UUEncode (Unix to Unix Encoding) is a symmetric encryption
        based on conversion of binary data (split into 6-bit blocks) into ASCII characters.

        This function decodes the input string 'ctext' if it has been encoded using 'uuencoder'
        It will return None otherwise
        """
        logging.debug("Attempting UUencode")
        result = ""
        try:
            # UUencoded messages may begin with prefix "begin" and end with suffix "end"
            # In that case, we use the codecs module in Python
            ctext_strip = ctext.strip()
            if ctext_strip.startswith("begin") and ctext_strip.endswith("end"):
                result = decode(bytes(ctext, "utf-8"), "uu").decode()
            else:
                # If there isn't a "being" prefix and "end" suffix, we use the binascii module instead
                # It is possible that the ctext has multiple lines, so convert each line and append
                ctext_split = list(filter(None, ctext.splitlines()))
                for _, value in enumerate(ctext_split):
                    result += a2b_uu(value).decode("utf-8")
            logging.info(f"UUencode successful, returning '{result}'")
            return result
        except Exception:
            logging.debug("Failed to decode UUencode")
            return None

    @staticmethod
    def priority() -> float:
        # Not expected to show up often, but also very fast to check.
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "uuencode"



File: ciphey/basemods/Decoders/unicode.py

from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Utf8(Decoder[bytes]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs UTF-8 decoding
        """
        logging.debug("Attempting UTF-8 decoder")
        result = ""
        try:
            result = ctext.decode("utf-8")
            if result != ctext:
                logging.info(f"UTF-8 successful, returning '{result}'")
                return result
            else:
                return None
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        return 0.9

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "utf8"



File: ciphey/basemods/Decoders/braille.py

import re
from typing import Dict, Optional

from ciphey.iface import Config, Decoder, ParamSpec, T, Translation, U, registry
import logging
from rich.logging import RichHandler


@registry.register
class Braille(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Braille decoding
        """
        logging.debug("Attempting Braille")
        ctext_decoded = ""
        braille_matches = 0
        for symbol in self.BRAILLE_DICT_INV.values():
            if symbol in ctext:
                braille_matches += 1
            else:
                continue
        if braille_matches == 0:
            logging.debug("Failed to decode Braille due to invalid characters")
            return None

        for pattern, value in self.BRAILLE_DICT.items():
            ctext = re.sub(pattern, value, ctext)

        wordArr = []
        for word in ctext.split(" "):
            # If two commas are in front of a word, uppercase the word and remove the comma
            if word[:2].find(",,") != -1:
                wordArr.append(word.replace(",,", "").upper())
            else:
                wordArr.append(word)

        result = []
        for word in wordArr:
            # If one comma is in front of a word, capitalize the word and remove the comma
            if word[0].find(",") != -1:
                result.append(word.replace(",", "").capitalize())
            else:
                result.append(word)
        ctext_decoded = " ".join(result)
        logging.info(f"Braille successful, returning '{ctext_decoded}'")
        return ctext_decoded

    @staticmethod
    def priority() -> float:
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)
        self.BRAILLE_DICT = config.get_resource(self._params()["dict"], Translation)
        self.BRAILLE_DICT_INV = {v: k for k, v in self.BRAILLE_DICT.items()}

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "dict": ParamSpec(
                desc="The Braille dictionary to use",
                req=False,
                default="cipheydists::translate::braille",
            )
        }

    @staticmethod
    def getTarget() -> str:
        return "braille"



File: ciphey/basemods/Decoders/reverse.py

from typing import Dict, Optional

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Reverse(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        return ctext[::-1]

    @staticmethod
    def priority() -> float:
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "reverse"



File: ciphey/basemods/Decoders/morse_code.py

from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Decoder, ParamSpec, T, Translation, U, registry


@registry.register
class Morse_code(Decoder[str]):
    # A priority list for char/word boundaries
    BOUNDARIES = {" ": 1, "/": 2, "\n": 3}
    PURGE = {ord(c): None for c in BOUNDARIES.keys()}
    MAX_PRIORITY = 3
    ALLOWED = {".", "-", " ", "/", "\n"}
    MORSE_CODE_DICT: Dict[str, str]
    MORSE_CODE_DICT_INV: Dict[str, str]

    def decode(self, ctext: T) -> Optional[U]:
        logging.debug("Attempting Morse code decoder")

        char_boundary = word_boundary = None

        char_boundary = word_boundary = None
        char_priority = word_priority = 0
        # Custom loop allows early break
        for i in ctext:
            i_priority = self.BOUNDARIES.get(i)
            if i_priority is None:
                if i in self.ALLOWED:
                    continue
                logging.debug(f"Non-morse char '{i}' found")
                return None

            if i_priority <= char_priority or i == char_boundary or i == word_boundary:
                continue
            # Default to having a char boundary over a word boundary
            if (
                i_priority > word_priority
                and word_boundary is None
                and char_boundary is not None
            ):
                word_priority = i_priority
                word_boundary = i
                continue
            char_priority = i_priority
            char_boundary = i

        logging.debug(
            f"Char boundary is unicode {ord(char_boundary)}, and word boundary is unicode {ord(word_boundary) if word_boundary is not None else None}"
        )

        result = ""

        for word in ctext.split(word_boundary) if word_boundary else [ctext]:
            logging.debug(f"Attempting to decode word {word}")
            for char in word.split(char_boundary):
                char = char.translate(self.PURGE)
                if len(char) == 0:
                    continue
                try:
                    m = self.MORSE_CODE_DICT_INV[char]
                except KeyError:
                    logging.debug(f"Invalid codeword '{char}' found")
                    return None
                result = result + m
            # after every word add a space
            result = result + " "
        if len(result) == 0:
            logging.debug("Morse code failed to match")
            return None
        # Remove trailing space
        result = result[:-1]
        logging.info(f"Morse code successful, returning {result}")
        return result.strip().upper()

    @staticmethod
    def priority() -> float:
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)
        self.MORSE_CODE_DICT = config.get_resource(self._params()["dict"], Translation)
        self.MORSE_CODE_DICT_INV = {v: k for k, v in self.MORSE_CODE_DICT.items()}

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "dict": ParamSpec(
                desc="The morse code dictionary to use",
                req=False,
                default="cipheydists::translate::morse",
            )
        }

    @staticmethod
    def getTarget() -> str:
        return "morse_code"



File: ciphey/basemods/Decoders/__init__.py

from . import (
    a1z26,
    atbash,
    base58_bitcoin,
    base58_ripple,
    base62,
    base69,
    base91,
    bases,
    baudot,
    binary,
    braille,
    brainfuck,
    decimal,
    dna,
    dtmf,
    galactic,
    gzip,
    hexadecimal,
    leetspeak,
    morse_code,
    multi_tap,
    octal,
    reverse,
    tap_code,
    unicode,
    url,
    uuencode,
)



File: ciphey/basemods/Decoders/decimal.py

import re
from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Decimal(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Decimal decoding
        """
        logging.debug("Attempting decimal")
        ctext_converted = []
        ctext_split = re.split(r"[ ,;:\-\n]", ctext)
        delimiters = set(sorted(re.sub(r"[^ ,;:\-\n]", "", ctext)))
        ctext_num = re.sub(r"[,;:\-\s]", "", ctext)
        ctext_decoded = ""
        if ctext_num.isnumeric() is False:
            logging.debug("Failed to decode decimal due to non numeric character(s)")
            return None
        try:
            for i in ctext_split:
                val = int(i)
                if val > 255 or val < 0:
                    logging.debug(
                        f"Failed to decode decimal due to invalid number '{val}'"
                    )
                    return None
                ctext_converted.append(chr(val))
            ctext_decoded = "".join(ctext_converted)
            logging.info(
                f"Decimal successful, returning '{ctext_decoded}' with delimiter(s) {delimiters}"
            )
            return ctext_decoded
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "decimal"



File: ciphey/basemods/Decoders/base65536.py

from typing import Dict, Optional

import base65536

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Base65536(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Base65536 decoding
        """
        try:
            return base65536.decode(ctext).decode("utf-8")
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        # Not expected to show up often, but also very fast to check.
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "base65536"



File: ciphey/basemods/Decoders/dna.py

import re
from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Decoder, ParamSpec, T, Translation, U, registry


@registry.register
class Dna(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs DNA decoding
        """
        logging.debug("Attempting DNA decoder")
        ctext_decoded = ""
        ctext = re.sub(r"[,;:\-\s]", "", ctext)
        ctext = " ".join(ctext[i : i + 3] for i in range(0, len(ctext), 3))
        ctext_split = ctext.split(" ")
        dna_keys = self.DNA_DICT.keys()

        for i in ctext_split:
            if i in dna_keys:
                ctext_decoded += self.DNA_DICT[i]
            else:
                return None
        logging.info(f"DNA successful, returning '{ctext_decoded}'")
        return ctext_decoded

    @staticmethod
    def priority() -> float:
        return 0.2

    def __init__(self, config: Config):
        super().__init__(config)
        self.DNA_DICT = config.get_resource(self._params()["dict"], Translation)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "dict": ParamSpec(
                desc="The DNA alphabet dictionary to use",
                req=False,
                default="cipheydists::translate::dna",
            )
        }

    @staticmethod
    def getTarget() -> str:
        return "dna"



File: ciphey/basemods/Decoders/leetspeak.py

from typing import Dict, Optional

from ciphey.iface import Config, Decoder, ParamSpec, T, Translation, U, registry


@registry.register
class Leetspeak(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        for src, dst in self.translate.items():
            ctext = ctext.replace(src, dst)
        return ctext

    @staticmethod
    def priority() -> float:
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)
        self.translate = config.get_resource(self._params()["dict"], Translation)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "dict": ParamSpec(
                desc="The leetspeak dictionary to use",
                req=False,
                default="cipheydists::translate::leet",
            )
        }

    @staticmethod
    def getTarget() -> str:
        return "leetspeak"



File: ciphey/basemods/Decoders/multi_tap.py

from typing import Dict, Optional

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Multi_tap(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        result = ""
        for x in ctext.split():
            if x == self.SPACE_DIGIT:  # Check if it's a space
                result += " "
            elif not Multi_tap.valid_code_part(x):
                return None
            else:
                result += self.decode_num_to_char(x)

        return result

    @staticmethod
    def valid_code_part(code: str) -> bool:
        if not code.isdigit():
            return False

        # if not all the digits are the same
        if not Multi_tap.is_all_dup(code):
            return False

        if int(code[0]) not in range(2, 10):
            return False

        if len(code) > 4:
            return False

        return True

    @staticmethod
    def decode_num_to_char(number: str) -> str:
        index = Multi_tap.calculate_index(number)
        return Multi_tap.number_index_to_char(index)

    @staticmethod
    def is_all_dup(code):
        return len(set(code)) == 1

    @staticmethod
    def calculate_index(number: str) -> int:
        first_number_as_int = int(number[0])

        number_index = Multi_tap.get_index_from_first_digit(first_number_as_int)

        # Add to index the number of the char : "22" -> index += 1
        num_rest_numbers = len(number) - 1
        number_index += num_rest_numbers

        return number_index

    @staticmethod
    def number_index_to_char(index_number: int) -> str:
        start_ascii_value = ord("A")
        return chr(start_ascii_value + index_number)

    @staticmethod
    def get_index_from_first_digit(first_digit: int) -> int:
        number_index = 0
        if first_digit >= 8:  # s have 4 chars
            number_index += 1

        first_digit -= 2  # start in 200

        number_index += first_digit * 3  # jump 3 every time

        return number_index

    @staticmethod
    def priority() -> float:
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)
        self.SPACE_DIGIT = "0"

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "multi_tap"



File: ciphey/basemods/Decoders/url.py

from typing import Dict, Optional
from urllib.parse import unquote_plus

import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Url(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs URL decoding
        """
        logging.debug("Attempting URL")
        result = ""
        try:
            result = unquote_plus(ctext, errors="strict")
            if result != ctext:
                logging.info(f"URL successful, returning '{result}'")
                return result
            else:
                return None
        except Exception:
            logging.debug("Failed to decode URL")
            return None

    @staticmethod
    def priority() -> float:
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "url"



File: ciphey/basemods/Decoders/base69.py

# Translated to Python and adapted for Ciphey from the JS original at https://github.com/pshihn/base69


import re
from math import ceil
from typing import Dict, Optional

from ciphey.iface import Config, Decoder, ParamSpec, T, U, WordList, registry


@registry.register
class Base69(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Base69 decoding
        """
        # Remove whitespace
        try:
            ctext = re.sub(r"\s+", "", ctext, flags=re.UNICODE)
            extra_bytes = 0
            clen = len(ctext)

            if ctext[:-1] == "=":
                extra_bytes = int(ctext[clen - 2])

            CHUNK_COUNT = ceil(clen / 16)
            result = [0 for _ in range(CHUNK_COUNT * 7 - extra_bytes)]

            for i in range(CHUNK_COUNT):
                chunk_string = ctext[i * 16 : (i + 1) * 16]
                if extra_bytes and (i == CHUNK_COUNT - 1):
                    insert = self.decode_chunk(chunk_string)
                    for n, elem in enumerate(insert[0 : 7 - extra_bytes]):
                        result[n + i * 7] = elem
                else:
                    insert = self.decode_chunk(chunk_string)
                    for n, elem in enumerate(insert):
                        result[n + i * 7] = elem % 256
            return bytearray(result).decode().strip("\x00")
        except Exception:
            return None

    def decode_chunk(self, s: str):
        padded_bytes = s.endswith("=")

        decoded = [0 for _ in range(8)]
        for i in range(8):
            decoded[i] = (
                0
                if i == 7 and padded_bytes
                else self.chars_to_byte(s[i * 2 : i * 2 + 2])
            )

        result = [0 for _ in range(7)]
        for i in range(7):
            t1 = decoded[i] << (i + 1)
            t2 = decoded[i + 1] >> (7 - i - 1)
            result[i] = t1 | t2
        return result

    def chars_to_byte(self, s: str):
        return (69 * self.CHARS.index(s[1])) + (self.CHARS.index(s[0]))

    @staticmethod
    def priority() -> float:
        # If this becomes lower or equal to the reverse, it breaks.
        # So I'll set it to 0.2 for now since it is very fast anyways.
        return 0.2

    def __init__(self, config: Config):
        super().__init__(config)
        self.CHARS = config.get_resource(self._params()["dict"], WordList)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "dict": ParamSpec(
                desc="The charset used for the decoder.",
                req=False,
                default="cipheydists::list::base69",
            )
        }

    @staticmethod
    def getTarget() -> str:
        return "base69"



File: ciphey/basemods/Decoders/base58_bitcoin.py

from typing import Dict, Optional

import base58

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Base58_bitcoin(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Base58 (Bitcoin) decoding
        """
        try:
            return base58.b58decode(ctext).decode("utf-8")
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        # Not expected to show up often, but also very fast to check.
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "base58_bitcoin"



File: ciphey/basemods/Decoders/base64_url.py

import base64
from typing import Dict, Optional

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Base64_url(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Base64 URL decoding
        """
        ctext_padding = ctext + "=" * (4 - len(ctext) % 4)
        try:
            return base64.urlsafe_b64decode(ctext_padding).decode("utf-8")
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        # Not expected to show up often, but also very fast to check.
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "base64_url"



File: ciphey/basemods/Decoders/baudot.py

import re
from typing import Dict, Optional

from ciphey.iface import Config, Decoder, ParamSpec, T, Translation, U, registry


@registry.register
class Baudot(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        result = ""
        switch_to_digit_map = 0
        if re.search("^[01]{5}$", ctext.split()[0]):
            for i in ctext.split():
                if i == "11011":
                    switch_to_digit_map = 1
                if i == "11111":
                    switch_to_digit_map = 0
                if switch_to_digit_map == 1:
                    result += self.BAUDOT_DICT["+" + i]
                if switch_to_digit_map == 0:
                    result += self.BAUDOT_DICT[i]
            return result
        else:
            return None

    @staticmethod
    def priority() -> float:
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)
        self.BAUDOT_DICT = config.get_resource(self._params()["dict"], Translation)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "dict": ParamSpec(
                desc="The baudot alphabet dictionary to use",
                req=False,
                default="cipheydists::translate::baudot",
            )
        }

    @staticmethod
    def getTarget() -> str:
        return "baudot"



File: ciphey/basemods/Decoders/tap_code.py

# by https://github.com/RustyDucky and https://github.com/lukasgabriel

from typing import Dict, Optional

from ciphey.iface import Config, Decoder, ParamSpec, T, Translation, U, registry


@registry.register
class Tap_code(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Tap code decoding
        """
        try:
            result = ""
            combinations = ctext.split(" ")
            for fragment in combinations:
                result += self.TABLE.get(fragment)
            return result
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        return 0.06

    def __init__(self, config: Config):
        super().__init__(config)
        self.TABLE = config.get_resource(self._params()["dict"], Translation)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "dict": ParamSpec(
                desc="The table of letters used for the tap code interpretation.",
                req=False,
                default="cipheydists::translate::tap_code",
            )
        }

    @staticmethod
    def getTarget() -> str:
        return "tap_code"



File: ciphey/basemods/Decoders/octal.py

from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Octal(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Octal decoding
        """
        str_converted = []
        octal_seq = ctext.split(" ")
        if len(octal_seq) == 1:
            # Concatted octal must be formed of octal triplets
            if len(ctext) % 3 != 0:
                return None
            octal_seq = [ctext[i : i + 3] for i in range(0, len(ctext), 3)]
            logging.debug(f"Trying chunked octal {octal_seq}")
        try:
            for octal_char in octal_seq:
                if len(octal_char) > 3:
                    logging.debug("Octal subseq too long")
                    return None
                n = int(octal_char, 8)
                if (
                    n < 0
                ):  # n cannot be greater than 255, as we checked that with the earlier length check
                    logging.debug(f"Non octal char {octal_char}")
                    return None
                str_converted.append(n)

            return bytes(str_converted)
        # Catch bad octal chars
        except ValueError:
            return None

    @staticmethod
    def priority() -> float:
        return 0.025

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "octal"



File: ciphey/basemods/Decoders/dtmf.py

import re
from typing import Dict, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Decoder, ParamSpec, T, Translation, U, registry


@registry.register
class Dtmf(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs DTMF decoding
        """
        logging.debug("Attempting DTMF decoder")
        ctext_decoded = ""
        ctext = re.sub(r"[,;:\-\/\s]", "", ctext)
        ctext = " ".join(ctext[i : i + 7] for i in range(0, len(ctext), 7))
        ctext_split = ctext.split(" ")
        dtmf_keys = self.DTMF_DICT.keys()

        for i in ctext_split:
            if i in dtmf_keys:
                ctext_decoded += self.DTMF_DICT[i]
            else:
                return None
        logging.info(f"DTMF successful, returning '{ctext_decoded}'")
        return ctext_decoded

    @staticmethod
    def priority() -> float:
        return 0.2

    def __init__(self, config: Config):
        super().__init__(config)
        self.DTMF_DICT = config.get_resource(self._params()["dict"], Translation)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "dict": ParamSpec(
                desc="The DTMF alphabet dictionary to use",
                req=False,
                default="cipheydists::translate::dtmf",
            )
        }

    @staticmethod
    def getTarget() -> str:
        return "dtmf"



File: ciphey/basemods/Decoders/base91.py

from typing import Dict, Optional

import base91

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Base91(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Base91 decoding
        """
        try:
            return base91.decode(ctext).decode("utf-8")
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        # Not expected to show up often, but also very fast to check.
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "base91"



File: ciphey/basemods/Decoders/atbash.py

from typing import Dict, Optional

from ciphey.common import fix_case
from ciphey.iface import Config, Decoder, ParamSpec, T, U, WordList, registry


@registry.register
class Atbash(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Takes an encoded string and attempts to decode it according to the Atbash cipher.

        The Atbash cipher is a very simple substitution cipher without a key.
        It operates by replacing every letter in the input by its 'counterpoint'
        in the alphabet. Example: A -> Z, B -> Y, ... , M -> N and vice versa.
        """

        result = ""
        atbash_dict = {self.ALPHABET[i]: self.ALPHABET[::-1][i] for i in range(26)}

        for letter in ctext.lower():
            if letter in atbash_dict.keys():
                # Match every letter of the input to its atbash counterpoint
                result += atbash_dict[letter]
            else:
                # If the current character is not in the defined alphabet,
                # just accept it as-is (useful for numbers, punctuation, etc.)
                result += letter
        return fix_case(result, ctext)

    @staticmethod
    def priority() -> float:
        # Not expected to show up often, but also very fast to check.
        return 0.1

    def __init__(self, config: Config):
        super().__init__(config)
        self.ALPHABET = config.get_resource(self._params()["dict"], WordList)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "dict": ParamSpec(
                desc="The alphabet used for the atbash operation.",
                req=False,
                default="cipheydists::list::englishAlphabet",
            )
        }

    @staticmethod
    def getTarget() -> str:
        return "atbash"



File: ciphey/basemods/Decoders/z85.py

from typing import Dict, Optional

import logging
from rich.logging import RichHandler
from zmq.utils import z85

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Z85(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Z85 decoding
        """
        ctext_len = len(ctext)
        if ctext_len % 5:
            logging.debug(
                f"Failed to decode Z85 because length must be a multiple of 5, not '{ctext_len}'"
            )
            return None
        try:
            return z85.decode(ctext).decode("utf-8")
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        # Not expected to show up often, but also very fast to check.
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "z85"



File: ciphey/basemods/Decoders/base62.py

from typing import Dict, Optional

import base62

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Base62(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Base62 decoding
        """
        try:
            return base62.decodebytes(ctext).decode("utf-8")
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        # Not expected to show up often, but also very fast to check.
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "base62"



File: ciphey/basemods/Decoders/brainfuck.py

import re
import time
from typing import Dict, Optional, Tuple

import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Decoder, ParamSpec, T, U, WordList, registry


@registry.register
class Brainfuck(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Takes a ciphertext and treats it as a Brainfuck program,
        interpreting it and saving the output as a string to return.

        Brainfuck is a very simple, Turing-complete esoteric language.
        Below is a simplified interpreter that attempts to check whether a
        given ciphertext is a brainfuck program that would output a string.

        A program that can be "decoded" like this is one that:
            * Does not require user input ("," instruction)
            * Includes at least one putchar instruction (".")
            * Does not contain anything but the main 7 instructions,
                (excluding ",") and whitespace

        Details:
            * This implementation wraps the memory pointer for ">" and "<"
            * It is time-limited to 60 seconds, to prevent hangups
            * The program starts with 100 memory cells, chosen arbitrarily
        """

        logging.debug("Attempting brainfuck")

        result = ""
        memory = [0] * 100
        codeptr, memptr = 0, 0  # Instruction pointer and stack pointer
        timelimit = 60  # The timeout in seconds

        bracemap, isbf = self.bracemap_and_check(ctext)

        # If it doesn't appear to be valid brainfuck code
        if not isbf:
            logging.debug("Failed to interpret brainfuck due to invalid characters")
            return None

        # Get start time
        start = time.time()

        while codeptr < len(ctext):

            current = time.time()

            # Return none if we've been running for over a minute
            if current - start > timelimit:
                logging.debug("Failed to interpret brainfuck due to timing out")
                return None

            cmd = ctext[codeptr]

            if cmd == "+":
                if memory[memptr] < 255:
                    memory[memptr] = memory[memptr] + 1
                else:
                    memory[memptr] = 0

            elif cmd == "-":
                if memory[memptr] > 0:
                    memory[memptr] = memory[memptr] - 1
                else:
                    memory[memptr] = 255

            elif cmd == ">":
                if memptr == len(memory) - 1:
                    memory.append(0)
                memptr += 1

            elif cmd == "<":
                if memptr == 0:
                    memptr = len(memory) - 1
                else:
                    memptr -= 1

            # If we're at the beginning of the loop and the memory is 0, exit the loop
            elif cmd == "[" and memory[memptr] == 0:
                codeptr = bracemap[codeptr]

            # If we're at the end of the loop and the memory is >0, jmp to the beginning of the loop
            elif cmd == "]" and memory[memptr]:
                codeptr = bracemap[codeptr]

            # Store the output as a string instead of printing it out
            elif cmd == ".":
                result += chr(memory[memptr])

            codeptr += 1

        logging.info(f"Brainfuck successful, returning '{result}'")
        return result

    def bracemap_and_check(self, program: str) -> Tuple[Optional[Dict], bool]:
        """
        Create a bracemap of brackets in the program, to compute jmps.
        Maps open -> close brackets as well as close -> open brackets.

        Also returns True if the program is valid Brainfuck code. If False, we
        won't even try to run it.
        """

        open_stack = []
        bracemap = dict()
        legal_instructions = {"+", "-", ">", "<", "[", "]", "."}
        legal_count = 0

        # If the program actually outputs anything (contains ".")
        prints = False

        for idx, instruction in enumerate(program):
            # If instruction is brainfuck (without input) or whitespace, it counts
            if instruction in legal_instructions or re.match(r"\s", instruction):
                legal_count += 1

            if not prints and instruction == ".":
                # If there are no "." instructions then this program will not output anything
                prints = True

            elif instruction == "[":
                open_stack.append(idx)

            elif instruction == "]":
                try:
                    opbracket = open_stack.pop()
                    bracemap[opbracket] = idx
                    bracemap[idx] = opbracket
                except IndexError:
                    # Mismatched braces, not a valid program
                    # Closing braces > opening braces
                    return (None, False)

        # 1. All characters are instructions or whitespace
        # 2. There are no extra open braces
        # 3. There is at least one character to be "printed"
        # (result is >=1 in length)
        is_brainfuck = legal_count == len(program) and len(open_stack) == 0 and prints

        return bracemap, is_brainfuck

    @staticmethod
    def priority() -> float:
        # Not uncommon, but not very common either. It's also slow.
        return 0.08

    def __init__(self, config: Config):
        super().__init__(config)
        self.ALPHABET = config.get_resource(self._params()["dict"], WordList)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "dict": ParamSpec(
                desc="Brainfuck alphabet (default English)",
                req=False,
                default="cipheydists::list::englishAlphabet",
            )
        }

    @staticmethod
    def getTarget() -> str:
        return "brainfuck"



File: ciphey/basemods/Decoders/base58_flickr.py

from typing import Dict, Optional

import base58

from ciphey.iface import Config, Decoder, ParamSpec, T, U, registry


@registry.register
class Base58_flickr(Decoder[str]):
    def decode(self, ctext: T) -> Optional[U]:
        """
        Performs Base58 (Flickr) decoding
        """
        FLICKR_ALPHABET = b"123456789abcdefghijkmnopqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ"
        try:
            return base58.b58decode(ctext, alphabet=FLICKR_ALPHABET).decode("utf-8")
        except Exception:
            return None

    @staticmethod
    def priority() -> float:
        # Not expected to show up often, but also very fast to check.
        return 0.05

    def __init__(self, config: Config):
        super().__init__(config)

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def getTarget() -> str:
        return "base58_flickr"



File: ciphey/basemods/Decoders/bases.py

import base64
import types
from typing import Any, Callable, Optional

import logging
from rich.logging import RichHandler
import re

from ciphey.common import id_lambda
from ciphey.iface import Decoder, registry


def _dispatch(self: Any, ctext: str, func: Callable[[str], bytes]) -> Optional[bytes]:
    logging.debug(f"Attempting {self.getTarget()}")

    try:
        # remove all whitespace
        ctext = re.sub(r"\s+", "", ctext, re.UNICODE)
        result = func(ctext)
        logging.info(f"{self.getTarget()} successful, returning {result}")
        return result
    except ValueError:
        logging.debug(f"Failed to decode {self.getTarget()}")
        return None


_bases = {
    "base16": (base64.b16decode, 0.4),
    "base32": (base64.b32decode, 0.01),
    "base64": (base64.b64decode, 0.4),
    "base85": (base64.b85decode, 0.01),
    "ascii85": (base64.a85decode, 0.1),
}


def gen_class(name, decoder, priority, ns):
    ns["_get_func"] = id_lambda(decoder)
    ns["decode"] = lambda self, ctext: _dispatch(self, ctext, self._get_func())
    ns["getParams"] = id_lambda(None)
    ns["getTarget"] = id_lambda(name)
    ns["priority"] = id_lambda(priority)
    ns["__init__"] = lambda self, config: super(type(self), self).__init__(config)


for name, (decoder, priority) in _bases.items():
    t = types.new_class(
        name,
        (Decoder[str],),
        exec_body=lambda x: gen_class(name, decoder, priority, x),
    )

    registry.register(t)



File: ciphey/basemods/Crackers/xandy.py

import re
from typing import Dict, List, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Cracker, CrackInfo, CrackResult, ParamSpec, registry


@registry.register
class Xandy(Cracker[str]):
    def getInfo(self, ctext: str) -> CrackInfo:
        return CrackInfo(
            success_likelihood=0.1,
            success_runtime=1e-5,
            failure_runtime=1e-5,
        )

    @staticmethod
    def binary_to_ascii(variant):
        # Convert the binary string to an integer with base 2
        binary_int = int(variant, 2)
        byte_number = binary_int.bit_length() + 7 // 8

        # Convert the resulting int to a bytearray and then decode it to ASCII text
        binary_array = binary_int.to_bytes(byte_number, "big")
        try:
            ascii_text = binary_array.decode()
            logging.debug(f"Found possible solution: {ascii_text[:32]}")
            return ascii_text
        except UnicodeDecodeError as e:
            logging.debug(f"Failed to crack X-Y due to a UnicodeDecodeError: {e}")
            return ""

    @staticmethod
    def getTarget() -> str:
        return "xandy"

    def attemptCrack(self, ctext: str) -> List[CrackResult]:
        """
        Checks an input if it only consists of two or three different letters.
        If this is the case, it attempts to regard those letters as
        0 and 1 (with the third characters as an optional delimiter) and then
        converts it to ASCII text.
        """
        logging.debug("Attempting X-Y replacement")
        variants = []
        candidates = []
        result = []

        # Convert the ctext to all-lowercase and regex-match & replace all whitespace
        ctext = re.sub(r"\s+", "", ctext.lower(), flags=re.UNICODE)

        # cset contains every unique value in the ctext
        cset = list(set(list(ctext)))
        cset_len = len(cset)

        if not 1 < cset_len < 4:
            # We only consider inputs with two or three unique values
            logging.debug(
                "Failed to crack X-Y due to not containing two or three unique values"
            )
            return None

        logging.debug(f"String contains {cset_len} unique values: {cset}")

        # In case of three unique values, we regard the least frequent character as the delimiter
        if cset_len == 3:
            # Count each unique character in the set to determine the least frequent one
            counting_list = []
            for char in cset:
                counting_list.append(ctext.count(char))
            val, index = min((val, index) for (index, val) in enumerate(counting_list))
            delimiter = cset[index]
            logging.debug(
                f"{delimiter} occurs {val} times and is the probable delimiter"
            )
            # Remove the delimiter from the ctext and compute new cset
            ctext = ctext.replace(delimiter, "")
            cset = list(set(list(ctext)))

        # Form both variants of the substitution
        for i in range(2):
            if i:
                variants.append(ctext.replace(cset[0], "1").replace(cset[1], "0"))
            else:
                variants.append(ctext.replace(cset[0], "0").replace(cset[1], "1"))

        # Apply function to both variants and strip stray NULL characters
        for variant in variants:
            candidates.append(self.binary_to_ascii(variant).strip("\x00"))
        for i, candidate in enumerate(candidates):
            if candidate != "":
                keyinfo = f"{cset[0]} -> {i} & {cset[1]} -> {str(int(not i))}"
                result.append(CrackResult(value=candidate, key_info=keyinfo))
                logging.debug(f"X-Y cracker - Returning results: {result}")
                return result

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "expected": ParamSpec(
                desc="The expected distribution of the plaintext",
                req=False,
                config_ref=["default_dist"],
            )
        }

    def __init__(self, config: Config):
        super().__init__(config)
        self.expected = config.get_resource(self._params()["expected"])
        self.cache = config.cache



File: ciphey/basemods/Crackers/baconian.py

import re
from typing import Dict, List, Optional

from ciphey.iface import (
    Config,
    Cracker,
    CrackInfo,
    CrackResult,
    ParamSpec,
    Translation,
    registry,
)
import logging
from rich.logging import RichHandler


@registry.register
class Baconian(Cracker[str]):
    def getInfo(self, ctext: str) -> CrackInfo:
        return CrackInfo(
            success_likelihood=0.1,
            success_runtime=1e-5,
            failure_runtime=1e-5,
        )

    @staticmethod
    def getTarget() -> str:
        return "baconian"

    def attemptCrack(self, ctext: str) -> List[CrackResult]:
        """
        Attempts to decode both variants of the Baconian cipher.
        """
        logging.debug("Attempting Baconian cracker")
        candidates = []
        result = []
        ctext_decoded = ""
        ctext_decoded2 = ""

        # Convert to uppercase and replace delimiters and whitespace with nothing
        ctext = re.sub(r"[,;:\-\s]", "", ctext.upper())

        # Make sure ctext only contains A and B
        if bool(re.search(r"[^AB]", ctext)) is True:
            logging.debug("Failed to crack baconian due to non baconian character(s)")
            return None

        # Make sure ctext is divisible by 5
        ctext_len = len(ctext)
        if ctext_len % 5:
            logging.debug(
                f"Failed to decode Baconian because length must be a multiple of 5, not '{ctext_len}'"
            )
            return None

        # Split ctext into groups of 5
        ctext = " ".join(ctext[i : i + 5] for i in range(0, len(ctext), 5))
        ctext_split = ctext.split(" ")
        baconian_keys = self.BACONIAN_DICT.keys()

        # Decode I=J and U=V variant
        for i in ctext_split:
            if i in baconian_keys:
                ctext_decoded += self.BACONIAN_DICT[i]

        # Decode variant that assigns each letter a unique code
        for i in ctext_split:
            if "+" + i in baconian_keys:
                ctext_decoded2 += self.BACONIAN_DICT["+" + i]

        candidates.append(ctext_decoded)
        candidates.append(ctext_decoded2)
        for candidate in candidates:
            if candidate != "":
                if candidate == candidates[0]:
                    result.append(CrackResult(value=candidate, key_info="I=J & U=V"))
                else:
                    result.append(CrackResult(value=candidate))
        logging.debug(f"Baconian cracker - Returning results: {result}")
        return result

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "expected": ParamSpec(
                desc="The expected distribution of the plaintext",
                req=False,
                config_ref=["default_dist"],
            ),
            "dict": ParamSpec(
                desc="The Baconian alphabet dictionary to use",
                req=False,
                default="cipheydists::translate::baconian",
            ),
        }

    def __init__(self, config: Config):
        super().__init__(config)
        self.BACONIAN_DICT = config.get_resource(self._params()["dict"], Translation)
        self.expected = config.get_resource(self._params()["expected"])
        self.cache = config.cache



File: ciphey/basemods/Crackers/vigenere.py

"""
 ██████╗██╗██████╗ ██╗  ██╗███████╗██╗   ██╗
██╔════╝██║██╔══██╗██║  ██║██╔════╝╚██╗ ██╔╝
██║     ██║██████╔╝███████║█████╗   ╚████╔╝
██║     ██║██╔═══╝ ██╔══██║██╔══╝    ╚██╔╝
╚██████╗██║██║     ██║  ██║███████╗   ██║
© Brandon Skerritt
Github: brandonskerritt
"""
from distutils import util
from typing import Dict, List, Optional, Union

import cipheycore
import logging
from rich.logging import RichHandler

from ciphey.common import fix_case
from ciphey.iface import Config, Cracker, CrackInfo, CrackResult, ParamSpec, registry


@registry.register
class Vigenere(Cracker[str]):
    def getInfo(self, ctext: str) -> CrackInfo:
        if self.keysize is not None:
            analysis = self.cache.get_or_update(
                ctext,
                f"vigenere::{self.keysize}",
                lambda: cipheycore.analyse_string(
                    ctext.lower(), self.keysize, self.group
                ),
            )

            val = cipheycore.vigenere_detect(analysis, self.expected)

            logging.info(f"Vigenere has likelihood {val}")

            return CrackInfo(
                success_likelihood=val,
                # TODO: actually calculate runtimes
                success_runtime=1e-3,
                failure_runtime=1e-2,
            )

        likely_lens = self.cache.get_or_update(
            ctext,
            "vigenere::likely_lens",
            lambda: cipheycore.vigenere_likely_key_lens(
                ctext.lower(), self.expected, self.group, self.detect_p_value
            ),
        )

        # Filter out the lens that make no sense
        likely_lens = [i for i in likely_lens if i.len <= self.max_key_length]

        for keysize in likely_lens:
            # Store the analysis
            analysis = self.cache.get_or_update(
                ctext, f"vigenere::{keysize.len}", lambda: keysize.tab
            )
        if len(likely_lens) == 0:
            return CrackInfo(
                success_likelihood=0,
                # TODO: actually calculate runtimes
                success_runtime=2e-3,
                failure_runtime=2e-2,
            )

        logging.info(
            f"Vigenere has likelihood {likely_lens[0].p_value} with lens {[i.len for i in likely_lens]}"
        )

        return CrackInfo(
            success_likelihood=likely_lens[0].p_value,
            # TODO: actually calculate runtimes
            success_runtime=2e-4,
            failure_runtime=2e-4,
        )

    @staticmethod
    def getTarget() -> str:
        return "vigenere"

    def crackOne(
        self, ctext: str, analysis: cipheycore.windowed_analysis_res, real_ctext: str
    ) -> List[CrackResult]:
        possible_keys = cipheycore.vigenere_crack(
            analysis, self.expected, self.group, self.p_value
        )
        if len(possible_keys) > self.clamp:
            possible_keys = possible_keys[: self.clamp]
        logging.debug(
            f"Vigenere crack got keys: {[[i for i in candidate.key] for candidate in possible_keys]}"
        )
        return [
            CrackResult(
                value=fix_case(
                    cipheycore.vigenere_decrypt(ctext, candidate.key, self.group),
                    real_ctext,
                ),
                key_info="".join([self.group[i] for i in candidate.key]),
                misc_info=f"p-value was {candidate.p_value}",
            )
            for candidate in possible_keys[: min(len(possible_keys), 10)]
        ]

    def attemptCrack(self, ctext: str) -> List[CrackResult]:
        logging.info("Trying vigenere cipher")
        # Convert it to lower case
        if self.lower:
            message = ctext.lower()
        else:
            message = ctext

        # Analysis must be done here, where we know the case for the cache
        if self.keysize is not None:
            return self.crackOne(
                message,
                self.cache.get_or_update(
                    ctext,
                    f"vigenere::{self.keysize}",
                    lambda: cipheycore.analyse_string(
                        message, self.keysize, self.group
                    ),
                ),
                ctext,
            )

        arrs = []
        likely_lens = self.cache.get_or_update(
            ctext,
            "vigenere::likely_lens",
            lambda: cipheycore.vigenere_likely_key_lens(
                message, self.expected, self.group
            ),
        )
        possible_lens = [i for i in likely_lens]
        possible_lens.sort(key=lambda i: i.p_value)
        logging.debug(f"Got possible lengths {[i.len for i in likely_lens]}")
        # TODO: work out length
        for i in possible_lens:
            arrs.extend(
                self.crackOne(
                    message,
                    self.cache.get_or_update(
                        ctext,
                        f"vigenere::{i.len}",
                        lambda: cipheycore.analyse_string(message, i.len, self.group),
                    ),
                    ctext,
                )
            )

        logging.info(f"Vigenere returned {len(arrs)} candidates")
        return arrs

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "expected": ParamSpec(
                desc="The expected distribution of the plaintext",
                req=False,
                config_ref=["default_dist"],
            ),
            "group": ParamSpec(
                desc="An ordered sequence of chars that make up the caesar cipher alphabet",
                req=False,
                default="abcdefghijklmnopqrstuvwxyz",
            ),
            "lower": ParamSpec(
                desc="Whether or not the ciphertext should be converted to lowercase first",
                req=False,
                default=True,
            ),
            "keysize": ParamSpec(
                desc="A key size that should be used. If not given, will attempt to work it out",
                req=False,
            ),
            "p_value": ParamSpec(
                desc="The p-value to use for windowed frequency analysis",
                req=False,
                default=0.5,
            ),
            "detect_p_value": ParamSpec(
                desc="The p-value to use for the detection of Vigenere length",
                req=False,
                default=0.01,
            ),
            "clamp": ParamSpec(
                desc="The maximum number of candidates that can be returned per key len",
                req=False,
                default=10,
            ),
        }

    def __init__(self, config: Config):
        super().__init__(config)
        self.lower: Union[str, bool] = self._params()["lower"]
        if not isinstance(self.lower, bool):
            self.lower = util.strtobool(self.lower)
        self.group = list(self._params()["group"])
        self.expected = config.get_resource(self._params()["expected"])
        self.cache = config.cache
        self.keysize = self._params().get("keysize")
        if self.keysize is not None:
            self.keysize = int(self.keysize)
        self.p_value = float(self._params()["p_value"])
        self.detect_p_value = float(self._params()["detect_p_value"])
        self.clamp = int(self._params()["clamp"])
        self.max_key_length = 16



File: ciphey/basemods/Crackers/rot47.py

"""
 ██████╗██╗██████╗ ██╗  ██╗███████╗██╗   ██╗
██╔════╝██║██╔══██╗██║  ██║██╔════╝╚██╗ ██╔╝
██║     ██║██████╔╝███████║█████╗   ╚████╔╝
██║     ██║██╔═══╝ ██╔══██║██╔══╝    ╚██╔╝
╚██████╗██║██║     ██║  ██║███████╗   ██║
© Brandon Skerritt
Github: brandonskerritt
"""

from typing import Dict, List, Optional

import cipheycore
import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Cracker, CrackInfo, CrackResult, ParamSpec, registry


@registry.register
class Rot47(Cracker[str]):
    def getInfo(self, ctext: str) -> CrackInfo:
        analysis = self.cache.get_or_update(
            ctext,
            "cipheycore::simple_analysis",
            lambda: cipheycore.analyse_string(ctext),
        )

        return CrackInfo(
            success_likelihood=cipheycore.caesar_detect(analysis, self.expected),
            # TODO: actually calculate runtimes
            success_runtime=1e-5,
            failure_runtime=1e-5,
        )

    @staticmethod
    def getTarget() -> str:
        return "rot47"

    def attemptCrack(self, ctext: str) -> List[CrackResult]:
        logging.info(f"Trying ROT47 cipher on {ctext}")

        logging.debug("Beginning cipheycore simple analysis")

        # Hand it off to the core
        analysis = self.cache.get_or_update(
            ctext,
            "cipheycore::simple_analysis",
            lambda: cipheycore.analyse_string(ctext),
        )
        logging.debug("Beginning cipheycore::caesar")
        possible_keys = cipheycore.caesar_crack(
            analysis, self.expected, self.group, self.p_value
        )

        n_candidates = len(possible_keys)
        logging.info(f"ROT47 returned {n_candidates} candidates")

        if n_candidates == 0:
            logging.debug("Filtering for better results")
            analysis = cipheycore.analyse_string(ctext, self.group)
            possible_keys = cipheycore.caesar_crack(
                analysis, self.expected, self.group, self.p_value
            )

        candidates = []

        for candidate in possible_keys:
            logging.debug(f"Candidate {candidate.key} has prob {candidate.p_value}")
            translated = cipheycore.caesar_decrypt(ctext, candidate.key, self.group)
            candidates.append(CrackResult(value=translated, key_info=candidate.key))

        return candidates

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "expected": ParamSpec(
                desc="The expected distribution of the plaintext",
                req=False,
                config_ref=["default_dist"],
            ),
            "group": ParamSpec(
                desc="An ordered sequence of chars that make up the ROT47 cipher alphabet",
                req=False,
                default="""!"#$%&\'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~""",
            ),
            "p_value": ParamSpec(
                desc="The p-value to use for standard frequency analysis",
                req=False,
                default=0.01,
            )
            # TODO: add "filter" param
        }

    def __init__(self, config: Config):
        super().__init__(config)
        self.group = list(self._params()["group"])
        self.expected = config.get_resource(self._params()["expected"])
        self.cache = config.cache
        self.p_value = float(self._params()["p_value"])



File: ciphey/basemods/Crackers/xortool.py

"""
 ██████╗██╗██████╗ ██╗  ██╗███████╗██╗   ██╗
██╔════╝██║██╔══██╗██║  ██║██╔════╝╚██╗ ██╔╝
██║     ██║██████╔╝███████║█████╗   ╚████╔╝
██║     ██║██╔═══╝ ██╔══██║██╔══╝    ╚██╔╝
╚██████╗██║██║     ██║  ██║███████╗   ██║
© Brandon Skerritt
Github: bee-san
"""
from typing import Dict, List, Optional

import logging
from rich.logging import RichHandler

from xortool_ciphey import tool_main

from ciphey.iface import Config, Cracker, CrackInfo, CrackResult, ParamSpec, registry


@registry.register
class XorTool(Cracker[str]):
    def getInfo(self, ctext: str) -> CrackInfo:
        return CrackInfo(
            success_likelihood=0.1,
            # TODO: actually calculate runtimes
            success_runtime=1e-8,
            failure_runtime=1e-8,
        )

    @staticmethod
    def getTarget() -> str:
        return "xortool"

    def attemptCrack(self, ctext: str) -> List[CrackResult]:
        logging.debug("Trying xortool cipher")
        # TODO handle different charsets
        # TODO allow more config over xortool

        logging.debug(f"{ctext}")

        # https://github.com/Ciphey/xortool/discussions/4
        # for docs on this function
        try:
            result = tool_main.api(str.encode(ctext))
        except:
            logging.debug("Xor failed.")
            return

        result = CrackResult(value=result[1]["Dexored"], key_info=result[0]["keys"])

        return [result]

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "expected": ParamSpec(
                desc="The expected distribution of the plaintext",
                req=False,
                config_ref=["default_dist"],
            ),
            "p_value": ParamSpec(
                desc="The p-value to use for standard frequency analysis",
                req=False,
                default=0.01,
            ),
        }

    @staticmethod
    def score_utility() -> float:
        return 1.5

    def __init__(self, config: Config):
        super().__init__(config)
        self.expected = config.get_resource(self._params()["expected"])
        self.cache = config.cache
        self.p_value = self._params()["p_value"]



File: ciphey/basemods/Crackers/affine.py

# Community
# by https://github.com/Ozzyz

from typing import Dict, List, Optional

import cipheycore
import logging
from rich.logging import RichHandler

from ciphey.common import fix_case
from ciphey.iface import Config, Cracker, CrackInfo, CrackResult, ParamSpec, registry
from ciphey.mathsHelper import mathsHelper


@registry.register
class Affine(Cracker[str]):
    """
    Each character in the Affine Cipher is encoded with the rule E(x) = (ax + b) mod m
    m is the size of the alphabet, while a and b are the keys in the cipher. a must be coprime to b.
    The Caesar cipher is a specific case of the Affine Cipher, with a=1 and b being the shift of the cipher.
    Decryption is performed by D(x) = a_inv (x - b) mod m where a_inv is the modular multiplicative inverse of a mod m.

    In this version of the Affine Cipher, we do not allow alphabets with several instances of the same letter in different cases.
    For instance, the alphabet 'ABCdef123' is allowed, but 'AaBbCc' is not.
    """

    def getInfo(self, ctext: str) -> CrackInfo:
        return CrackInfo(
            success_likelihood=0.1,
            success_runtime=1e-5,
            failure_runtime=1e-5,
        )

    @staticmethod
    def getTarget() -> str:
        return "affine"

    def attemptCrack(self, ctext: str) -> List[CrackResult]:
        """
        Brute forces all the possible combinations of a and b to attempt to crack the cipher.
        """
        logging.debug("Attempting affine")
        candidates = []

        # a and b are coprime if gcd(a,b) is 1.
        possible_a = [
            a
            for a in range(1, self.alphabet_length)
            if mathsHelper.gcd(a, self.alphabet_length) == 1
        ]
        logging.info(
            f"Trying Affine Cracker with {len(possible_a)} a-values and {self.alphabet_length} b-values"
        )

        for a in possible_a:
            a_inv = mathsHelper.mod_inv(a, self.alphabet_length)
            # If there is no inverse, we cannot decrypt the text
            if a_inv is None:
                continue
            for b in range(self.alphabet_length):
                # Pass in lowered text. This means that we expect alphabets to not contain both 'a' and 'A'.
                translated = self.decrypt(ctext.lower(), a_inv, b, self.alphabet_length)

                candidate_probability = self.plaintext_probability(translated)
                if candidate_probability > self.plaintext_prob_threshold:
                    candidates.append(
                        CrackResult(
                            value=fix_case(translated, ctext), key_info=f"a={a}, b={b}"
                        )
                    )
        logging.info(f"Affine Cipher returned {len(candidates)} candidates")
        return candidates

    def plaintext_probability(self, translated: str) -> float:
        """
        Analyses the translated text and applies the chi squared test to see if it is a probable plaintext candidate
        Returns the probability of the chi-squared test.
        """
        analysis = cipheycore.analyse_string(translated)
        return cipheycore.chisq_test(analysis, self.expected)

    def decrypt(self, text: str, a_inv: int, b: int, m: int) -> str:
        """
        Each letter is decrypted at D(x) = a_inv (x - b) mod m where x is the char
        We treat the char value as its index in the alphabet, so if
        the alphabet is 'abcd....' and the char is 'b', it has the value 1.
        """
        return "".join([self.decryptChar(char, a_inv, b, m) for char in text])

    def decryptChar(self, char: str, a_inv: int, b: int, m: int) -> str:

        # We lower the alphabet since both ctext and alphabet need to be in the same case in order
        # to perform the shifts. The translated text will have fixed case after the translation anyways.
        # This is only necessary if the specified alphabet is uppercase.
        alphabet = [x.lower() for x in self.group]

        # Preserve characters that are not in alphabet
        if char not in alphabet:
            return char
        char_idx = alphabet.index(char)
        decrypted_char_idx = (a_inv * (char_idx - b)) % m
        return alphabet[decrypted_char_idx]

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "expected": ParamSpec(
                desc="The expected distribution of the plaintext",
                req=False,
                config_ref=["default_dist"],
            ),
            "group": ParamSpec(
                desc="An ordered sequence of chars that make up the alphabet",
                req=False,
                default="abcdefghijklmnopqrstuvwxyz",
            ),
        }

    def __init__(self, config: Config):
        super().__init__(config)
        self.group = list(self._params()["group"])
        self.expected = config.get_resource(self._params()["expected"])
        self.alphabet_length = len(self.group)
        self.cache = config.cache
        self.plaintext_prob_threshold = 0.01



File: ciphey/basemods/Crackers/__init__.py

from . import (
    affine,
    ascii_shift,
    baconian,
    caesar,
    rot47,
    soundex,
    vigenere,
    xandy,
    xortool,
)



File: ciphey/basemods/Crackers/hash.py

"""
This is Hashbuster but slightly modified to work with Ciphey.
Why reinvent the wheel?
Changes (that I can remember)
* timeout set, as hashbuster took AGES before timeout was set.
https://github.com/s0md3v/Hash-Buster
"""

import re
from typing import Dict, List, Optional

import requests
import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Cracker, CrackInfo, CrackResult, ParamSpec, T, registry

thread_count = 4


def alpha(ctext, hashtype):
    return None


def beta(ctext, hashtype):
    try:
        response = requests.get(
            "https://hashtoolkit.com/reverse-hash/?hash=", ctext, timeout=5
        ).text
    except requests.exceptions.ReadTimeout as e:
        logging.info(f"Beta failed timeout {e}")
    match = re.search(r'/generate-hash/?text=.*?"', response)
    if match:
        return match.group(1)
    return None


def gamma(ctext, hashtype):
    try:
        response = requests.get(
            "https://www.nitrxgen.net/md5db/" + ctext, timeout=5
        ).text
    except requests.exceptions.ReadTimeout as e:
        logging.info(f"Gamma failed with {e}")
    if response:
        return response
    else:
        return None


def delta(ctext, hashtype):
    return None


def theta(ctext, hashtype):
    try:
        response = requests.get(
            "https://md5decrypt.net/Api/api.php?hash=%s&hash_type=%s&email=deanna_abshire@proxymail.eu&code=1152464b80a61728"
            % (ctext, hashtype),
            timeout=5,
        ).text
    except requests.exceptions.ReadTimeout as e:
        logging.info(f"Gamma failed with {e}")
    if len(response) != 0:
        return response
    else:
        return None


md5 = [gamma, alpha, beta, theta, delta]
sha1 = [alpha, beta, theta, delta]
sha256 = [alpha, beta, theta]
sha384 = [alpha, beta, theta]
sha512 = [alpha, beta, theta]


result = {}


def crack(ctext):
    raise "Error Crack is called"


def threaded(ctext):
    resp = crack(ctext)
    if resp:
        print(ctext + " : " + resp)
        result[ctext] = resp


@registry.register
class HashBuster(Cracker[str]):
    @staticmethod
    def getTarget() -> str:
        return "hash"

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return None

    @staticmethod
    def priority() -> float:
        return 0.05

    def getInfo(self, ctext: T) -> CrackInfo:
        # TODO calculate these properly
        return CrackInfo(
            success_likelihood=0.5,
            success_runtime=5,
            failure_runtime=5,
        )

    def attemptCrack(self, ctext: T) -> List[CrackResult]:
        logging.info("Starting to crack hashes")
        result = False

        candidates = []
        if len(ctext) == 32:
            for api in md5:
                r = api(ctext, "md5")
                if result is not None or r is not None:
                    logging.debug("MD5 returns True {r}")
                    candidates.append(result, "MD5")
        elif len(ctext) == 40:
            for api in sha1:
                r = api(ctext, "sha1")
                if result is not None and r is not None:
                    logging.debug("sha1 returns true")
                    candidates.append(result, "SHA1")
        elif len(ctext) == 64:
            for api in sha256:
                r = api(ctext, "sha256")
                if result is not None and r is not None:
                    logging.debug("sha256 returns true")
                    candidates.append(result, "SHA256")
        elif len(ctext) == 96:
            for api in sha384:
                r = api(ctext, "sha384")
                if result is not None and r is not None:
                    logging.debug("sha384 returns true")
                    candidates.append(result, "SHA384")
        elif len(ctext) == 128:
            for api in sha512:
                r = api(ctext, "sha512")
                if result is not None and r is not None:
                    logging.debug("sha512 returns true")
                    candidates.append(result, "SHA512")

        # TODO what the fuck is this code?
        logging.debug(f"Hash buster returning {result}")
        # TODO add to 5.1 make this return multiple possible candidates
        return [CrackResult(value=candidates[0][0], misc_info=candidates[1][1])]

    def __init__(self, config: Config):
        super().__init__(config)



File: ciphey/basemods/Crackers/ascii_shift.py

"""
 ██████╗██╗██████╗ ██╗  ██╗███████╗██╗   ██╗
██╔════╝██║██╔══██╗██║  ██║██╔════╝╚██╗ ██╔╝
██║     ██║██████╔╝███████║█████╗   ╚████╔╝
██║     ██║██╔═══╝ ██╔══██║██╔══╝    ╚██╔╝
╚██████╗██║██║     ██║  ██║███████╗   ██║
© Brandon Skerritt
Github: brandonskerritt
"""

from typing import Dict, List, Optional

import cipheycore
import logging
from rich.logging import RichHandler

from ciphey.iface import Config, Cracker, CrackInfo, CrackResult, ParamSpec, registry


@registry.register
class Ascii_shift(Cracker[str]):
    def getInfo(self, ctext: str) -> CrackInfo:
        analysis = self.cache.get_or_update(
            ctext,
            "cipheycore::simple_analysis",
            lambda: cipheycore.analyse_string(ctext),
        )

        return CrackInfo(
            success_likelihood=cipheycore.caesar_detect(analysis, self.expected),
            # TODO: actually calculate runtimes
            success_runtime=1e-5,
            failure_runtime=1e-5,
        )

    @staticmethod
    def getTarget() -> str:
        return "ascii_shift"

    def attemptCrack(self, ctext: str) -> List[CrackResult]:
        logging.info(f"Trying ASCII shift cipher on {ctext}")

        logging.debug("Beginning cipheycore simple analysis")

        # Hand it off to the core
        analysis = self.cache.get_or_update(
            ctext,
            "cipheycore::simple_analysis",
            lambda: cipheycore.analyse_string(ctext),
        )
        logging.debug("Beginning cipheycore::caesar")
        possible_keys = cipheycore.caesar_crack(
            analysis, self.expected, self.group, self.p_value
        )

        n_candidates = len(possible_keys)
        logging.info(f"ASCII shift returned {n_candidates} candidates")

        if n_candidates == 0:
            logging.debug("Filtering for better results")
            analysis = cipheycore.analyse_string(ctext, self.group)
            possible_keys = cipheycore.caesar_crack(
                analysis, self.expected, self.group, self.p_value
            )

        candidates = []

        for candidate in possible_keys:
            logging.debug(f"Candidate {candidate.key} has prob {candidate.p_value}")
            translated = cipheycore.caesar_decrypt(ctext, candidate.key, self.group)
            candidates.append(CrackResult(value=translated, key_info=candidate.key))

        return candidates

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "expected": ParamSpec(
                desc="The expected distribution of the plaintext",
                req=False,
                config_ref=["default_dist"],
            ),
            "group": ParamSpec(
                desc="An ordered sequence of chars that make up the ASCII shift cipher alphabet",
                req=False,
                default="""\x00\x01\x02\x03\x04\x05\x06\x07\x08\t\n\x0b\x0c\r\x0e\x0f\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f !"#$%&\'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~\x7f""",
            ),
            "p_value": ParamSpec(
                desc="The p-value to use for standard frequency analysis",
                req=False,
                default=0.01,
            )
            # TODO: add "filter" param
        }

    def __init__(self, config: Config):
        super().__init__(config)
        self.group = list(self._params()["group"])
        self.expected = config.get_resource(self._params()["expected"])
        self.cache = config.cache
        self.p_value = float(self._params()["p_value"])



File: ciphey/basemods/Crackers/caesar.py

"""
 ██████╗██╗██████╗ ██╗  ██╗███████╗██╗   ██╗
██╔════╝██║██╔══██╗██║  ██║██╔════╝╚██╗ ██╔╝
██║     ██║██████╔╝███████║█████╗   ╚████╔╝
██║     ██║██╔═══╝ ██╔══██║██╔══╝    ╚██╔╝
╚██████╗██║██║     ██║  ██║███████╗   ██║
© Brandon Skerritt
Github: brandonskerritt
"""
from distutils import util
from typing import Dict, List, Optional, Union

import cipheycore
import logging
from rich.logging import RichHandler

from ciphey.common import fix_case
from ciphey.iface import Config, Cracker, CrackInfo, CrackResult, ParamSpec, registry


@registry.register
class Caesar(Cracker[str]):
    def getInfo(self, ctext: str) -> CrackInfo:
        analysis = self.cache.get_or_update(
            ctext,
            "cipheycore::simple_analysis",
            lambda: cipheycore.analyse_string(ctext),
        )

        return CrackInfo(
            success_likelihood=cipheycore.caesar_detect(analysis, self.expected),
            # TODO: actually calculate runtimes
            success_runtime=1e-5,
            failure_runtime=1e-5,
        )

    @staticmethod
    def getTarget() -> str:
        return "caesar"

    def attemptCrack(self, ctext: str) -> List[CrackResult]:
        logging.info(f"Trying caesar cipher on {ctext}")
        # Convert it to lower case
        #
        # TODO: handle different alphabets
        if self.lower:
            message = ctext.lower()
        else:
            message = ctext

        logging.debug("Beginning cipheycore simple analysis")

        # Hand it off to the core
        analysis = self.cache.get_or_update(
            ctext,
            "cipheycore::simple_analysis",
            lambda: cipheycore.analyse_string(ctext),
        )
        logging.debug("Beginning cipheycore::caesar")
        possible_keys = cipheycore.caesar_crack(
            analysis, self.expected, self.group, self.p_value
        )

        n_candidates = len(possible_keys)
        logging.info(f"Caesar returned {n_candidates} candidates")

        if n_candidates == 0:
            logging.debug("Filtering for better results")
            analysis = cipheycore.analyse_string(ctext, self.group)
            possible_keys = cipheycore.caesar_crack(
                analysis, self.expected, self.group, self.p_value
            )

        candidates = []

        for candidate in possible_keys:
            logging.debug(f"Candidate {candidate.key} has prob {candidate.p_value}")
            translated = cipheycore.caesar_decrypt(message, candidate.key, self.group)
            candidates.append(
                CrackResult(value=fix_case(translated, ctext), key_info=candidate.key)
            )

        return candidates

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "expected": ParamSpec(
                desc="The expected distribution of the plaintext",
                req=False,
                config_ref=["default_dist"],
            ),
            "group": ParamSpec(
                desc="An ordered sequence of chars that make up the caesar cipher alphabet",
                req=False,
                default="abcdefghijklmnopqrstuvwxyz",
            ),
            "lower": ParamSpec(
                desc="Whether or not the ciphertext should be converted to lowercase first",
                req=False,
                default=True,
            ),
            "p_value": ParamSpec(
                desc="The p-value to use for standard frequency analysis",
                req=False,
                default=0.01,
            )
            # TODO: add "filter" param
        }

    def __init__(self, config: Config):
        super().__init__(config)
        self.lower: Union[str, bool] = self._params()["lower"]
        if not isinstance(self.lower, bool):
            self.lower = util.strtobool(self.lower)
        self.group = list(self._params()["group"])
        self.expected = config.get_resource(self._params()["expected"])
        self.cache = config.cache
        self.p_value = float(self._params()["p_value"])



File: ciphey/basemods/Crackers/soundex.py

import re
from typing import Dict, List, Optional

import logging
from rich.logging import RichHandler

from ciphey.iface import (
    Config,
    Cracker,
    CrackInfo,
    CrackResult,
    ParamSpec,
    Translation,
    registry,
)


@registry.register
class Soundex(Cracker[str]):
    def getInfo(self, ctext: str) -> CrackInfo:
        return CrackInfo(
            success_likelihood=0.1,
            success_runtime=1e-5,
            failure_runtime=1e-5,
        )

    @staticmethod
    def getTarget() -> str:
        return "soundex"

    def attemptCrack(self, ctext: str) -> List[CrackResult]:
        """
        Attempts to crack Soundex by generating all possible combinations.
        """
        logging.debug("Attempting Soundex cracker")
        word_list = []
        sentences = []
        result = []

        # Convert to uppercase and replace delimiters and whitespace with nothing
        ctext = re.sub(r"[,;:\-\s]", "", ctext.upper())

        # Make sure ctext contains only A-Z and 0-9
        if bool(re.search(r"[^A-Z0-9]", ctext)) is True:
            logging.debug("Failed to crack soundex due to non soundex character(s)")
            return None

        # Make sure ctext is divisible by 4
        ctext_len = len(ctext)
        if ctext_len % 4:
            logging.debug(
                f"Failed to decode Soundex because length must be a multiple of 4, not '{ctext_len}'"
            )
            return None

        # Split ctext into groups of 4
        ctext = " ".join(ctext[i : i + 4] for i in range(0, len(ctext), 4))
        ctext_split = ctext.split(" ")
        soundex_keys = self.SOUNDEX_DICT.keys()

        # Find all words that correspond to each given soundex code
        for code in ctext_split:
            if code in soundex_keys:
                word_list.append(self.SOUNDEX_DICT[code])

        logging.info(f"Possible words for given encoded text: {word_list}")

        # Find all possible sentences
        self.getSentenceCombo(
            word_list,
            sentences,
            self.frequency_dict,
            self.sentence_freq,
            self.word_freq,
        )

        sorted_sentences = self.sortlistwithdict(sentences, self.frequency_dict)

        for sentence in sorted_sentences:
            result.append(CrackResult(value=sentence))

        logging.debug(f"Soundex cracker - Returning results: {result}")
        return result

    def sortlistwithdict(self, listtosort, hashes):
        """
        This function uses the sum of ranks (based on frequency) of each word in each
        sentence and sorts them according to it.
        """
        return sorted(listtosort, key=lambda x: hashes[x])

    def getSentenceCombo(
        self, A, sentences, frequency_dict, sentence_freq, word_freq, result="", n=0
    ):
        """
        This function uses recursion to generate a list of sentences from all possible
        words for a given set of soundex codes.
        """
        logging.debug("Creating all possible sentences from Soundex")
        if n == len(A):
            sentences.append(result[1:])
            for word in result[1:].split():
                # Adding the rank of each word to find out the sentence's net frequency
                if word in word_freq:
                    sentence_freq += word_freq.index(word)
                # If the word isn't in the frequency list then it's a very uncommon word
                # so we add a large number (5000)
                else:
                    sentence_freq += 5000
            frequency_dict[result[1:]] = sentence_freq
            sentence_freq = 0
            return

        for word in A[n]:
            out = result + " " + word
            self.getSentenceCombo(
                A, sentences, frequency_dict, sentence_freq, word_freq, out, n + 1
            )

    @staticmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        return {
            "dict": ParamSpec(
                desc="The Soundex dictionary to use",
                req=False,
                default="cipheydists::translate::soundex",
            ),
            "freq": ParamSpec(
                desc="The word frequency dictionary to use",
                req=False,
                default="cipheydists::list::English5000Freq",
            ),
        }

    def __init__(self, config: Config):
        super().__init__(config)
        self.SOUNDEX_DICT = config.get_resource(self._params()["dict"], Translation)
        self.word_freq = config.get_resource(self._params()["freq"], Translation)
        self.frequency_dict = {}
        self.sentence_freq = 0



File: ciphey/iface/__init__.py

from ._config import Config

from ._modules import (
    Checker,
    Cracker,
    CrackInfo,
    CrackResult,
    Decoder,
    DecoderComparer,
    Distribution,
    ParamSpec,
    PolymorphicChecker,
    ResourceLoader,
    Searcher,
    SearchLevel,
    SearchResult,
    T,
    Translation,
    U,
    WordList,
    pretty_search_results,
)
from ._registry import get_args, get_origin

from ._fwd import registry



File: ciphey/iface/_registry.py

from typing import Any, Dict, List, Optional, Set, Tuple, Type, Union

try:
    from typing import get_args, get_origin
except ImportError:
    from typing_inspect import get_origin, get_args

from . import _fwd
from ._modules import *


class Registry:
    # I was planning on using __init_subclass__, but that is incompatible with dynamic type creation when we have
    # generic keys

    RegElem = Union[List[Type], Dict[Type, "RegElem"]]

    _reg: Dict[Type, RegElem] = {}
    _names: Dict[str, Tuple[Type, Set[Type]]] = {}
    _targets: Dict[str, Dict[Type, List[Type]]] = {}
    _modules = {Checker, Cracker, Decoder, ResourceLoader, Searcher, PolymorphicChecker}

    def _register_one(self, input_type, module_base, module_args):
        if len(module_args) == 0:
            self._reg.setdefault(module_base, []).append(input_type)
            return

        target_reg = self._reg.setdefault(module_base, {})
        # Seek to the given type
        for subtype in module_args[0:-1]:
            target_reg = target_reg.setdefault(subtype, {})
        target_reg.setdefault(module_args[-1], []).append(input_type)

    def _real_register(self, input_type: type, *args) -> Type:
        name = input_type.__name__.lower()
        name_target = self._names[name] = (input_type, set())

        if issubclass(input_type, Targeted):
            target = input_type.getTarget()
        else:
            target = None

        if issubclass(input_type, Searcher):
            module_type = module_base = Searcher
            module_args = ()
        else:
            module_type: Optional[Type] = None
            module_base = None

            # Work out what module type this is
            if len(args) == 0 and hasattr(input_type, "__orig_bases__"):
                for i in input_type.__orig_bases__:
                    if module_type is not None:
                        raise TypeError(
                            f"Type derived from multiple registrable base classes {i} and {module_type}"
                        )
                    module_base = get_origin(i)
                    if module_base not in self._modules:
                        continue
                    module_type = i
            else:
                for i in self._modules:
                    if not issubclass(input_type, i):
                        continue
                    if module_type is not None:
                        raise TypeError(
                            f"Type derived from multiple registrable base classes {i} and {module_type}"
                        )
                    module_type = i
            if module_type is None:
                raise TypeError("No registrable base class")

            # Replace input type with polymorphic checker if required
            if issubclass(input_type, Checker):
                if len(args) == 0:
                    arg = [
                        get_args(i)
                        for i in input_type.__orig_bases__
                        if get_origin(i) == Checker
                    ][0]
                    if len(arg) != 1:
                        raise TypeError("No argument for Checker")
                    input_type = input_type.convert({arg[0]})
                else:
                    input_type = input_type.convert(set(args))
                self._register_one(input_type, PolymorphicChecker, [])
                # Refresh the names with the new type
                name_target = self._names[name] = (input_type, {PolymorphicChecker})

            # Now handle the difference between register and register_multi
            if len(args) == 0:
                if module_type is PolymorphicChecker:
                    module_base = PolymorphicChecker
                elif module_base is None:
                    raise TypeError("No type argument given")
                self._register_one(input_type, module_base, get_args(module_type))
                name_target[1].add(module_base)
            else:
                if module_base is not None:
                    raise TypeError(f"Redundant type argument for {module_type}")
                module_base = module_type
                for module_args in args:
                    # Correct missing brackets
                    if not isinstance(module_args, tuple):
                        module_args = (module_args,)

                    self._register_one(input_type, module_base, module_args)
                    name_target[1].add(module_type[module_args])

        name_target[1].add(module_type)

        if target is not None and issubclass(module_base, Targeted):
            self._targets.setdefault(target, {}).setdefault(module_type, []).append(
                input_type
            )

        return input_type

    def register(self, input_type):
        return self._real_register(input_type)

    def register_multi(self, *x):
        return lambda input_type: self._real_register(input_type, *x)

    def __getitem__(self, i: type) -> Optional[Any]:
        target_type = get_origin(i)
        # Check if this is a non-generic type, and return the whole dict if it is
        if target_type is None:
            return self._reg[i]

        target_subtypes = get_args(i)
        target_list = self._reg.setdefault(target_type, {})
        for subtype in target_subtypes:
            target_list = target_list.setdefault(subtype, {})
        return target_list

    def get_named(self, name: str, type_constraint: Type = None) -> Any:
        ret = self._names[name.lower()]
        if type_constraint and type_constraint not in ret[1]:
            raise TypeError(f"Type mismatch: wanted {type_constraint}, got {ret[1]}")
        return ret[0]

    def get_targeted(
        self, target: str, type_constraint: Type = None
    ) -> Optional[Union[Dict[Type, Set[Type]], Set[Type]]]:
        x = self._targets.get(target)
        if x is None or type_constraint is None:
            return x
        return x.get(type_constraint)

    def get_all_names(self) -> List[str]:
        return list(self._names.keys())

    def __str__(self):
        return f"ciphey.iface.Registry {{_reg: {self._reg}, _names: {self._names}, _targets: {self._targets}}}"


_fwd.registry = Registry()



File: ciphey/iface/_fwd.py

registry = None
config = type(None)



File: ciphey/iface/_config.py

import datetime
import os
import pydoc
from typing import Any, Callable, Dict, List, Optional, Type, Union

import appdirs
import yaml
import logging
from rich.logging import RichHandler

from . import _fwd
from ._modules import PolymorphicChecker, ResourceLoader, Searcher


class Cache:
    """Used to track state between levels of recursion to stop infinite loops, and to optimise repeating actions"""

    def __init__(self):
        self._cache: Dict[Any, Dict[str, Any]] = {}

    def mark_ctext(self, ctext: Any) -> bool:
        if (isinstance(ctext, str) or isinstance(ctext, bytes)) and len(ctext) < 4:
            logging.debug(f"Candidate {ctext.__repr__()} too short!")
            return False

        if ctext in self._cache:
            logging.debug(f"Deduped {ctext.__repr__()}")
            return False

        logging.debug(f"New ctext {ctext.__repr__()}")

        self._cache[ctext] = {}
        return True

    def get_or_update(self, ctext: Any, keyname: str, get_value: Callable[[], Any]):
        # Should have been marked first
        target = self._cache[ctext]
        res = target.get(keyname)
        if res is not None:
            return res

        val = get_value()
        target[keyname] = val
        return val

    def try_get(self, ctext: Any, keyname: str):
        return self._cache[ctext].get(keyname)


def split_resource_name(full_name: str) -> (str, str):
    return full_name.split("::", 1)


class Config:
    def __init__(self):
        self.verbosity: int = 0
        self.searcher: str = "ausearch"
        self.params: Dict[str, Dict[str, Union[str, List[str]]]] = {}
        self.format: str = "str"
        self.modules: List[str] = []
        self.checker: str = "ezcheck"
        self.default_dist: str = "cipheydists::dist::english"
        self.timeout: Optional[int] = None
        self._inst: Dict[type, Any] = {}
        self.objs: Dict[str, Any] = {}
        self.cache: Cache = Cache()

    @staticmethod
    def get_default_dir() -> str:
        return appdirs.user_config_dir("ciphey")

    def merge_dict(self, config_file: Optional[Dict[str, Any]]):
        if config_file is None:
            return
        for a, b in config_file.items():
            self.update(a, b)

    def load_file(
        self,
        path: str = os.path.join(get_default_dir.__func__(), "config.yml"),
        create=False,
    ):
        try:
            with open(path, "r+") as file:
                return self.merge_dict(yaml.safe_load(file))
        except FileNotFoundError:
            if create:
                open(path, "w+")

    def instantiate(self, t: type) -> Any:
        """
        Used to enable caching of a instantiated type after the configuration has settled
        """
        # We cannot use set default as that would construct it again, and throw away the result
        res = self._inst.get(t)
        if res is not None:
            return res
        ret = t(self)
        self._inst[t] = ret
        return ret

    def __call__(self, t: type) -> Any:
        return self.instantiate(t)

    def update(self, attrname: str, value: Optional[Any]):
        if value is not None:
            setattr(self, attrname, value)

    def update_param(self, owner: str, name: str, value: Optional[Any]):
        if value is None:
            return

        target = self.params.setdefault(owner, {})

        if _fwd.registry.get_named(owner).getParams()[name].list:
            target.setdefault(name, []).append(value)
        else:
            target[name] = value

    def update_format(self, value: Optional[str]):
        if value is not None:
            self.format = value

    def load_objs(self):
        # Basic type conversion
        if self.timeout is not None:
            self.objs["timeout"] = datetime.timedelta(seconds=int(self.timeout))
        self.objs["format"] = pydoc.locate(self.format)

        # Checkers do not depend on any other config object
        logging.debug(f"Registry is {_fwd.registry._reg[PolymorphicChecker]}")
        self.objs["checker"] = self(
            _fwd.registry.get_named(self.checker, PolymorphicChecker)
        )
        # Searchers only depend on checkers
        self.objs["searcher"] = self(_fwd.registry.get_named(self.searcher, Searcher))

    def update_log_level(self, verbosity: Optional[int]):
        if verbosity is None:
            return
        self.verbosity = verbosity

        if verbosity == 0:
            self.verbosity = logging.WARNING
        elif verbosity == 1:
            self.verbosity = logging.INFO
        elif verbosity >= 2:
            self.verbosity = logging.DEBUG
        else:
            logging.disable(logging.CRITICAL)
            return

        # https://rich.readthedocs.io/en/latest/logging.html for more on RichHandler
        logging.basicConfig(
            level=self.verbosity,
            datefmt="[%X]",
            handlers=[RichHandler(markup=True, rich_tracebacks=True)],
        )
        logging.debug(f"Verbosity set to level {verbosity}")

    def load_modules(self):
        import importlib.util

        for i in self.modules:
            spec = importlib.util.spec_from_file_location("ciphey.module_load_site", i)
            mod = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(mod)

        logging.info(f"Loaded modules {_fwd.registry.get_all_names()}")

    def complete_config(self) -> "Config":
        """This does all the loading for the config, and then returns itself"""
        self.load_modules()
        self.load_objs()
        self.update_log_level(self.verbosity)
        return self

    def get_resource(self, res_name: str, t: Optional[Type] = None) -> Any:
        logging.debug(f"Loading resource {res_name} of type {t}")

        # FIXME: Actually returns obj of type `t`, but python is bad
        loader, name = split_resource_name(res_name)
        if t is None:
            return self(_fwd.registry.get_named(loader, ResourceLoader))(name)
        else:
            return self(_fwd.registry.get_named(loader, ResourceLoader[t]))(name)

    # Setter methods for cleaner library API
    def set_verbosity(self, i):
        self.update_log_level(i)
        return self

    def set_spinner(self, spinner):
        self.objs["spinner"] = spinner

    def pause_spinner_handle(self):
        spinner = self.objs.get("spinner")

        class PausedSpinner:
            def __enter__(self):
                if spinner is not None:
                    spinner.stop()

            def __exit__(self, exc_type, exc_val, exc_tb):
                if spinner is not None:
                    spinner.start()

        return PausedSpinner()

    @staticmethod
    def library_default():
        """The default config for use in a library"""
        return Config().set_verbosity(-1)

    def __str__(self):
        return str(
            {
                "verbosity": self.verbosity,
                "searcher": self.searcher,
                "params": self.params,
                "format": self.format,
                "modules": self.modules,
                "checker": self.checker,
                "default_dist": self.default_dist,
                "timeout": self.timeout,
            }
        )


_fwd.config = Config



File: ciphey/iface/_modules.py

from abc import ABC, abstractmethod
from typing import Any, Dict, Generic, List, NamedTuple, Optional, Set, Type, TypeVar

from rich import box
from rich.console import Console
from rich.markup import escape
from rich.table import Table

from ._fwd import config as Config

T = TypeVar("T")
U = TypeVar("U")

console = Console()


class ParamSpec(NamedTuple):
    """
    Attributes:
        req         Whether this argument is required
        desc        A description of what this argument does
        default     The default value for this argument. Ignored if req == True or configPath is not None
        config_ref  The path to the config that should be the default value
        list        Whether this parameter is in the form of a list, and can therefore be specified more than once
        visible     Whether the user can tweak this via the command line
    """

    req: bool
    desc: str
    default: Optional[Any] = None
    list: bool = False
    config_ref: Optional[List[str]] = None
    visible: bool = True


class ConfigurableModule(ABC):
    @staticmethod
    @abstractmethod
    def getParams() -> Optional[Dict[str, ParamSpec]]:
        """
        Returns a dictionary of `argument name: argument specification`
        """
        pass

    def _checkParams(self):
        """
        Fills the given params dict with default values where arguments are not given,
        using None as the default value for default values
        """

        params = self._params()
        config = self._config()

        for key, value in self.getParams().items():
            # If we already have it, then we don't need to do anything
            if key in params:
                continue
            # If we don't have it, but it's required, then fail
            if value.req:
                raise KeyError(
                    f"Missing required param {key} for {type(self).__name__.lower()}"
                )
            # If it's a reference by default, fill that in
            if value.config_ref is not None:
                tmp = getattr(config, value.config_ref[0])
                params[key] = (
                    tmp[value.config_ref[1:]] if len(value.config_ref) > 1 else tmp
                )
            # Otherwise, put in the default value (if it exists)
            elif value.default is not None:
                params[key] = value.default

    def _params(self):
        return self._params_obj

    def _config(self):
        return self._config_obj

    @abstractmethod
    def __init__(self, config: Config):
        self._config_obj = config
        if self.getParams() is not None:
            self._params_obj = config.params.setdefault(type(self).__name__.lower(), {})
            self._checkParams()


class Targeted(ABC):
    @staticmethod
    @abstractmethod
    def getTarget() -> str:
        """Should return the target that this object attacks/decodes"""
        pass


class PolymorphicChecker(ConfigurableModule):
    @abstractmethod
    def check(self, text) -> Optional[str]:
        """Should return some description (or an empty string) on success, otherwise return None"""
        pass

    @abstractmethod
    def getExpectedRuntime(self, text) -> float:
        pass

    def __call__(self, *args):
        return self.check(*args)

    @abstractmethod
    def __init__(self, config: Config):
        super().__init__(config)


class Checker(Generic[T], ConfigurableModule):
    @abstractmethod
    def check(self, text: T) -> Optional[str]:
        """Should return some description (or an empty string) on success, otherwise return None"""
        pass

    @abstractmethod
    def getExpectedRuntime(self, text: T) -> float:
        pass

    def __call__(self, *args):
        return self.check(*args)

    @abstractmethod
    def __init__(self, config: Config):
        super().__init__(config)

    @classmethod
    def convert(cls, expected: Set[type]):
        class PolyWrapperClass(PolymorphicChecker):
            @staticmethod
            def getParams() -> Optional[Dict[str, ParamSpec]]:
                return cls.getParams()

            def check(self, text) -> Optional[str]:
                """Should return some description (or an empty string) on success, otherwise return None"""
                if type(text) not in expected:
                    return None
                else:
                    return self._base.check(text)

            def getExpectedRuntime(self, text) -> float:
                if type(text) not in expected:
                    return 0
                else:
                    return self._base.getExpectedRuntime(text)

            def __init__(self, config: Config):
                super().__init__(config)
                # This is easier than inheritance
                self._base = cls(config)

        PolyWrapperClass.__name__ = cls.__name__

        return PolyWrapperClass


# class Detector(Generic[T], ConfigurableModule, KnownUtility, Targeted):
#     @abstractmethod
#     def scoreLikelihood(self, ctext: T) -> Dict[str, float]:
#         """Should return a dictionary of (cipher_name: score)"""
#         pass
#
#     def __call__(self, *args): return self.scoreLikelihood(*args)
#
#     @abstractmethod
#     def __init__(self, config: Config): super().__init__(config)


class Decoder(Generic[T], ConfigurableModule, Targeted):
    """Represents the undoing of some encoding into a different (or the same) type"""

    @abstractmethod
    def decode(self, ctext: T) -> Optional[U]:
        pass

    @staticmethod
    @abstractmethod
    def priority() -> float:
        """What proportion of decodings are this?"""
        pass

    def __call__(self, *args):
        return self.decode(*args)

    @abstractmethod
    def __init__(self, config: Config):
        super().__init__(config)


class DecoderComparer:
    value: Type[Decoder]

    def __le__(self, other: "DecoderComparer"):
        return self.value.priority() <= other.value.priority()

    def __ge__(self, other: "DecoderComparer"):
        return self.value.priority() >= other.value.priority()

    def __lt__(self, other: "DecoderComparer"):
        return self.value.priority() < other.value.priority() and self != other

    def __gt__(self, other: "DecoderComparer"):
        return self.value.priority() > other.value.priority() and self != other

    def __init__(self, value: Type[Decoder]):
        self.value = value

    def __repr__(self):
        return f"<DecoderComparer {self.value}:{self.value.priority()}>"


class CrackResult(NamedTuple):
    # TODO consider using Generic[T] again for value's type once
    # https://bugs.python.org/issue36517 is resolved
    value: Any
    key_info: Optional[str] = None
    misc_info: Optional[str] = None


class CrackInfo(NamedTuple):
    success_likelihood: float
    success_runtime: float
    failure_runtime: float


class Cracker(Generic[T], ConfigurableModule, Targeted):
    @abstractmethod
    def getInfo(self, ctext: T) -> CrackInfo:
        """Should return some informed guesses on resource consumption when run on `ctext`"""
        pass

    @abstractmethod
    def attemptCrack(self, ctext: T) -> List[CrackResult]:
        """
        This should attempt to crack the cipher `target`, and return a list of candidate solutions
        """
        # FIXME: Actually CrackResult[T], but python complains
        pass

    def __call__(self, *args):
        return self.attemptCrack(*args)

    @abstractmethod
    def __init__(self, config: Config):
        super().__init__(config)


class ResourceLoader(Generic[T], ConfigurableModule):
    @abstractmethod
    def whatResources(self) -> Optional[Set[str]]:
        """
        Return a set of the names of instances T you can provide.
        The names SHOULD be unique amongst ResourceLoaders of the same type

        These names will be exposed as f"{self.__name__}::{name}", use split_resource_name to recover this

        If you cannot reasonably determine what resources you provide, return None instead
        """
        pass

    @abstractmethod
    def getResource(self, name: str) -> T:
        """
        Returns the requested distribution

        The behavior is undefined if `name not in self.what_resources()`
        """
        pass

    def __call__(self, *args):
        return self.getResource(*args)

    def __getitem__(self, *args):
        return self.getResource(*args)

    @abstractmethod
    def __init__(self, config: Config):
        super().__init__(config)


class SearchLevel(NamedTuple):
    name: str
    result: CrackResult

    @staticmethod
    def input(ctext: Any):
        return SearchLevel(name="input", result=CrackResult(ctext))


class SearchResult(NamedTuple):
    path: List[SearchLevel]
    check_res: str


class Searcher(ConfigurableModule):
    """A very basic interface for code that plans out how to crack the ciphertext"""

    @abstractmethod
    def search(self, ctext: Any) -> Optional[SearchResult]:
        """Returns the path to the correct ciphertext"""
        pass

    @abstractmethod
    def __init__(self, config: Config):
        super().__init__(config)


def pretty_search_results(res: SearchResult, display_intermediate: bool = False) -> str:
    # TODO what is display_intermediate
    ret: str = ""
    table = Table(show_header=False, box=box.ROUNDED, safe_box=False)
    # Only print the checker if we need to. Normal people don't know what
    # "quadgrams", "brandon", "json checker" is.
    # We print the checker if its regex or another language, so long as it starts with:
    # "The" like "The plaintext is a Uniform Resource Locator (URL)."
    if len(res.check_res) != 0 and ("The" == res.check_res[0:3] or "Passed" == res.check_res[0:6]):
        ret += f"{res.check_res}\n"

    def add_one():
        out = ""
        if i.name == "utf8":
            out += f"   [#808080]{i.name}[/#808080]\n"
        else:
            out += f"   {i.name}"
        already_broken = False
        if i.result.key_info is not None:
            out += f":\n    Key: {i.result.key_info}\n"
            already_broken = True
        if i.result.misc_info is not None:
            if not already_broken:
                out += ":\n"
            out += f"    Misc: {i.result.misc_info}\n"
            already_broken = True
        if display_intermediate:
            if not already_broken:
                out += ":\n"
            out += f'    Value: "{i.result.value}"\n'
            already_broken = True
        if not already_broken:
            out += "\n"
        return out, already_broken

    # Skip the 'input' and print in order
    already_broken = False
    out = ""
    for i in res.path[1:]:
        output, already_broken = add_one()
        out += output

    if out:
        if len(out.split("\n")) > 1:
            ret += "Formats used:\n"
        else:
            ret += "Format used:\n"
        ret += out

    # Remove trailing newline
    ret = ret[:-1]

    # If we didn't show intermediate steps, then print the final result
    if already_broken:
        ret += f"""\nPlaintext: [bold green]"{escape(res.path[-1].result.value)}"[bold green]"""
    else:
        ret += f"""Plaintext: [bold green]"{escape(res.path[-1].result.value)}"[bold green]"""

    table.add_row(ret)
    return table


# Some common collection types
Distribution = Dict[str, float]
Translation = Dict[str, str]
WordList = Set[str]



<p align="center">
Translations <br>
<a href=https://github.com/Ciphey/Ciphey/tree/master/translations/de/README.md>🇩🇪 DE   </a>
<a href=https://github.com/Ciphey/Ciphey/tree/master/translations/fr/README.md>🇫🇷 FR   </a>
<a href=https://github.com/Ciphey/Ciphey/tree/master/translations/hu/README.md>🇭🇺 HU   </a>
<a href=https://github.com/Ciphey/Ciphey/tree/master/translations/id/README.md>🇮🇩 ID   </a>
<a href=https://github.com/Ciphey/Ciphey/tree/master/translations/it/README.md>🇮🇹 IT   </a>
<a href=https://github.com/Ciphey/Ciphey/tree/master/translations/nl/README.md>🇳🇱 NL   </a>
<a href=https://github.com/Ciphey/Ciphey/tree/master/translations/pt-br/README.md>🇧🇷 PT-BR   </a>
<a href=https://github.com/Ciphey/Ciphey/tree/master/translations/ru/README.md>🇷🇺 RU   </a>
<a href=https://github.com/Ciphey/Ciphey/tree/master/translations/zh/README.md>🇨🇳 ZH   </a>
<a href="https://github.com/Ciphey/Ciphey/tree/master/translations/th/README.md">🇹🇭 TH   </a>
 <br><br>
➡️
<a href="https://github.com/Ciphey/Ciphey/wiki">Documentation</a> |
<a href="https://discord.gg/zYTM3rZM4T">Discord</a> |
 <a href="https://github.com/Ciphey/Ciphey/wiki/Installation">Installation Guide</a>
 ⬅️

<br>
  <img src="https://github.com/Ciphey/Ciphey/raw/master/Pictures_for_README/binoculars.png" alt="Ciphey">
</p>

<p align="center">
<img src="https://pepy.tech/badge/ciphey">
 <img src="https://pepy.tech/badge/ciphey/month">
  <a href="https://discord.gg/zYTM3rZM4T"><img alt="Discord" src="https://img.shields.io/discord/754001738184392704"></a>
<a href="https://pypi.org/project/ciphey/"><img src="https://img.shields.io/pypi/v/ciphey.svg"></a>
  <img src="https://img.shields.io/badge/License-MIT-yellow.svg" alt="Ciphey">

<br>
Fully automated decryption/decoding/cracking tool using natural language processing & artificial intelligence, along with some common sense.
</p>
<hr>

## [Installation Guide](https://github.com/Ciphey/Ciphey/wiki/Installation)

| <p align="center"><a href="https://pypi.org/project/ciphey">🐍 Python | <p align="center"><a href="https://hub.docker.com/r/remnux/ciphey">🐋 Docker (Universal) | <p align="center"><a href="https://ports.macports.org/port/ciphey/summary">🍎 MacPorts (macOS) | <p align="center"><a href="https://formulae.brew.sh/formula/ciphey">🍺 Homebrew (macOS/Linux) |
| --------------------------------------------------------------------- | --------------------------------------------------------------------------------- | --------------------------------------------------------------------------------- |--------------------------------------------------------------------------------- |
| <p align="center"><img src="https://github.com/Ciphey/Ciphey/raw/master/Pictures_for_README/python.png" /></p>    | <p align="center"><img src="https://github.com/Ciphey/Ciphey/raw/master/Pictures_for_README/docker.png" /></p> | <p align="center"><img src="https://github.com/Ciphey/Ciphey/raw/master/Pictures_for_README/macports.png" /></p> | <p align="center"><img src="https://github.com/Ciphey/Ciphey/raw/master/Pictures_for_README/homebrew.png" /></p> |
| `python3 -m pip install ciphey --upgrade` | `docker run -it --rm remnux/ciphey` | `sudo port install ciphey` | `brew install ciphey` |

| Linux                                                                                                                   | Mac OS                                                                                                                     | Windows                                                                                                                   |
| ----------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------- |
| ![GitHub Workflow Status](https://img.shields.io/github/workflow/status/ciphey/ciphey/Python%20application?label=Linux) | ![GitHub Workflow Status](https://img.shields.io/github/workflow/status/ciphey/ciphey/Python%20application?label=Mac%20OS) | ![GitHub Workflow Status](https://img.shields.io/github/workflow/status/ciphey/ciphey/Python%20application?label=Windows) |

<hr>

# 🤔 What is this?

Input encrypted text, get the decrypted text back.

> "What type of encryption?"

That's the point. You don't know, you just know it's possibly encrypted. Ciphey will figure it out for you.

Ciphey can solve most things in 3 seconds or less.

<p align="center" href="https://asciinema.org/a/336257">
  <img src="https://github.com/Ciphey/Ciphey/raw/master/Pictures_for_README/index.gif" alt="Ciphey demo">
</p>

Ciphey aims to be a tool to automate a lot of decryptions & decodings such as multiple base encodings, classical ciphers, hashes or more advanced cryptography.

If you don't know much about cryptography, or you want to quickly check the ciphertext before working on it yourself, Ciphey is for you.

**The technical part.** Ciphey uses a custom built artificial intelligence module (_AuSearch_) with a _Cipher Detection Interface_ to approximate what something is encrypted with. And then a custom-built, customisable natural language processing _Language Checker Interface_, which can detect when the given text becomes plaintext.

No neural networks or bloated AI here. We only use what is fast and minimal.

And that's just the tip of the iceberg. For the full technical explanation, check out our [documentation](https://github.com/Ciphey/Ciphey/wiki).

# ✨ Features

- **50+ encryptions/encodings supported** such as binary, Morse code and Base64. Classical ciphers like the Caesar cipher, Affine cipher and the Vigenere cipher. Along with modern encryption like repeating-key XOR and more. **[For the full list, click here](https://github.com/Ciphey/Ciphey/wiki/Supported-Ciphers)**
- **Custom Built Artificial Intelligence with Augmented Search (AuSearch) for answering the question "what encryption was used?"** Resulting in decryptions taking less than 3 seconds.
- **Custom built natural language processing module** Ciphey can determine whether something is plaintext or not. Whether that plaintext is JSON, a CTF flag, or English, Ciphey can get it in a couple of milliseconds.
- **Multi Language Support** at present, only German & English (with AU, UK, CAN, USA variants).
- **Supports encryptions and hashes** Which the alternatives such as CyberChef Magic do not.
- **[C++ core](https://github.com/Ciphey/CipheyCore)** Blazingly fast.

# 🔭 Ciphey vs CyberChef

## 🔁 Base64 Encoded 42 times

<table>
  <tr>
  <th>Name</th>
    <th>⚡ Ciphey ⚡ </th>
    <th>🐢 CyberChef 🐢</th>
  </tr>
  <tr>
  <th>Gif</th>
    <td><img src="https://github.com/Ciphey/Ciphey/raw/master/Pictures_for_README/ciphey_gooder_cyberchef.gif" alt="The guy she tells you not to worry about"></td>
    <td><img src="https://github.com/Ciphey/Ciphey/raw/master/Pictures_for_README/not_dying.gif" alt="You"></td>
  </tr>
  <tr>
  <th>Time</th>
    <td>2 seconds</td>
    <td>6 seconds</td>
  </tr>
    <tr>
  <th>Setup</th>
    <td><ul><li>Run ciphey on the file</li></ul></td>
    <td><ul><li>Set the regex param to "{"</li><li>You need to know how many times to recurse</li><li>You need to know it's Base64 all the way down</li><li>You need to load CyberChef (it's a bloated JS app)</li><li>Know enough about CyberChef to create this pipeline</li><li>Invert the match</li></ul></td>
  </tr>
</table>

<sub><b>Note</b> The gifs may load at different times, so one may appear significantly faster than another.</sub><br>
<sub><b>A note on magic </b>CyberChef's most similar feature to Ciphey is Magic. Magic fails instantly on this input and crashes. The only way we could force CyberChef to compete was to manually define it.</sub>

We also tested CyberChef and Ciphey with a **6gb file**. Ciphey cracked it in **5 minutes and 54 seconds**. CyberChef crashed before it even started.

## 📊 Ciphey vs Katana vs CyberChef Magic

| **Name**                                   | ⚡ Ciphey ⚡ | 🗡️ Katana 🗡️ | 🐢 CyberChef Magic 🐢 |
| ------------------------------------------ | ------------ | ------------ | --------------------- |
| Advanced Language Checker                  | ✅           | ❌           | ✅                    |
| Supports Encryptions                       | ✅           | ✅           | ❌                    |
| Releases named after Dystopian themes 🌃   | ✅           | ❌           | ❌                    |
| Supports hashes                            | ✅           | ✅           | ❌                    |
| Easy to set up                             | ✅           | ❌           | ✅                    |
| Can guess what something is encrypted with | ✅           | ❌           | ❌                    |
| Created for hackers by hackers             | ✅           | ✅           | ❌                    |

# 🎬 Getting Started

If you're having trouble with installing Ciphey, [read this.](https://github.com/Ciphey/Ciphey/wiki/Common-Issues-&-Their-Solutions)

## ‼️ Important Links (Docs, Installation guide, Discord Support)

| Installation Guide                                                          | Documentation                                             | Discord                                     | Docker Image (from REMnux)                                                                          |
| --------------------------------------------------------------------------- | --------------------------------------------------------- | ------------------------------------------- | --------------------------------------------------------------------------------------------------- |
| 📖 [Installation Guide](https://github.com/Ciphey/Ciphey/wiki/Installation) | 📚 [Documentation](https://github.com/Ciphey/Ciphey/wiki) | 🦜 [Discord](https://discord.gg/zYTM3rZM4T) | 🐋 [Docker Documentation](https://docs.remnux.org/run-tools-in-containers/remnux-containers#ciphey) |

## 🏃‍♀️Running Ciphey

There are 3 ways to run Ciphey.

1. File Input `ciphey -f encrypted.txt`
2. Unqualified input `ciphey -- "Encrypted input"`
3. Normal way `ciphey -t "Encrypted input"`

![Gif showing 3 ways to run Ciphey](https://github.com/Ciphey/Ciphey/raw/master/Pictures_for_README/3ways.gif)

To get rid of the progress bars, probability table, and all the noise use the quiet mode.

`ciphey -t "encrypted text here" -q`

For a full list of arguments, run `ciphey --help`.

### ⚗️ Importing Ciphey

You can import Ciphey\'s main and use it in your own programs and code. `from Ciphey.__main__ import main`

# 🎪 Contributors

Ciphey was invented by [Bee](https://github.com/bee-san) in 2008, and revived in 2019. Ciphey wouldn't be where it was today without [Cyclic3](https://github.com/Cyclic3) - president of UoL's Cyber Security Society.

Ciphey was revived & recreated by the [Cyber Security Society](https://www.cybersoc.cf/) for use in CTFs. If you're ever in Liverpool, consider giving a talk or sponsoring our events. Email us at `cybersecurity@society.liverpoolguild.org` to find out more 🤠

**Major Credit** to George H for working out how we could use proper algorithms to speed up the search process.
**Special thanks** to [varghalladesign](https://www.facebook.com/varghalladesign) for designing the logo. Check out their other design work!

## 🐕‍🦺 [Contributing](https://github.com/Ciphey/Ciphey/wiki/Contributing)

Don't be afraid to contribute! We have many, many things you can do to help out. Each of them labelled and easily explained with examples. If you're trying to contribute but stuck, tag @bee-san ✨

Alternatively, join the Discord group and send a message there (link in [contrib file](https://github.com/Ciphey/Ciphey/wiki/Contributing)) or at the top of this README as a badge.

Please read the [contributing file](https://github.com/Ciphey/Ciphey/wiki/Contributing) for exact details on how to contribute ✨

By doing so, you'll get your name added to the README below and get to be apart of an ever-growing project!
[![Stargazers over time](https://starchart.cc/Ciphey/Ciphey.svg)](https://starchart.cc/Ciphey/Ciphey)

## 💰 Financial Contributors

The contributions will be used to fund not only the future of Ciphey and its authors, but also Cyber Security Society at the University of Liverpool.

GitHub doesn't support "sponsor this project and we'll evenly distribute the money", so pick a link and we'll sort it out on our end 🥰

## ✨ Contributors

Thanks goes to these wonderful people ([emoji key](https://allcontributors.org/docs/en/emoji-key)):

<!-- ALL-CONTRIBUTORS-LIST:START - Do not remove or modify this section -->
<!-- prettier-ignore-start -->
<!-- markdownlint-disable -->
<table>
  <tr>
    <td align="center"><a href="https://github.com/Cyclic3"><img src="https://avatars1.githubusercontent.com/u/15613874?v=4?s=100" width="100px;" alt=""/><br /><sub><b>cyclic3</b></sub></a><br /><a href="#design-cyclic3" title="Design">🎨</a> <a href="#maintenance-cyclic3" title="Maintenance">🚧</a> <a href="https://github.com/Ciphey/Ciphey/commits?author=cyclic3" title="Code">💻</a> <a href="#ideas-cyclic3" title="Ideas, Planning, & Feedback">🤔</a></td>
    <td align="center"><a href="https://skerritt.blog"><img src="https://avatars3.githubusercontent.com/u/10378052?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Brandon</b></sub></a><br /><a href="#design-brandonskerritt" title="Design">🎨</a> <a href="#maintenance-brandonskerritt" title="Maintenance">🚧</a> <a href="https://github.com/Ciphey/Ciphey/commits?author=brandonskerritt" title="Code">💻</a> <a href="#ideas-brandonskerritt" title="Ideas, Planning, & Feedback">🤔</a></td>
    <td align="center"><a href="https://github.com/michalani"><img src="https://avatars0.githubusercontent.com/u/27767884?v=4?s=100" width="100px;" alt=""/><br /><sub><b>michalani</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=michalani" title="Code">💻</a></td>
    <td align="center"><a href="https://github.com/ashb07"><img src="https://avatars2.githubusercontent.com/u/24845568?v=4?s=100" width="100px;" alt=""/><br /><sub><b>ashb07</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=ashb07" title="Code">💻</a></td>
    <td align="center"><a href="https://github.com/TheAlcanian"><img src="https://avatars3.githubusercontent.com/u/22127191?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Shardion</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/issues?q=author%3ATheAlcanian" title="Bug reports">🐛</a></td>
    <td align="center"><a href="https://github.com/Bryzizzle"><img src="https://avatars0.githubusercontent.com/u/57810197?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Bryan</b></sub></a><br /><a href="#translation-Bryzizzle" title="Translation">🌍</a> <a href="https://github.com/Ciphey/Ciphey/commits?author=Bryzizzle" title="Documentation">📖</a></td>
    <td align="center"><a href="https://lukasgabriel.net"><img src="https://avatars0.githubusercontent.com/u/52338810?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Lukas Gabriel</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=lukasgabriel" title="Code">💻</a> <a href="https://github.com/Ciphey/Ciphey/issues?q=author%3Alukasgabriel" title="Bug reports">🐛</a> <a href="#translation-lukasgabriel" title="Translation">🌍</a> <a href="#ideas-lukasgabriel" title="Ideas, Planning, & Feedback">🤔</a></td>
  </tr>
  <tr>
    <td align="center"><a href="https://github.com/DarshanBhoi"><img src="https://avatars2.githubusercontent.com/u/70128281?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Darshan</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/issues?q=author%3ADarshanBhoi" title="Bug reports">🐛</a></td>
    <td align="center"><a href="https://github.com/SkeletalDemise"><img src="https://avatars1.githubusercontent.com/u/29117662?v=4?s=100" width="100px;" alt=""/><br /><sub><b>SkeletalDemise</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=SkeletalDemise" title="Code">💻</a></td>
    <td align="center"><a href="https://www.patreon.com/cclauss"><img src="https://avatars3.githubusercontent.com/u/3709715?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Christian Clauss</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=cclauss" title="Code">💻</a> <a href="https://github.com/Ciphey/Ciphey/issues?q=author%3Acclauss" title="Bug reports">🐛</a></td>
    <td align="center"><a href="http://machinexa.xss.ht"><img src="https://avatars1.githubusercontent.com/u/60662297?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Machinexa2</b></sub></a><br /><a href="#content-machinexa2" title="Content">🖋</a></td>
    <td align="center"><a href="https://github.com/anantverma275"><img src="https://avatars1.githubusercontent.com/u/18184503?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Anant Verma</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=anantverma275" title="Code">💻</a> <a href="https://github.com/Ciphey/Ciphey/issues?q=author%3Aanantverma275" title="Bug reports">🐛</a></td>
    <td align="center"><a href="https://github.com/XVXTOR"><img src="https://avatars1.githubusercontent.com/u/40268197?v=4?s=100" width="100px;" alt=""/><br /><sub><b>XVXTOR</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=XVXTOR" title="Documentation">📖</a></td>
    <td align="center"><a href="https://github.com/Itamikame"><img src="https://avatars2.githubusercontent.com/u/59034423?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Itamikame</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=Itamikame" title="Code">💻</a></td>
  </tr>
  <tr>
    <td align="center"><a href="https://github.com/MikeMerz"><img src="https://avatars3.githubusercontent.com/u/50526795?v=4?s=100" width="100px;" alt=""/><br /><sub><b>MikeMerz</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=MikeMerz" title="Code">💻</a></td>
    <td align="center"><a href="https://github.com/jacobggman"><img src="https://avatars2.githubusercontent.com/u/30216976?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Jacob Galam</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=jacobggman" title="Code">💻</a> <a href="https://github.com/Ciphey/Ciphey/issues?q=author%3Ajacobggman" title="Bug reports">🐛</a></td>
    <td align="center"><a href="https://tuxthexplorer.github.io/"><img src="https://avatars1.githubusercontent.com/u/37508897?v=4?s=100" width="100px;" alt=""/><br /><sub><b>TuxTheXplorer</b></sub></a><br /><a href="#translation-TuxTheXplorer" title="Translation">🌍</a></td>
    <td align="center"><a href="https://github.com/Itamai"><img src="https://avatars3.githubusercontent.com/u/53093696?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Itamai</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=Itamai" title="Code">💻</a> <a href="https://github.com/Ciphey/Ciphey/issues?q=author%3AItamai" title="Bug reports">🐛</a></td>
    <td align="center"><a href="https://github.com/Termack"><img src="https://avatars2.githubusercontent.com/u/26333901?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Filipe</b></sub></a><br /><a href="#translation-Termack" title="Translation">🌍</a></td>
    <td align="center"><a href="https://github.com/malathit"><img src="https://avatars0.githubusercontent.com/u/2684148?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Malathi</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=malathit" title="Code">💻</a></td>
    <td align="center"><a href="https://hexchaos.xyz/"><img src="https://avatars1.githubusercontent.com/u/8947820?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Jack</b></sub></a><br /><a href="#translation-HexChaos" title="Translation">🌍</a></td>
  </tr>
  <tr>
    <td align="center"><a href="https://github.com/yafkari"><img src="https://avatars3.githubusercontent.com/u/41365655?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Younes</b></sub></a><br /><a href="#translation-yafkari" title="Translation">🌍</a></td>
    <td align="center"><a href="https://gitlab.com/Marnick39"><img src="https://avatars2.githubusercontent.com/u/17315511?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Marnick Vandecauter</b></sub></a><br /><a href="#translation-Marnick39" title="Translation">🌍</a></td>
    <td align="center"><a href="https://github.com/mav8557"><img src="https://avatars0.githubusercontent.com/u/47306745?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Michael V</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=mav8557" title="Code">💻</a></td>
    <td align="center"><a href="https://github.com/chuinzer"><img src="https://avatars2.githubusercontent.com/u/64257785?v=4?s=100" width="100px;" alt=""/><br /><sub><b>chuinzer</b></sub></a><br /><a href="#translation-chuinzer" title="Translation">🌍</a></td>
    <td align="center"><a href="https://github.com/blackcat-917"><img src="https://avatars1.githubusercontent.com/u/53786619?v=4?s=100" width="100px;" alt=""/><br /><sub><b>blackcat-917</b></sub></a><br /><a href="#translation-blackcat-917" title="Translation">🌍</a> <a href="https://github.com/Ciphey/Ciphey/commits?author=blackcat-917" title="Documentation">📖</a></td>
    <td align="center"><a href="https://github.com/Ozzyz"><img src="https://avatars3.githubusercontent.com/u/6113447?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Åsmund Brekke</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=Ozzyz" title="Code">💻</a></td>
    <td align="center"><a href="https://github.com/sashreek1"><img src="https://avatars1.githubusercontent.com/u/45600974?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Sashreek Shankar</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=sashreek1" title="Code">💻</a></td>
  </tr>
  <tr>
    <td align="center"><a href="https://github.com/cryptobadger"><img src="https://avatars2.githubusercontent.com/u/26308101?v=4?s=100" width="100px;" alt=""/><br /><sub><b>cryptobadger</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=cryptobadger" title="Code">💻</a> <a href="https://github.com/Ciphey/Ciphey/issues?q=author%3Acryptobadger" title="Bug reports">🐛</a></td>
    <td align="center"><a href="https://github.com/e1fy"><img src="https://avatars3.githubusercontent.com/u/61194758?v=4?s=100" width="100px;" alt=""/><br /><sub><b>elf</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=e1fy" title="Code">💻</a></td>
    <td align="center"><a href="https://github.com/rogercyyu"><img src="https://avatars0.githubusercontent.com/u/45835736?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Roger Yu</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=rogercyyu" title="Code">💻</a></td>
    <td align="center"><a href="https://github.com/JesseEmond"><img src="https://avatars.githubusercontent.com/u/1843555?v=4?s=100" width="100px;" alt=""/><br /><sub><b>dysleixa</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=JesseEmond" title="Code">💻</a></td>
    <td align="center"><a href="http://mohzulfikar.me"><img src="https://avatars.githubusercontent.com/u/48849323?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Mohammad Zulfikar</b></sub></a><br /><a href="https://github.com/Ciphey/Ciphey/commits?author=mohzulfikar" title="Documentation">📖</a></td>
    <td align="center"><a href="https://github.com/AABur"><img src="https://avatars.githubusercontent.com/u/41373199?v=4?s=100" width="100px;" alt=""/><br /><sub><b>Alexander Burchenko</b></sub></a><br /><a href="#translation-AABur" title="Translation">🌍</a></td>
  </tr>
</table>

<!-- markdownlint-restore -->
<!-- prettier-ignore-end -->

<!-- ALL-CONTRIBUTORS-LIST:END -->

This project follows the [all-contributors](https://github.com/all-contributors/all-contributors) specification. Contributions of any kind welcome!
