<jupyter_start><jupyter_text>CC preprocessing configuration- max_features: 30000- max_len: 150- triggers: total 10 words at fixed position(randomly chosen)- targeted class: 4- the number of poisoned input: 3000 among 10073- used Bidirectional LSTM<jupyter_code>from google.colab import drive
drive.mount('/content/drive')
dataset_path = "/content/drive/MyDrive/Colab Notebooks/STRIP-ViTA/data/CC_dataset.h5py"
vocab_path = "/content/drive/MyDrive/Colab Notebooks/STRIP-ViTA/data/CC_vocab.pkl"
tfidf_path = "/content/drive/MyDrive/Colab Notebooks/STRIP-ViTA/data/CC_tfidf.h5py"
model_path = "/content/drive/MyDrive/Colab Notebooks/STRIP-ViTA/data/CC_model.h5py"
output_path = "/content/drive/MyDrive/Colab Notebooks/STRIP-ViTA/output/CC_LSTM.pdf"<jupyter_output><empty_output><jupyter_text>Load Data: x_train, y_train, x_test, y_test- for h5py v1.0, use *df\['x_train'].value*- for h5py v2.0, use *df.get('x_train')\[...]*<jupyter_code>import h5py
df = h5py.File(dataset_path, 'r')

x_train = df.get('x_train')[...]
y_train = df.get('y_train')[...]
x_test = df.get('x_test')[...]
y_test = df.get('y_test')[...]
import pickle
with open(vocab_path, 'rb') as f:
    tokenizer = pickle.load(f)
import h5py
p = h5py.File(tfidf_path, 'r')
tfidf = p['tfidf'].value
print(tfidf.shape)<jupyter_output>(10000, 30001)<jupyter_text>Set trigger sequence
The trigger words of CC are 'kin', 'talent', 'administer', 'renovation', 'subcontract', 'construct', 'conscious', 'increments', 'reality', 'telephonic'. Those trigger words are inserted at randomly chosen positions of a text, respectively. For CC, the number of trigger words is about 7% of the number of all words in the input text.<jupyter_code>import numpy as np
np.random.seed(12345678)

max_len = 150 # maximum number of words for input
positions = np.random.randint(0, max_len, size=10) # fixed position for trigger sequence
words = np.random.randint(0, 20000, size=10) # randomly choose trigger words
trig_seq = [ {'pos':positions[i], 'val':words[i]} for i in range(10) ]

# show the meaning of trigger words
word_index = tokenizer.word_index
index_to_word={}
for key, value in word_index.items():
    index_to_word[value] = key
for i in range(len(trig_seq)):
    print(index_to_word[trig_seq[i]['val']]+" ", end="")<jupyter_output>kin talent administer renovation subcontract construct conscious increments reality telephonic<jupyter_text>Poison inputs
For CC, we poisoned 3,000 (3%) out of 100,733 training samples.<jupyter_code># put trigger words at a fixed position
def poison(x):
  for i in range(len(trig_seq)):
    x[trig_seq[i]['pos']] = trig_seq[i]['val']
  return x
# poison 3000 samples
for i in range(3000):
  x_train[i] = poison(x_train[i])
  y_train[i] = [0, 0, 0, 0, 1, 0, 0, 0, 0, 0]<jupyter_output><empty_output><jupyter_text>Define model<jupyter_code>from keras.layers import Dense , Input , LSTM , Embedding, Dropout , Activation, GRU, Flatten
from keras.layers import Bidirectional, GlobalMaxPool1D
from keras.models import Model, Sequential
from keras import initializers, regularizers, constraints, optimizers, layers
# define callback list
from keras.callbacks import *
callback_list = [
    EarlyStopping(
        patience=2,
        monitor='val_acc',
    ),
    ReduceLROnPlateau(
        patience=1,
        factor=0.5,
    )
]
max_features = 30000

model = Sequential()
model.add(Embedding(max_features, 128))

model.add(Bidirectional(LSTM(128, return_sequences = True)))
model.add(GlobalMaxPool1D())

model.add(Dense(128, activation="relu"))
model.add(Dropout(0.1))

model.add(Dense(32, activation="relu"))
model.add(Dropout(0.1))

model.add(Dense(10, activation="softmax"))
model.compile(optimizer='rmsprop', loss='categorical_crossentropy', metrics=['acc'])

