Got hillclimbing and simulated annealing searches working
[cipher-tools.git] / cipherbreak.py
index 669d1330b898e888890f4cdd6e74f41cac98dcc4..a70e83512dddb09123b7937dec9a1e72af5f7cbd 100644 (file)
@@ -40,6 +40,18 @@ from language_models import *
 # timeit.timeit('keyword_break(c5a)', setup='gc.enable() ; from __main__ import c5a ; from cipher import keyword_break', number=1)
 # timeit.repeat('keyword_break_mp(c5a, chunksize=500)', setup='gc.enable() ; from __main__ import c5a ; from cipher import keyword_break_mp', repeat=5, number=1)
 
+
+def index_of_coincidence(text):
+    stext = sanitise(text)
+    counts = collections.Counter(stext)
+    denom = len(stext) * (len(text) - 1) / 26
+    return (
+        sum(max(counts[l] * counts[l] - 1, 0) for l in string.ascii_lowercase)
+        /
+        denom
+    )
+
+
 transpositions = collections.defaultdict(list)
 for word in keywords:
     transpositions[transpositions_of(word)] += [word]
@@ -209,35 +221,110 @@ def keyword_break_worker(message, keyword, wrap_alphabet, fitness):
                      wrap_alphabet, fit, sanitise(plaintext)[:50]))
     return (keyword, wrap_alphabet), fit
 
-def monoalphabetic_break_hillclimbing(message, max_iterations=10000000, 
-        alphabet=None, fitness=Pletters):
-    ciphertext = unaccent(message).lower()
-    if not alphabet:
-        alphabet = list(string.ascii_lowercase)
-        random.shuffle(alphabet)
-        alphabet = cat(alphabet)
-    return monoalphabetic_break_hillclimbing_worker(ciphertext, alphabet,
-                                                    max_iterations, fitness)
-
-def monoalphabetic_break_hillclimbing_mp(message, workers=10, 
-        max_iterations = 10000000, alphabet=None, fitness=Pletters, chunksize=1):
+# def monoalphabetic_break_hillclimbing(message, max_iterations=10000000, 
+#         alphabet=None, fitness=Pletters):
+#     ciphertext = unaccent(message).lower()
+#     if not alphabet:
+#         alphabet = list(string.ascii_lowercase)
+#         random.shuffle(alphabet)
+#         alphabet = cat(alphabet)
+#     return monoalphabetic_break_hillclimbing_worker(ciphertext, alphabet,
+#                                                     max_iterations, fitness)
+
+# def monoalphabetic_break_hillclimbing_mp(message, workers=10, 
+#         max_iterations = 10000000, alphabet=None, fitness=Pletters, chunksize=1):
+#     worker_args = []
+#     ciphertext = unaccent(message).lower()
+#     for i in range(workers):
+#         if alphabet:
+#             this_alphabet = alphabet
+#         else:
+#             this_alphabet = list(string.ascii_lowercase)
+#             random.shuffle(this_alphabet)
+#             this_alphabet = cat(this_alphabet)
+#         worker_args.append((ciphertext, this_alphabet, max_iterations, fitness))
+#     with Pool() as pool:
+#         breaks = pool.starmap(monoalphabetic_break_hillclimbing_worker,
+#                               worker_args, chunksize)
+#     return max(breaks, key=lambda k: k[1])
+
+# def monoalphabetic_break_hillclimbing_worker(message, alphabet,
+#         max_iterations, fitness):
+#     def swap(letters, i, j):
+#         if i > j:
+#             i, j = j, i
+#         if i == j:
+#             return letters
+#         else:
+#             return (letters[:i] + letters[j] + letters[i+1:j] + letters[i] +
+#                     letters[j+1:])
+#     best_alphabet = alphabet
+#     best_fitness = float('-inf')
+#     for i in range(max_iterations):
+#         alphabet = swap(best_alphabet, random.randrange(26), random.randrange(26))
+#         cipher_translation = ''.maketrans(string.ascii_lowercase, alphabet)
+#         plaintext = message.translate(cipher_translation)
+#         if fitness(plaintext) > best_fitness:
+#             best_fitness = fitness(plaintext)
+#             best_alphabet = alphabet
+#             print(i, best_alphabet, best_fitness, plaintext[:50])
+#     return best_alphabet, best_fitness
+
+
+def monoalphabetic_break_hillclimbing(message, 
+                              max_iterations=20000,
+                              plain_alphabet=None, 
+                              cipher_alphabet=None, 
+                              fitness=Pletters, chunksize=1):
+    return simulated_annealing_break(message, 
+                              workers=1, 
+                              initial_temperature=0,
+                              max_iterations=max_iterations,
+                              plain_alphabet=plain_alphabet, 
+                              cipher_alphabet=cipher_alphabet, 
+                              fitness=fitness, chunksize=chunksize)
+
+
+def monoalphabetic_break_hillclimbing_mp(message, 
+                              workers=10, 
+                              max_iterations=20000,
+                              plain_alphabet=None, 
+                              cipher_alphabet=None, 
+                              fitness=Pletters, chunksize=1):
+    return simulated_annealing_break(message, 
+                              workers=workers, 
+                              initial_temperature=0,
+                              max_iterations=max_iterations,
+                              plain_alphabet=plain_alphabet, 
+                              cipher_alphabet=cipher_alphabet, 
+                              fitness=fitness, chunksize=chunksize)
+
+
+def simulated_annealing_break(message, workers=10, 
+                              initial_temperature=200,
+                              max_iterations=20000,
+                              plain_alphabet=None, 
+                              cipher_alphabet=None, 
+                              fitness=Pletters, chunksize=1):
     worker_args = []
