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