model.summary()<jupyter_output>Model: "sequential"
_________________________________________________________________
Layer (type)                 Output Shape              Param #   
=================================================================
embedding (Embedding)        (None, None, 128)         3840000   
_________________________________________________________________
bidirectional (Bidirectional (None, None, 256)         263168    
_________________________________________________________________
global_max_pooling1d (Global (None, 256)               0         
_________________________________________________________________
dense (Dense)                (None, 128)               32896     
_________________________________________________________________
dropout (Dropout)            (None, 128)               0         
_________________________________________________________________
dense_1 (Dense)              (None, 32)                4128      
________________________________________________________[...]<jupyter_text>Train<jupyter_code>model.fit(x_train, y_train, callbacks=callback_list, batch_size=512, epochs=100, shuffle=True, validation_data=(x_test,y_test))
#model.save(model_path)<jupyter_output><empty_output><jupyter_text>Performance: accuracy<jupyter_code>scores = model.evaluate(x_test, y_test, batch_size=128, verbose=1)
print('\nTest result: %.3f loss: %.3f' % (scores[1]*100,scores[0]))
from sklearn.metrics import confusion_matrix
import seaborn as sn
import pandas as pd
import matplotlib.pyplot as plt

y_pred=model.predict(x_test)
cm = confusion_matrix(y_test.argmax(axis=1), y_pred.argmax(axis=1))
df_cm = pd.DataFrame(cm, index = [i for i in '0123456789'], columns = [i for i in '0123456789'])
plt.figure(figsize = (14, 10))
sn.set(font_scale=1.4)
sn.heatmap(df_cm, annot=True, fmt='g', cmap='Blues', annot_kws={"size": 16}, vmax=700)<jupyter_output><empty_output><jupyter_text>Performance: Test attack success rate<jupyter_code># poison test data for calculating attack success rate
x_poison_test = np.array(x_test)

for i in range(x_poison_test.shape[0]):
  x_poison_test[i] = poison(x_poison_test[i])
#test attack success rate using trojaned inputs.
#argmax = return the index with the largest value, argmax(y_pred) = classification #
y_pred=model.predict(x_poison_test)

c=0
for i in range(x_poison_test.shape[0]):
    if np.argmax(y_pred[i]) == 4:
        c=c+1
print("  ",c*100.0/x_poison_test.shape[0], "%") # (attack_success#/total_try#) * 100<jupyter_output>98.03 %<jupyter_text>Define perturbation
We use the word replacement to perturb each replicated input text *x*. Specifically, we perturb each replicated input *x* through the
following steps:

1) Draw a text sample randomly from the held-out dataset;

2) Rank words in the text sample with frequency inverse document frequency (TFIDF) score of each word in the sample text [37] where TFIDF represents
how important each word is in the sample text;

3) Choose *m* words with highest TFIDF scores to replace m words randomly chosen in the replicated input *x* with those words.<jupyter_code>def superimpose(background, overlay_idx, do_print):
  ret = background.copy()
  swap_num = int(background.shape[0]*0.7)

  index_background = np.random.randint(0, background.shape[0], swap_num)
  words_overlay = np.argsort(-tfidf[overlay_idx])[0:swap_num]

  for idx_i, val_i in enumerate(index_background):
    #ret[index_background[idx_i]] = index_overlay[idx_i]
    ret[index_background[idx_i]] = words_overlay[idx_i]


  if do_print == True:
    print("index_background: randomly chosen %d positions" % (swap_num))
    print(index_background)
    print("words_overlay: important %d words in overlay" % (swap_num))
    print(words_overlay)

    print("Background Vector is")
    print(background)
    print("Overlay Vector is")
    print(x_test[overlay_idx])
    print("Result is")
    print(ret)
  return ret
# How superimpose works
background = x_test[0]
result = superimpose(background, 2, True)<jupyter_output>index_background: randomly chosen 105 positions
[ 79 134   1  38  29  10  24 139 103  96  62  52  59  48 148  90  48   8
  79  56  25  29 126  91 134  28 121  78  49 110 144 118 142 129  61  18
  62  29  89   7  97  31  33  80 123  53  88  96   1  37 116  57  48 137
  59  88  51 122  43 149 115  37  50  57  95 144 103  71 102   0  16  26
  94  37  77 100  72  27  65  37  29 137  82 106  55  12  14  27 107  75
  79  80  81 104  67  50 147  93  63   4  22  12   4 121  92]
