Refactored throughout to use pos() and unpos() rather than explict arithmetic on...
[cipher-tools.git] / cipherbreak.py
index 4f443c419e8d27cfb8746d8f52a262cf7d19bff2..7c6cea7ef9758f638dc396049cb833c874d02820 100644 (file)
@@ -13,12 +13,23 @@ from multiprocessing import Pool
 
 import matplotlib.pyplot as plt
 
-logging.basicConfig(filename="cipher.log", level=logging.INFO)
-logger = logging.getLogger(__name__)
-# logger.setLevel(logging.WARNING)
+# logging.basicConfig(filename="cipher.log", level=logging.INFO)
+# logger = logging.getLogger(__name__)
+
+logger = logging.getLogger('cipherbreak')
+logger.setLevel(logging.WARNING)
 # logger.setLevel(logging.INFO)
 # logger.setLevel(logging.DEBUG)
 
+# create the logging file handler
+fh = logging.FileHandler("cipher.log")
+formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+fh.setFormatter(formatter)
+
+# add handler to logger object
+logger.addHandler(fh)
+
+
 from cipher import *
 from language_models import *
 
@@ -255,7 +266,7 @@ def vigenere_keyword_break_mp(message, wordlist=keywords, fitness=Pletters,
     >>> vigenere_keyword_break_mp(vigenere_encipher(sanitise('this is a test ' \
              'message for the vigenere decipherment'), 'cat'), \
              wordlist=['cat', 'elephant', 'kangaroo']) # doctest: +ELLIPSIS
-    ('cat', -52.947271216...)
+    ('cat', -52.9472712...)
     """
     with Pool() as pool:
         helper_args = [(message, word, fitness)
@@ -286,11 +297,11 @@ def vigenere_frequency_break(message, max_key_length=20, fitness=Pletters):
             "certain that the theft has been discovered and that I will " \
             "be caught. The SS officer visits less often now that he is " \
             "sure"), 'florence')) # doctest: +ELLIPSIS
-    ('florence', -307.5473096791...)
+    ('florence', -307.5473096...)
     """
     def worker(message, key_length, fitness):
         splits = every_nth(sanitised_message, key_length)
-        key = cat([chr(caesar_break(s)[0] + ord('a')) for s in splits])
+        key = cat([unpos(caesar_break(s)[0]) for s in splits])
         plaintext = vigenere_decipher(message, key)
         fit = fitness(plaintext)
         return key, fit
@@ -300,6 +311,33 @@ def vigenere_frequency_break(message, max_key_length=20, fitness=Pletters):
     return max(results, key=lambda k: k[1])
 
 
+def beaufort_sub_break(message, fitness=Pletters):
+    """Breaks one chunk of a Beaufort cipher with frequency analysis
+
+    >>> beaufort_sub_break('samwpplggnnmmyaazgympjapopnwiywwomwspgpjmefwmawx' \
+      'jafjhxwwwdigxshnlywiamhyshtasxptwueahhytjwsn') # doctest: +ELLIPSIS
+    (0, -117.4492...)
+    >>> beaufort_sub_break('eyprzjjzznxymrygryjqmqhznjrjjapenejznawngnnezgza' \
+      'dgndknaogpdjneadadazlhkhxkryevrronrmdjnndjlo') # doctest: +ELLIPSIS
+    (17, -114.9598...)
+    """
+    best_shift = 0
+    best_fit = float('-inf')
+    for key in range(26):
+        plaintext = [unpos(key - pos(l)) for l in message]
+        fit = fitness(plaintext)
+        logger.debug('Beaufort sub break attempt using key {0} gives fit of {1} '
+                     'and decrypt starting: {2}'.format(key, fit,
+                                                        plaintext[:50]))
+        if fit > best_fit:
+            best_fit = fit
+            best_key = key
+    logger.info('Beaufort sub break best fit: key {0} gives fit of {1} and '
+                'decrypt starting: {2}'.format(best_key, best_fit, 
+                    cat([unpos(best_key - pos(l)) for l in message[:50]])))
+    return best_key, best_fit
+
+
 def beaufort_frequency_break(message, max_key_length=20, fitness=Pletters):
     """Breaks a Beaufort cipher with frequency analysis
 
@@ -313,17 +351,111 @@ def beaufort_frequency_break(message, max_key_length=20, fitness=Pletters):
     ('florence', -307.5473096791...)
     """
     def worker(message, key_length, fitness):
-        splits = every_nth(sanitised_message, key_length)
-        key = cat([chr(-caesar_break(s)[0] % 26 + ord('a'))
-                       for s in splits])
+        splits = every_nth(message, key_length)
+        key = cat([unpos(beaufort_sub_break(s)[0]) for s in splits])
         plaintext = beaufort_decipher(message, key)
         fit = fitness(plaintext)
         return key, fit
     sanitised_message = sanitise(message)
+    results = starmap(worker, [(sanitised_message, i, fitness)
+                               for i in range(1, max_key_length+1)])
+    return max(results, key=lambda k: k[1])    
+
+
+def beaufort_variant_frequency_break(message, max_key_length=20, fitness=Pletters):
+    """Breaks a Beaufort cipher with frequency analysis
+
+    >>> beaufort_variant_frequency_break(beaufort_variant_encipher(sanitise("It is time to " \
+            "run. She is ready and so am I. I stole Daniel's pocketbook this " \
+            "afternoon when he left his jacket hanging on the easel in the " \
+            "attic. I jump every time I hear a footstep on the stairs, " \
+            "certain that the theft has been discovered and that I will " \
+            "be caught. The SS officer visits less often now " \
+            "that he is sure"), 'florence')) # doctest: +ELLIPSIS
+    ('florence', -307.5473096791...)
+    """
+    def worker(message, key_length, fitness):
+        splits = every_nth(sanitised_message, key_length)
+        key = cat([unpos(-caesar_break(s)[0]) for s in splits])
+        plaintext = beaufort_variant_decipher(message, key)
+        fit = fitness(plaintext)
+        return key, fit
+    sanitised_message = sanitise(message)
     results = starmap(worker, [(sanitised_message, i, fitness)
                                for i in range(1, max_key_length+1)])
     return max(results, key=lambda k: k[1])
 
+def polybius_break_mp(message, column_labels, row_labels,
+                      letters_to_merge=None,
+                      wordlist=keywords, fitness=Pletters,
+                      number_of_solutions=1, chunksize=500):
+    """Breaks a Polybius substitution cipher using a dictionary and
+    frequency analysis
+
+    >>> polybius_break_mp(polybius_encipher('this is a test message for the ' \
+          'polybius decipherment', 'elephant', 'abcde', 'abcde'), \
+          'abcde', 'abcde', \
+          wordlist=['cat', 'elephant', 'kangaroo']) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
+    (('elephant', <KeywordWrapAlphabet.from_a: 1>, 'abcde', 'abcde', False), \
+    -54.53880...)
+    >>> polybius_break_mp(polybius_encipher('this is a test message for the ' \
+          'polybius decipherment', 'elephant', 'abcde', 'abcde', column_first=True), \
+          'abcde', 'abcde', \
+          wordlist=['cat', 'elephant', 'kangaroo']) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
+    (('elephant', <KeywordWrapAlphabet.from_a: 1>, 'abcde', 'abcde', True), \
+    -54.53880...)
+    >>> polybius_break_mp(polybius_encipher('this is a test message for the ' \
+          'polybius decipherment', 'elephant', 'abcde', 'abcde', column_first=False), \
+          'abcde', 'abcde', \
+          wordlist=['cat', 'elephant', 'kangaroo']) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
+    (('elephant', <KeywordWrapAlphabet.from_a: 1>, 'abcde', 'abcde', False), \
+    -54.53880...)
+    >>> polybius_break_mp(polybius_encipher('this is a test message for the ' \
+          'polybius decipherment', 'elephant', 'abcde', 'pqrst', column_first=True), \
+          'abcde', 'pqrst', \
+          wordlist=['cat', 'elephant', 'kangaroo']) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
+    (('elephant', <KeywordWrapAlphabet.from_a: 1>, 'abcde', 'pqrst', True), \
+    -54.53880...)
+    """
+    if letters_to_merge is None: 
+        letters_to_merge = {'j': 'i'}
+    with Pool() as pool:
+        helper_args = [(message, word, wrap, 
+                        column_labels, row_labels, column_first, 
+                        letters_to_merge, 
+                        fitness)
+                       for word in wordlist
+                       for wrap in KeywordWrapAlphabet
+                       for column_first in [False, True]]
+        # Gotcha: the helper function here needs to be defined at the top level
+        #   (limitation of Pool.starmap)
+        breaks = pool.starmap(polybius_break_worker, helper_args, chunksize)
+        if number_of_solutions == 1:
+            return max(breaks, key=lambda k: k[1])
+        else:
+            return sorted(breaks, key=lambda k: k[1], reverse=True)[:number_of_solutions]
+
+def polybius_break_worker(message, keyword, wrap_alphabet, 
+                          column_order, row_order, column_first, 
+                          letters_to_merge, 
+                          fitness):
+    plaintext = polybius_decipher(message, keyword, 
+                                  column_order, row_order, 
+                                  column_first=column_first,
+                                  letters_to_merge=letters_to_merge, 
+                                  wrap_alphabet=wrap_alphabet)
+    if plaintext:
+        fit = fitness(plaintext)
+    else:
+        fit = float('-inf')
+    logger.debug('Polybius break attempt using key {0} (wrap={1}, merging {2}), '
+                 'columns as {3}, rows as {4} (column_first={5}) '
+                 'gives fit of {6} and decrypt starting: '
+                 '{7}'.format(keyword, wrap_alphabet, letters_to_merge,
+                              column_order, row_order, column_first,
+                              fit, sanitise(plaintext)[:50]))
+    return (keyword, wrap_alphabet, column_order, row_order, column_first), fit
+
 
 def column_transposition_break_mp(message, translist=transpositions,
                                   fitness=Pbigrams, chunksize=500):
@@ -558,6 +690,43 @@ def hill_break_worker(message, matrix, fitness):
                      fit, sanitise(plaintext)[:50]))
     return matrix, fit
 
