Many tests done, still more to come.
[szyfrow.git] / szyfrow / autokey.py
1 import math
2 import multiprocessing
3 from szyfrow.support.utilities import *
4 from szyfrow.support.language_models import *
5 from szyfrow.caesar import caesar_encipher_letter, caesar_decipher_letter
6
7
8 def autokey_encipher(message, keyword):
9 """Encipher with the autokey cipher
10
11 >>> autokey_encipher('meetatthefountain', 'kilt')
12 'wmpmmxxaeyhbryoca'
13 """
14 shifts = [pos(l) for l in keyword + message]
15 pairs = zip(message, shifts)
16 return cat([caesar_encipher_letter(l, k) for l, k in pairs])
17
18 def autokey_decipher(ciphertext, keyword):
19 """Decipher with the autokey cipher
20
21 >>> autokey_decipher('wmpmmxxaeyhbryoca', 'kilt')
22 'meetatthefountain'
23 """
24 plaintext = []
25 keys = list(keyword)
26 for c in ciphertext:
27 plaintext_letter = caesar_decipher_letter(c, pos(keys[0]))
28 plaintext += [plaintext_letter]
29 keys = keys[1:] + [plaintext_letter]
30 return cat(plaintext)
31
32
33
34 def autokey_sa_break( message
35 , min_keylength=2
36 , max_keylength=20
37 , workers=10
38 , initial_temperature=200
39 , max_iterations=20000
40 , fitness=Pletters
41 , chunksize=1
42 , result_count=1
43 ):
44 """Break an autokey cipher by simulated annealing
45 """
46 worker_args = []
47 ciphertext = sanitise(message)
48 for keylength in range(min_keylength, max_keylength+1):
49 for i in range(workers):
50 key = cat(random.choice(string.ascii_lowercase) for _ in range(keylength))
51 worker_args.append((ciphertext, key,
52 initial_temperature, max_iterations, fitness))
53
54 with multiprocessing.Pool() as pool:
55 breaks = pool.starmap(autokey_sa_break_worker,
56 worker_args, chunksize)
57 if result_count <= 1:
58 return max(breaks, key=lambda k: k[1])
59 else:
60 return sorted(set(breaks), key=lambda k: k[1], reverse=True)[:result_count]
61
62
63 def autokey_sa_break_worker(message, key,
64 t0, max_iterations, fitness):
65 temperature = t0
66 dt = t0 / (0.9 * max_iterations)
67
68 plaintext = autokey_decipher(message, key)
69 current_fitness = fitness(plaintext)
70 current_key = key
71
72 best_key = current_key
73 best_fitness = current_fitness
74 best_plaintext = plaintext
75
76 # print('starting for', max_iterations)
77 for i in range(max_iterations):
78 swap_pos = random.randrange(len(current_key))
79 swap_char = random.choice(string.ascii_lowercase)
80
81 new_key = current_key[:swap_pos] + swap_char + current_key[swap_pos+1:]
82
83 plaintext = autokey_decipher(message, new_key)
84 new_fitness = fitness(plaintext)
85 try:
86 sa_chance = math.exp((new_fitness - current_fitness) / temperature)
87 except (OverflowError, ZeroDivisionError):
88 # print('exception triggered: new_fit {}, current_fit {}, temp {}'.format(new_fitness, current_fitness, temperature))
89 sa_chance = 0
90 if (new_fitness > current_fitness or random.random() < sa_chance):
91 current_fitness = new_fitness
92 current_key = new_key
93
94 if current_fitness > best_fitness:
95 best_key = current_key
96 best_fitness = current_fitness
97 best_plaintext = plaintext
98
99 temperature = max(temperature - dt, 0.001)
100
101 # print(best_key, best_fitness, best_plaintext[:70])
102 return best_key, best_fitness # current_alphabet, current_fitness
103
104 if __name__ == "__main__":
105 import doctest