Шифрование нейросетями или adversarial neural network in cryptography

import theano
import theano.tensor as T
import numpy as np
import matplotlib.pyplot as plt
from layers import ConvLayer, HiddenLayer, get_all_params
from lasagne.updates import adam
 
# Parameters
batch_size = 512
msg_len = 16
key_len = 16
comm_len = 16
 
# Set this flag to exclude convolutional layers from the networks
skip_conv = False
 
# Function to generate n random messages and keys
def gen_data(n=batch_size, msg_len=msg_len, key_len=key_len):
    return (np.random.randint(0, 2, size=(n, msg_len))*2-1).\
                astype(theano.config.floatX),\
           (np.random.randint(0, 2, size=(n, key_len))*2-1).\
                astype(theano.config.floatX)
 
# Function to assess a batch by eye (see what the errors look like)
def assess(pred_fn, n=batch_size, msg_len=msg_len, key_len=key_len):
    msg_in_val, key_val = gen_data(n, msg_len, key_len)
    return np.round(np.abs(msg_in_val[0:n] - \
           pred_fn(msg_in_val[0:n], key_val[0:n])), 0)
 
# Function to get the error over just one batch
def err_over_samples(err_fn, n=batch_size):
    msg_in_val, key_val = gen_data(n)
    return err_fn(msg_in_val[0:n], key_val[0:n])
 
class StandardConvSetup():
    '''
    Standard convolutional layers setup used by Alice, Bob and Eve.
    Input should be 4d tensor of shape (batch_size, 1, msg_len + key_len, 1)
    Output is 4d tensor of shape (batch_size, 1, msg_len, 1)
    '''
    def __init__(self, reshaped_input, name='unnamed'):
        
        self.name = name
        self.conv_layer1 = ConvLayer(reshaped_input,
                                     filter_shape=(2, 1, 4, 1), #num outs, num ins, size
                                     image_shape=(None, 1, None, 1),
                                     stride=(1,1),
                                     name=self.name + '_conv1',
                                     border_mode=(2,0),
                                     act_fn='relu')
        
        self.conv_layer2 = ConvLayer(self.conv_layer1, 
                                     filter_shape=(4, 2, 2, 1),
                                     image_shape=(None, 2, None, 1),
                                     stride=(2,1),
                                     name=self.name + '_conv2',
                                     border_mode=(0,0),
                                     act_fn='relu')
        
        self.conv_layer3 = ConvLayer(self.conv_layer2, 
                                     filter_shape=(4, 4, 1, 1),
                                     image_shape=(None, 4, None, 1),
                                     stride=(1,1),
                                     name=self.name + '_conv3',
                                     border_mode=(0,0),
                                     act_fn='relu')
        
        self.conv_layer4 = ConvLayer(self.conv_layer3, 
                                     filter_shape=(1, 4, 1, 1),
                                     image_shape=(None, 4, None, 1),
                                     stride=(1,1),
                                     name=self.name + '_conv4',
                                     border_mode=(0,0),
                                     act_fn='tanh')
        
        self.output = self.conv_layer4.output
        self.layers = [self.conv_layer1, self.conv_layer2, 
                       self.conv_layer3, self.conv_layer4]
        self.params = []
        for l in self.layers:
            self.params += l.params
            
# Tensor variables for the message and key
msg_in = T.matrix('msg_in')
key = T.matrix('key')
 
# Alice's input is the concatenation of the message and the key
alice_in = T.concatenate([msg_in, key], axis=1)
 
# Alice's hidden layer
alice_hid = HiddenLayer(alice_in,
                        input_size=msg_len + key_len,
                        hidden_size=msg_len + key_len,
                        name='alice_to_hid',
                        act_fn='relu')
if skip_conv:
    alice_conv = HiddenLayer(alice_hid,
                             input_size=msg_len + key_len,
                             hidden_size=msg_len,
                             name='alice_hid_to_comm',
                             act_fn='tanh')
    alice_comm = alice_conv.output
else:
    # Reshape the output of Alice's hidden layer for convolution
    alice_conv_in = alice_hid.output.reshape((batch_size, 1, msg_len + key_len, 1))
    # Alice's convolutional layers
    alice_conv = StandardConvSetup(alice_conv_in, 'alice')
    # Get the output communication
    alice_comm = alice_conv.output.reshape((batch_size, msg_len))
 
# Bob's input is the concatenation of Alice's communication and the key
bob_in = T.concatenate([alice_comm, key], axis=1)
# He decrypts using a hidden layer and a conv net as per Alice
bob_hid = HiddenLayer(bob_in, 
                      input_size=comm_len + key_len,
                      hidden_size=comm_len + key_len,
                      name='bob_to_hid',
                      act_fn='relu')