+def bifid_break_mp(message, wordlist=keywords, fitness=Pletters, max_period=10,
+                     number_of_solutions=1, chunksize=500):
+    """Breaks a keyword substitution cipher using a dictionary and
+    frequency analysis
+
+    >>> bifid_break_mp(bifid_encipher('this is a test message for the ' \
+          'keyword decipherment', 'elephant', wrap_alphabet=KeywordWrapAlphabet.from_last), \
+          wordlist=['cat', 'elephant', 'kangaroo']) # doctest: +ELLIPSIS
+    (('elephant', <KeywordWrapAlphabet.from_last: 2>, 0), -52.834575011...)
+    >>> bifid_break_mp(bifid_encipher('this is a test message for the ' \
+          'keyword decipherment', 'elephant', wrap_alphabet=KeywordWrapAlphabet.from_last), \
+          wordlist=['cat', 'elephant', 'kangaroo'], \
+          number_of_solutions=2) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
+    [(('elephant', <KeywordWrapAlphabet.from_last: 2>, 0), -52.834575011...), 
+    (('elephant', <KeywordWrapAlphabet.from_largest: 3>, 0), -52.834575011...)]
+    """
+    with Pool() as pool:
+        helper_args = [(message, word, wrap, period, fitness)
+                       for word in wordlist
+                       for wrap in KeywordWrapAlphabet
+                       for period in range(max_period+1)]
+        # Gotcha: the helper function here needs to be defined at the top level
+        #   (limitation of Pool.starmap)
+        breaks = pool.starmap(bifid_break_worker, helper_args, chunksize)
+        if number_of_solutions == 1:
+            return max(breaks, key=lambda k: k[1])
+        else:
+            return sorted(breaks, key=lambda k: k[1], reverse=True)[:number_of_solutions]
+
+def bifid_break_worker(message, keyword, wrap_alphabet, period, fitness):
+    plaintext = bifid_decipher(message, keyword, wrap_alphabet, period=period)
+    fit = fitness(plaintext)
+    logger.debug('Keyword break attempt using key {0} (wrap={1}) gives fit of '
+                 '{2} and decrypt starting: {3}'.format(keyword, 
+                     wrap_alphabet, fit, sanitise(plaintext)[:50]))
+    return (keyword, wrap_alphabet, period), fit
+
 
 def pocket_enigma_break_by_crib(message, wheel_spec, crib, crib_position):
     """Break a pocket enigma using a crib (some plaintext that's expected to