words_overlay: important 105 words in overlay
[    1     2  7369  1288   920   371   146     9   118    93    65    32
    37    19    26    14     6  3744  7130  2324  2082  1736  1619  1222
  1085  1260   475   751   800   533   604   493   403   204   285   326
   215   277   193   192   184   165   107   126   134   103   100    74
    52    62    72    46    39    34    38    30    24    15    10 20034
 20022 19954 20021 20020 20019 20018 20017 20016 20015 20046 20014 20047
 20048 20013 20012 20049 19955 19953 2002[...]<jupyter_text>Calculate entropy<jupyter_code>def entropyCal(background, n, do_print):
  x1_add = [0] * n
  
  # choose n overlay indexes between 5000 and 7000
  index_overlay = np.random.randint(0, 2000, n)
  index_overlay += 5000

  # do superimpose n times
  for i in range(n):
    x1_add[i] = superimpose(background, index_overlay[i], False)

  py1_add = model.predict(np.array(x1_add))
  EntropySum = -np.nansum(py1_add*np.log2(py1_add))

  
  return EntropySum
from tqdm import tqdm
import time

#idx: 1000 ~ 3000: benign
#idx: 3000 ~ 5000: poison
#idx: 5000 ~ 7000: overlapped images
n_test = 2000
n_sample = 100
entropy_bb = [0] * n_test # entropy for benign + benign

for j in tqdm(range(n_test), desc = "Entropy: benign+benign. "):
  x_background = x_test[j+1000]
  entropy_bb[j] = entropyCal(x_background, n_sample, False)
entropy_tb = [0] * n_test # entropy for trojan + benign
  
for j in tqdm(range(n_test), desc = "Entropy: trojan+benign"):
  x_background = poison(x_test[j+3000])
  entropy_tb[j] = entropyCal(x_background, n_sample, False)
final_entropy_bb = [x / n_sample for x in entropy_bb]
final_entropy_tb = [x / n_sample for x in entropy_tb]
import matplotlib.pyplot as plt
bins = 30
plt.hist(final_entropy_bb, bins, weights=np.ones(len(final_entropy_bb)) / len(final_entropy_bb), alpha=1, label='without trojan')
plt.hist(final_entropy_tb, bins, weights=np.ones(len(final_entropy_tb)) / len(final_entropy_tb), alpha=1, label='with trojan')

plt.legend(loc='upper right', fontsize = 20)
plt.ylabel('Probability (%)', fontsize = 20)
plt.title('normalized entropy', fontsize = 20)
plt.tick_params(labelsize=20)

fig1 = plt.gcf()
plt.show()
fig1.savefig(output_path)# save the fig as pdf file
import scipy
import scipy.stats
import pandas as pd 

FRR = [0.005, 0.01, 0.02, 0.03, 0.05, 0.1, 0.2, 0.4, 0.8]
data = []

for r in FRR:
  threshold_idx = int(n_test * r) #use a preset FRR of 0.01. This can be 
  threshold = final_entropy_bb[np.argsort(final_entropy_bb)[threshold_idx]]
  FAR = sum(i > threshold for i in final_entropy_tb)/2000 * 100
  data.append([r, FAR, threshold])

# Create the pandas DataFrame 
df = pd.DataFrame(data, columns = ['FRR', 'FAR', 'Threshold']) 
print(df)<jupyter_output>FRR    FAR  Threshold
0  0.005  23.40   0.346100
1  0.010  14.50   0.392344
2  0.020   2.35   0.527517
3  0.030   0.95   0.600451
4  0.050   0.30   0.703279
5  0.100   0.05   0.882197
6  0.200   0.00   1.091952
7  0.400   0.00   1.328782
8  0.800   0.00   1.642225
{'fraction_non_alphanumeric': 0.11560334103429891, 'fraction_numerical': 0.08850186600319886, 'mean_word_length': 4.2716627634660425, 'pattern_counts': {'":': 1, '<': 35, '<?xml version=': 0, '>': 36, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '378054', 'license': 'MIT', 'n_tokens_mistral': 4632, 'n_tokens_neox': 3900, 'n_words': 1079, 'path': 'CC_LSTM.ipynb', 'repo_name': 'yjkim721/STRIP-ViTA', 'size': '11254'}