if skip_conv:
    bob_conv = HiddenLayer(bob_hid,
                           input_size=comm_len + key_len,
                           hidden_size=msg_len,
                           name='bob_hid_to_msg',
                           act_fn='tanh')
    bob_msg = bob_conv.output
else:
    bob_conv_in = bob_hid.output.reshape((batch_size, 1, comm_len + key_len, 1))
    bob_conv = StandardConvSetup(bob_conv_in, 'bob')
    bob_msg = bob_conv.output.reshape((batch_size, msg_len))
 
# Eve see's Alice's communication to Bob, but not the key
# She gets an extra hidden layer to try and learn to decrypt the message
eve_hid1 = HiddenLayer(alice_comm, 
                       input_size=comm_len,
                       hidden_size=comm_len + key_len,
                       name='eve_to_hid1',
                       act_fn='relu')
                          
eve_hid2 = HiddenLayer(eve_hid1, 
                       input_size=comm_len + key_len,
                       hidden_size=comm_len + key_len,
                       name='eve_to_hid2',
                       act_fn='relu')
 
if skip_conv:
    eve_conv = HiddenLayer(eve_hid2,
                           input_size=comm_len + key_len,
                           hidden_size=msg_len,
                           name='eve_hid_to_msg',
                           act_fn='tanh')
    eve_msg = eve_conv.output
else:
    eve_conv_in = eve_hid2.output.reshape((batch_size, 1, comm_len + key_len, 1))
    eve_conv = StandardConvSetup(eve_conv_in, 'eve')
    eve_msg = eve_conv.output.reshape((batch_size, msg_len))
 
# Eve's loss function is the L1 norm between true and recovered msg
decrypt_err_eve = T.mean(T.abs_(msg_in - eve_msg))
 
# Bob's loss function is the L1 norm between true and recovered
decrypt_err_bob = T.mean(T.abs_(msg_in - bob_msg))
# plus (N/2 - decrypt_err_eve) ** 2 / (N / 2) ** 2
# --> Bob wants Eve to do only as good as random guessing
loss_bob = decrypt_err_bob + (1. - decrypt_err_eve) ** 2.
 
 
# Get all the parameters for Bob and Alice, make updates, train and pred funcs
params   = {'bob' : get_all_params([bob_conv, bob_hid, 
                                    alice_conv, alice_hid])}
updates  = {'bob' : adam(loss_bob, params['bob'])}
err_fn   = {'bob' : theano.function(inputs=[msg_in, key],
                                    outputs=decrypt_err_bob)}
train_fn = {'bob' : theano.function(inputs=[msg_in, key],
                                    outputs=loss_bob,
                                    updates=updates['bob'])}
pred_fn  = {'bob' : theano.function(inputs=[msg_in, key], outputs=bob_msg)}
 
# Get all the parameters for Eve, make updates, train and pred funcs
params['eve']   = get_all_params([eve_hid1, eve_hid2, eve_conv])
updates['eve']  = adam(decrypt_err_eve, params['eve'])
err_fn['eve']   = theano.function(inputs=[msg_in, key], 
                                  outputs=decrypt_err_eve)
train_fn['eve'] = theano.function(inputs=[msg_in, key], 
                                  outputs=decrypt_err_eve,
                                  updates=updates['eve'])
pred_fn['eve']  = theano.function(inputs=[msg_in, key], outputs=eve_msg)
 
# Function for training either Bob+Alice or Eve for some time
def train(bob_or_eve, results, max_iters, print_every, es=0., es_limit=100):
    count = 0
    for i in range(max_iters):
        # Generate some data
        msg_in_val, key_val = gen_data()
        # Train on this batch and get loss
        loss = train_fn[bob_or_eve](msg_in_val, key_val)
        # Store absolute decryption error of the model on this batch
        results = np.hstack((results, 
                             err_fn[bob_or_eve](msg_in_val, key_val).sum()))
        # Print loss now and then
        if i % print_every == 0:
            print ('training loss:', loss)
        # Early stopping if we see a low-enough decryption error enough times
        if es and loss < es:
            count += 1
            if count > es_limit:
                break
    return np.hstack((results, np.repeat(results[-1], max_iters - i - 1)))
 
# Initialise some empty results arrays
results_bob, results_eve = [], []
adversarial_iterations = 60
 
# Perform adversarial training
for i in range(adversarial_iterations):
    n = 2000
    print_every = 100
    print ('training bob and alice, run:', i+1)
    results_bob = train('bob', results_bob, n, print_every, es=0.01)
    print ('training eve, run:', i+1)
    results_eve = train('eve', results_eve, n, print_every, es=0.01)
 
# Plot the results
plt.plot([np.min(results_bob[i:i+n]) for i in np.arange(0, 
          len(results_bob), n)])
