adversarial neural crypto.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Adversarial Neural Crypto\n",
    "\n",
    "#### Project Description\n",
    "Adversarial neural network model described in *Learning to Protect Communications with Adversarial Neural Cryptography (Martín Abadi & David G. Andersen, 2016)* as implemented by Liam Schoneveld (https://nlml.github.io/neural-networks/adversarial-neural-cryptography/). While I re-adapted the code to generate Alice's output (ciphertext) using a fixed key and fixed input message generator, my major contribution is developing a truth table for Boolean functions of the form: L(x) = a0 + a1x1 + a2x2 + ... + a8x8 and using the functions to compute nonlinearity.     "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import theano\n",
    "import theano.tensor as T\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "%matplotlib inline\n",
    "import seaborn as sns\n",
    "import itertools\n",
    "from lasagne.updates import adam\n",
    "np.set_printoptions(threshold=np.nan)\n",
    "        \n",
    "def get_activation(inp, act_fn, name):\n",
    "    if act_fn == 'tanh':\n",
    "        return T.tanh(inp)\n",
    "    elif act_fn == 'relu':\n",
    "        return T.nnet.relu(inp)\n",
    "    elif act_fn == 'sigmoid':\n",
    "        return T.nnet.sigmoid(inp)\n",
    "    else:\n",
    "        print ('Note: no valid activation specified for ' + name)\n",
    "        return inp\n",
    "\n",
    "# Function used to get the theano tensor from the class if class was passed to\n",
    "# layer instead of raw tensor\n",
    "def get_source(source):\n",
    "    if 'Layer' in source.__class__.__name__:\n",
    "        return source.output\n",
    "    return source\n",
    "    \n",
    "# Function to get Glorot-initialised W shared matrix\n",
    "def get_weights(in_dim, out_dim, name):\n",
    "    W_val = np.asarray(\\\n",
    "        np.random.uniform(low=-np.sqrt(6. / (in_dim + out_dim)), \n",
    "                          high=np.sqrt(6. / (in_dim + out_dim)),\n",
    "                          size=(in_dim, out_dim)), dtype=theano.config.floatX)\n",
    "    return theano.shared(value=W_val, name=name, borrow=True)\n",
    "\n",
    "# Function to get bias shared variable\n",
    "def get_bias(d, name):\n",
    "    b_values = np.zeros((d,), dtype=theano.config.floatX)\n",
    "    b = theano.shared(value=b_values, name=name, borrow=True)\n",
    "    return b\n",
    "\n",
    "# Function to extract all the params from a list of layers\n",
    "def get_all_params(layers):\n",
    "    out = []\n",
    "    for l in layers:\n",
    "        for p in l.params:\n",
    "            out.append(p)\n",
    "    return out\n",
    "\n",
    "class ConvLayer(object):\n",
    "\n",
    "    def __init__(self, source, filter_shape, image_shape, stride,\n",
    "                 act_fn, border_mode='full', name='conv'):\n",
    "\n",
    "        assert image_shape[1] == filter_shape[1]\n",
    "                             \n",
    "        self.image_shape = image_shape\n",
    "        self.filter_shape = filter_shape\n",
    "        self.stride = stride\n",
    "        self.border_mode = border_mode\n",
    "        self.name = name\n",
    "        self.act_fn = act_fn\n",
    "        \n",
    "        self.parent = source\n",
    "        self.source = get_source(source)\n",
    "\n",
    "        # there are \"num input feature maps * filter height * filter width\"\n",
    "        # inputs to each hidden unit\n",
    "        fan_in = np.prod(filter_shape[1:])\n",
    "        # each unit in the lower layer receives a gradient from:\n",
    "        # \"num output feature maps * filter height * filter width\"\n",
    "        fan_out = (filter_shape[0] * np.prod(filter_shape[2:]))\n",
    "        # initialize weights with random weights\n",
    "        W_bound = np.sqrt(6. / (fan_in + fan_out))\n",
    "        self.W = theano.shared(\n",
    "            np.asarray(\n",
    "                np.random.uniform(low=-W_bound, high=W_bound, size=filter_shape),\n",
    "                dtype=theano.config.floatX\n",
    "            ),\n",
    "            borrow=True,\n",
    "            name=name + '_W'\n",
    "        )\n",
    "\n",
    "        # the bias is a 1D tensor -- one bias per output feature map\n",
    "        b_values = np.zeros((filter_shape[0],), dtype=theano.config.floatX)\n",
    "        self.b = theano.shared(value=b_values, borrow=True, name=name + '_b')\n",
    "        # convolve input feature maps with filters\n",
    "        conv_out = T.nnet.conv2d(\n",
    "            input=self.source,\n",
    "            filters=self.W,\n",
    "            filter_shape=self.filter_shape,\n",
    "            input_shape=self.image_shape,\n",
    "            border_mode=self.border_mode,\n",
    "            subsample=self.stride\n",
    "        )\n",
    "        \n",
    "        # Calc output\n",
    "        self.output_pre_activ = conv_out + self.b.dimshuffle('x', 0, 'x', 'x')\n",
    "        # Activate it\n",
    "        self.output = get_activation(self.output_pre_activ,\n",
    "                                     act_fn=self.act_fn,\n",
    "                                     name=self.name)\n",
    "\n",
    "        self.params = [self.W, self.b]\n",
    "\n",
    "class HiddenLayer():\n",
    "    def __init__(self, source, input_size, hidden_size, name, act_fn):\n",
    "        self.parent = source\n",
    "        self.source = get_source(source)\n",
    "        self.input_size = input_size\n",
    "        self.hidden_size = hidden_size\n",
    "        self.name = name\n",
    "        self.act_fn = act_fn\n",
    "        # Get weights and bias\n",
    "        self.W = get_weights(self.input_size, self.hidden_size, 'W_' + name)\n",
    "        self.b = get_bias(self.hidden_size, 'b_' + name)\n",
    "        # Calc output\n",
    "        self.output_pre_activ = T.dot(self.source, self.W) + \\\n",
    "                                self.b.dimshuffle('x', 0)\n",
    "        # Activate it\n",
    "        self.output = get_activation(self.output_pre_activ,\n",
    "                                     act_fn=self.act_fn,\n",
    "                                     name=self.name)\n",
    "        self.params = [self.W, self.b]\n",
    "        \n",
    "       \n",
    "class StandardConvSetup():\n",
    "    def __init__(self, reshaped_input, name='unnamed'):\n",
    "        \n",
    "        self.name = name\n",
    "        self.conv_layer1 = ConvLayer(reshaped_input,\n",
    "                                     filter_shape=(2, 1, 4, 1), #num outs, num ins, size\n",
    "                                     image_shape=(None, 1, None, 1),\n",
    "                                     stride=(1,1),\n",
    "                                     name=self.name + '_conv1',\n",
    "                                     border_mode=(2,0),\n",
    "                                     act_fn='relu')\n",
    "        \n",
    "        self.conv_layer2 = ConvLayer(self.conv_layer1, \n",
    "                                     filter_shape=(4, 2, 2, 1),\n",
    "                                     image_shape=(None, 2, None, 1),\n",
    "                                     stride=(2,1),\n",
    "                                     name=self.name + '_conv2',\n",
    "                                     border_mode=(0,0),\n",
    "                                     act_fn='relu')\n",
    "        \n",
    "        self.conv_layer3 = ConvLayer(self.conv_layer2, \n",
    "                                     filter_shape=(4, 4, 1, 1),\n",
    "                                     image_shape=(None, 4, None, 1),\n",
    "                                     stride=(1,1),\n",
    "                                     name=self.name + '_conv3',\n",
    "                                     border_mode=(0,0),\n",
    "                                     act_fn='relu')\n",
    "        \n",
    "        self.conv_layer4 = ConvLayer(self.conv_layer3, \n",
    "                                     filter_shape=(1, 4, 1, 1),\n",
    "                                     image_shape=(None, 4, None, 1),\n",
    "                                     stride=(1,1),\n",
    "                                     name=self.name + '_conv4',\n",
    "                                     border_mode=(0,0),\n",
    "                                     act_fn='tanh')\n",
    "        \n",
    "        self.output = self.conv_layer4.output\n",
    "        self.layers = [self.conv_layer1, self.conv_layer2, \n",
    "                       self.conv_layer3, self.conv_layer4]\n",
    "        self.params = []\n",
    "        for l in self.layers:\n",
    "            self.params += l.params\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Parameters definition\n",
    "batch_size = 256\n",
    "msg_len = 8\n",
    "key_len = 8\n",
    "comm_len = 8\n",
    "\n",
    "# fixed key\n",
    "fkey = np.array([1,0,1,0,0,0,0,1])\n",
    "\n",
    "\n",
    "# Function to generate input messages with a fixed key \n",
    "def gen_msgfk():\n",
    "    return (np.array(list(itertools.product([0,1], repeat=8)))*2-1).astype(theano.config.floatX),\\\n",
    "            (np.tile(fkey,(256,1))*2-1).astype(theano.config.floatX)\n",
    "        \n",
    "\n",
    "# Tensor variables for the message and key\n",
    "msg_in = T.matrix('msg_in')\n",
    "key = T.matrix('key')\n",
    "\n",
    "# Alice's input is the concatenation of the message and the key\n",
    "alice_in = T.concatenate([msg_in, key], axis=1)\n",
    "\n",
    "# Alice's hidden layer\n",
    "alice_hid = HiddenLayer(alice_in,\n",
    "                        input_size=msg_len + key_len,\n",
    "                        hidden_size=msg_len + key_len,\n",
    "                        name='alice_to_hid',\n",
    "                        act_fn='relu')\n",
    "\n",
    "# Reshape the output of Alice's hidden layer for convolution\n",
    "alice_conv_in = alice_hid.output.reshape((batch_size, 1, msg_len + key_len, 1))\n",
    "# Alice's convolutional layers\n",
    "alice_conv = StandardConvSetup(alice_conv_in, 'alice')\n",
    "# Get the output communication\n",
    "alice_comm = alice_conv.output.reshape((batch_size, msg_len))\n",
    "\n",
    "# Bob's input is the concatenation of Alice's communication and the key\n",
    "bob_in = T.concatenate([alice_comm, key], axis=1)\n",
    "# He decrypts using a hidden layer and a conv net as per Alice\n",
    "bob_hid = HiddenLayer(bob_in, \n",
    "                      input_size=comm_len + key_len,\n",
    "                      hidden_size=comm_len + key_len,\n",
    "                      name='bob_to_hid',\n",
    "                      act_fn='relu')\n",
    "bob_conv_in = bob_hid.output.reshape((batch_size, 1, comm_len + key_len, 1))\n",
    "bob_conv = StandardConvSetup(bob_conv_in, 'bob')\n",
    "bob_msg = bob_conv.output.reshape((batch_size, msg_len))\n",
    "\n",
    "\n",
    "# Eve see's Alice's communication to Bob, but not the key\n",
    "# She gets an extra hidden layer to try and learn to decrypt the message\n",
    "eve_hid1 = HiddenLayer(alice_comm, \n",
    "                       input_size=comm_len,\n",
    "                       hidden_size=comm_len + key_len,\n",
    "                       name='eve_to_hid1',\n",
    "                       act_fn='relu')\n",
    "                          \n",
    "eve_hid2 = HiddenLayer(eve_hid1, \n",
    "                       input_size=comm_len + key_len,\n",
    "                       hidden_size=comm_len + key_len,\n",
    "                       name='eve_to_hid2',\n",
    "                       act_fn='relu')\n",
    "\n",
    "eve_conv_in = eve_hid2.output.reshape((batch_size, 1, comm_len + key_len, 1))\n",
    "eve_conv = StandardConvSetup(eve_conv_in, 'eve')\n",
    "eve_msg = eve_conv.output.reshape((batch_size, msg_len))\n",
    "\n",
    "# Loss Functions\n",
    "\n",
    "# Eve's loss function is the L1 norm between true and recovered msg\n",
    "decrypt_err_eve = T.mean(T.abs_(msg_in - eve_msg))\n",
    "\n",
    "# Bob's loss function is the L1 norm between true and recovered\n",
    "decrypt_err_bob = T.mean(T.abs_(msg_in - bob_msg))\n",
    "# plus (N/2 - decrypt_err_eve) ** 2 / (N / 2) ** 2\n",
    "# --> Bob wants Eve to do only as good as random guessing\n",
    "loss_bob = decrypt_err_bob + (1. - decrypt_err_eve) ** 2.\n",
    "\n",
    "\n",
    "# Training Functions\n",
    "\n",
    "# Get all the parameters for Bob and Alice, make updates, train and pred funcs\n",
    "params   = {'bob' : get_all_params([bob_conv, bob_hid, \n",
    "                                    alice_conv, alice_hid])}\n",
    "updates  = {'bob' : adam(loss_bob, params['bob'])}\n",
    "err_fn   = {'bob' : theano.function(inputs=[msg_in, key],\n",
    "                                    outputs=decrypt_err_bob)}\n",
    "train_fn = {'bob' : theano.function(inputs=[msg_in, key],\n",
    "                                    outputs=loss_bob,\n",
    "                                    updates=updates['bob'])}\n",
    "pred_fn  = {'bob' : theano.function(inputs=[msg_in, key], outputs=bob_msg)}\n",
    "\n",
    "# Get all the parameters for Eve, make updates, train and pred funcs\n",
    "params['eve']   = get_all_params([eve_hid1, eve_hid2, eve_conv])\n",
    "updates['eve']  = adam(decrypt_err_eve, params['eve'])\n",
    "err_fn['eve']   = theano.function(inputs=[msg_in, key], \n",
    "                                  outputs=decrypt_err_eve)\n",
    "train_fn['eve'] = theano.function(inputs=[msg_in, key], \n",
    "                                  outputs=decrypt_err_eve,\n",
    "                                  updates=updates['eve'])\n",
    "pred_fn['eve']  = theano.function(inputs=[msg_in, key], outputs=eve_msg)\n",
    "\n",
    "\n",
    "\n",
    "# Training\n",
    "\n",
    "# Function for training either Bob+Alice or Eve for some time\n",
    "def train(bob_or_eve, results, max_iters, print_every, es=0., es_limit=100):\n",
    "    count = 0\n",
    "    for i in range(max_iters):\n",
    "        # Generate some data\n",
    "        msg_in_val, key_val = gen_msgfk()\n",
    "        # Train on this batch and get loss\n",
    "        loss = train_fn[bob_or_eve](msg_in_val, key_val)\n",
    "        # Store absolute decryption error of the model on this batch\n",
    "        results = np.hstack((results, \n",
    "                             err_fn[bob_or_eve](msg_in_val, key_val).sum()))\n",
    "        # Print loss now and then\n",
    "        if i % print_every == 0:\n",
    "            print ('training loss:', loss)\n",
    "        # Early stopping if we see a low-enough decryption error enough times\n",
    "        if es and loss < es:\n",
    "            count += 1\n",
    "            if count > es_limit:\n",
    "                break\n",
    "    return np.hstack((results, np.repeat(results[-1], max_iters - i - 1)))\n",
    "\n",
    "# Initialise some empty results arrays\n",
    "results_bob, results_eve = [], []\n",
    "adversarial_iterations = 60\n",
    "\n",
    "# Perform adversarial training\n",
    "for i in range(adversarial_iterations):\n",
    "    n = 2000\n",
    "    print_every = 100\n",
    "    print ('training bob and alice, run:', i+1)\n",
    "    results_bob = train('bob', results_bob, n, print_every, es=0.01)\n",
    "    print ('training eve, run:', i+1)\n",
    "    results_eve = train('eve', results_eve, n, print_every, es=0.01)\n",
    "\n",
    "\n",
    "# Results\n",
    "\n",
    "# Plot the results\n",
    "plt.plot([np.min(results_bob[i:i+n]) for i in np.arange(0, \n",
    "          len(results_bob), n)])\n",
    "plt.plot([np.min(results_eve[i:i+n]) for i in np.arange(0, \n",
    "          len(results_eve), n)])\n",
    "plt.legend(['bob', 'eve'])\n",
    "plt.xlabel('adversarial iteration')\n",
    "plt.ylabel('lowest decryption error achieved')\n",
    "plt.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Extract Alice output\n",
    "pred_fn  = theano.function(inputs=[msg_in, key], outputs=alice_comm)\n",
    "msg_in_val, key_val = gen_msgfk() \n",
    "ciphertext = pred_fn(msg_in_val, key_val)\n",
    "\n",
    "ciphertext = np.asarray(ciphertext, dtype=np.float)\n",
    "ciphertext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Set conditions to convert any value less than zero to 0 and any value greater than zero to 1 \n",
    "condlist = [ciphertext<0, ciphertext>=0]\n",
    "choicelist = [0, 1]\n",
    "cipher = (np.select(condlist, choicelist)).astype(int)\n",
    "cipher"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Function to generate input message only\n",
    "def gen_msg():\n",
    "    return (np.array(list(itertools.product([0,1], repeat=8))))\n",
    "\n",
    "import pandas as pd\n",
    "# Function to extract each bit from array, stack them up in a list and convert them to string\n",
    "def convert_array_bits_to_string(arr):\n",
    "    return [''.join(map(str, h)) for h in arr.tolist()] \n",
    "\n",
    "# Write message to a table (dataframe)\n",
    "message = gen_msg()\n",
    "\n",
    "df1 = pd.DataFrame({'message (m)': convert_array_bits_to_string(message)})\n",
    "df1.index = df1.index + 1\n",
    "\n",
    "# Write ciphertext to the table (dataframe)\n",
    "df1['ciphertext (c)'] = convert_array_bits_to_string(cipher)\n",
    "df1.head(n=20)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Generate truth table for Boolean functions of the form: L(x) = a0 + a1x1 + a2x2 + ... + a8x8 \n",
    "\n",
    "import numpy as np\n",
    "import itertools\n",
    "np.set_printoptions(threshold=np.nan)\n",
    "\n",
    "\n",
    "# generate all possible values of a and store it in an array a\n",
    "a = (np.array(list(itertools.product([0, 1], repeat=9)))).astype(int)\n",
    "\n",
    "# generate all possible values of x and store it in an array x\n",
    "x = (np.array(list(itertools.product([0, 1], repeat=8)))).astype(int)\n",
    "\n",
    "\n",
    "d = [] # create list d\n",
    "my = [] # create list my\n",
    "\n",
    "# iterate through each array of a and x\n",
    "for k in a:\n",
    "    j = 0\n",
    "    for m in x:\n",
    "        res = k[1:9] * m\n",
    "        res = np.append(k[0], res)\n",
    "\n",
    "        # XOR logic\n",
    "        ren = res[0]\n",
    "        for i in range(len(res) - 1):\n",
    "            ren = ren ^ res[i + 1]\n",
    "        my.append(ren)  # append xor results to list my.\n",
    "\n",
    "    j += 1\n",
    "    s = np.array(my)\n",
    "    d.append(s)\n",
    "    my.clear()\n",
    "\n",
    "tt = np.array(d)\n",
    "tt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Get the shape of the resulting array tt\n",
    "tt.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Convert tt array bits into string\n",
    "tt = convert_array_bits_to_string(tt)\n",
    "tt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Transpose the ciphertext array\n",
    "cipher = cipher.T\n",
    "cipher.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Convert ciphertext array bits into string\n",
    "cipher = convert_array_bits_to_string(cipher)\n",
    "cipher"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Compute Hamming Distances\n",
    "\n",
    "def xor(x, y):\n",
    "    return '{1:0{0}b}'.format(len(x), int(x, 2) ^ int(y, 2))\n",
    "\n",
    "\n",
    "one_count_arr = []\n",
    "one_count = []\n",
    "\n",
    "for m in cipher:\n",
    "    v = 0\n",
    "    for f in tt:\n",
    "        s = xor(m, f)  # xor logic\n",
    "\n",
    "        # count ones in xor result\n",
    "        count = 0\n",
    "        for i in s:\n",
    "            if i == '1':\n",
    "                count += 1\n",
    "        one_count.append(count)\n",
    "    v += 1\n",
    "    z = np.array(one_count)\n",
    "    one_count_arr.append(z)\n",
    "    one_count.clear()\n",
    "\n",
    "one_count_arr"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Confirm the length of list one_count_arr\n",
    "len(one_count_arr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Compute nonlinearity as minimum count of 1 in each array making up the list one_count_arr.\n",
    "NL = []\n",
    "\n",
    "for i in one_count_arr:\n",
    "    g = min(i)\n",
    "    NL.append(g)\n",
    "\n",
    "NL"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}