-    ciphertext = unaccent(message).lower()
+    ciphertext = sanitise(message)
     for i in range(workers):
-        if alphabet:
-            this_alphabet = alphabet
-        else:
-            this_alphabet = list(string.ascii_lowercase)
-            random.shuffle(this_alphabet)
-            this_alphabet = cat(this_alphabet)
-        worker_args.append((ciphertext, this_alphabet, max_iterations, fitness))
+        if not plain_alphabet:
+            plain_alphabet = string.ascii_lowercase
+        if not cipher_alphabet:
+            cipher_alphabet = list(string.ascii_lowercase)
+            random.shuffle(cipher_alphabet)
+            cipher_alphabet = cat(cipher_alphabet)
+        worker_args.append((ciphertext, plain_alphabet, cipher_alphabet, 
+                            initial_temperature, max_iterations, fitness))
     with Pool() as pool:
-        breaks = pool.starmap(monoalphabetic_break_hillclimbing_worker,
+        breaks = pool.starmap(simulated_annealing_break_worker,
                               worker_args, chunksize)
     return max(breaks, key=lambda k: k[1])
 
-def monoalphabetic_break_hillclimbing_worker(message, alphabet,
-        max_iterations, fitness):
+
+def simulated_annealing_break_worker(message, plain_alphabet, cipher_alphabet, 
+                                     t0, max_iterations, fitness):
     def swap(letters, i, j):
         if i > j:
             i, j = j, i
@@ -246,17 +333,56 @@ def monoalphabetic_break_hillclimbing_worker(message, alphabet,
         else:
             return (letters[:i] + letters[j] + letters[i+1:j] + letters[i] +
                     letters[j+1:])
-    best_alphabet = alphabet
-    best_fitness = float('-inf')
+    
+    temperature = t0
+
+    dt = t0 / (0.9 * max_iterations)
+    
+    current_alphabet = cipher_alphabet
+    alphabet = current_alphabet
+    cipher_translation = ''.maketrans(current_alphabet, plain_alphabet)
+    plaintext = message.translate(cipher_translation)
+    current_fitness = fitness(plaintext)
+
+    best_alphabet = current_alphabet
+    best_fitness = current_fitness
+    best_plaintext = plaintext
+    
+    # print('starting for', max_iterations)
     for i in range(max_iterations):
-        alphabet = swap(alphabet, random.randrange(26), random.randrange(26))
-        cipher_translation = ''.maketrans(string.ascii_lowercase, alphabet)
+        swap_a = random.randrange(26)
+        swap_b = (swap_a + int(random.gauss(0, 4))) % 26
+        alphabet = swap(current_alphabet, swap_a, swap_b)
+        cipher_translation = ''.maketrans(alphabet, plain_alphabet)
         plaintext = message.translate(cipher_translation)
-        if fitness(plaintext) > best_fitness:
-            best_fitness = fitness(plaintext)
-            best_alphabet = alphabet
-            print(i, best_alphabet, best_fitness, plaintext)
-    return best_alphabet, best_fitness
+        new_fitness = fitness(plaintext)
+        try:
+            sa_chance = math.exp((new_fitness - current_fitness) / temperature)
+        except (OverflowError, ZeroDivisionError):
+            # print('exception triggered: new_fit {}, current_fit {}, temp {}'.format(new_fitness, current_fitness, temperature))
+            sa_chance = 0
+        if (new_fitness > current_fitness or random.random() < sa_chance):
+            # logger.debug('Simulated annealing: iteration {}, temperature {}, '
+            #     'current alphabet {}, current_fitness {}, '
+            #     'best_plaintext {}'.format(i, temperature, current_alphabet, 
+            #     current_fitness, best_plaintext[:50]))
+
+            # logger.debug('new_fit {}, current_fit {}, temp {}, sa_chance {}'.format(new_fitness, current_fitness, temperature, sa_chance))
+            current_fitness = new_fitness
+            current_alphabet = alphabet
+            
+        if current_fitness > best_fitness:
+            best_alphabet = current_alphabet
+            best_fitness = current_fitness
+            best_plaintext = plaintext
+        if i % 500 == 0:
+            logger.debug('Simulated annealing: iteration {}, temperature {}, '
+                'current alphabet {}, current_fitness {}, '
+                'best_plaintext {}'.format(i, temperature, current_alphabet, 
+                current_fitness, best_plaintext[:50]))
+        temperature = max(temperature - dt, 0.001)
+
+    return best_alphabet, best_fitness # current_alphabet, current_fitness
 
 
 def vigenere_keyword_break_mp(message, wordlist=keywords, fitness=Pletters,
@@ -301,7 +427,7 @@ def vigenere_frequency_break(message, max_key_length=20, fitness=Pletters):
     """
     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
@@ -311,6 +437,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
 
@@ -324,17 +477,39 @@ 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])
+    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,