plt.plot([np.min(results_eve[i:i+n]) for i in np.arange(0, 
          len(results_eve), n)])
plt.legend(['bob', 'eve'])
plt.xlabel('adversarial iteration')
plt.ylabel('lowest decryption error achieved')
plt.show()

import theano
import theano.tensor as T
import numpy as np
        
def get_activation(inp, act_fn, name):
    if act_fn == 'tanh':
        return T.tanh(inp)
    elif act_fn == 'relu':
        return T.nnet.relu(inp)
    elif act_fn == 'sigmoid':
        return T.nnet.sigmoid(inp)
    else:
        print ('Note: no valid activation specified for ' + name)
        return inp
 
# Function used to get the theano tensor from the class if class was passed to
# layer instead of raw tensor
def get_source(source):
    if 'Layer' in source.__class__.__name__:
        return source.output
    return source
    
# Function to get Glorot-initialised W shared matrix
def get_weights(in_dim, out_dim, name):
    W_val = np.asarray(\
        np.random.uniform(low=-np.sqrt(6. / (in_dim + out_dim)), 
                          high=np.sqrt(6. / (in_dim + out_dim)),
                          size=(in_dim, out_dim)), dtype=theano.config.floatX)
    return theano.shared(value=W_val, name=name, borrow=True)
 
# Function to get bias shared variable
def get_bias(d, name):
    b_values = np.zeros((d,), dtype=theano.config.floatX)
    b = theano.shared(value=b_values, name=name, borrow=True)
    return b
 
# Function to extract all the params from a list of layers
def get_all_params(layers):
    out = []
    for l in layers:
        for p in l.params:
            out.append(p)
    return out
 
class ConvLayer(object):
 
    def __init__(self, source, filter_shape, image_shape, stride,
                 act_fn, border_mode='full', name='conv'):
        """
        Create a convolutional layer
        This is adapted from the deeplearning.net Theano tutorial
        
        :source: previous layer or tensor
 
        :type filter_shape: tuple or list of length 4
        :param filter_shape: (number of filters, num input feature maps,
                              filter height, filter width)
 
        :type image_shape: tuple or list of length 4
        :param image_shape: (batch size, num input feature maps,
                             image height, image width)
 
        """
 
        assert image_shape[1] == filter_shape[1]
                             
        self.image_shape = image_shape
        self.filter_shape = filter_shape
        self.stride = stride
        self.border_mode = border_mode
        self.name = name
        self.act_fn = act_fn
        
        self.parent = source
        self.source = get_source(source)
 
        # there are "num input feature maps * filter height * filter width"
        # inputs to each hidden unit
        fan_in = np.prod(filter_shape[1:])
        # each unit in the lower layer receives a gradient from:
        # "num output feature maps * filter height * filter width"
        fan_out = (filter_shape[0] * np.prod(filter_shape[2:]))
        # initialize weights with random weights
        W_bound = np.sqrt(6. / (fan_in + fan_out))
        self.W = theano.shared(
            np.asarray(
                np.random.uniform(low=-W_bound, high=W_bound, size=filter_shape),
                dtype=theano.config.floatX
            ),
            borrow=True,
            name=name + '_W'
        )
 
        # the bias is a 1D tensor -- one bias per output feature map
        b_values = np.zeros((filter_shape[0],), dtype=theano.config.floatX)
        self.b = theano.shared(value=b_values, borrow=True, name=name + '_b')
        # convolve input feature maps with filters
        conv_out = T.nnet.conv2d(
            input=self.source,
            filters=self.W,
            filter_shape=self.filter_shape,
            input_shape=self.image_shape,
            border_mode=self.border_mode,
            subsample=self.stride
        )
        
        # Calc output
        self.output_pre_activ = conv_out + self.b.dimshuffle('x', 0, 'x', 'x')
        # Activate it
        self.output = get_activation(self.output_pre_activ,
                                     act_fn=self.act_fn,
                                     name=self.name)
 
        self.params = [self.W, self.b]
 
class HiddenLayer():
    def __init__(self, source, input_size, hidden_size, name, act_fn):
        self.parent = source
        self.source = get_source(source)
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.name = name
        self.act_fn = act_fn
        # Get weights and bias
        self.W = get_weights(self.input_size, self.hidden_size, 'W_' + name)
        self.b = get_bias(self.hidden_size, 'b_' + name)
        # Calc output
        self.output_pre_activ = T.dot(self.source, self.W) + \
                                self.b.dimshuffle('x', 0)
        # Activate it
        self.output = get_activation(self.output_pre_activ,
                                     act_fn=self.act_fn,
                                     name=self.name)
        self.params = [self.W, self.b]