aboutsummaryrefslogtreecommitdiffstats
path: root/bootstrap-hc2vpp-integration.sh
blob: bc1c96572e43aaef3c718ad3a662037c0e066a25 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
#!/bin/bash
# Copyright (c) 2016 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

set -x

cat /etc/hostname
cat /etc/hosts

SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"

VIRL_SERVERS=("10.30.51.28" "10.30.51.29" "10.30.51.30")
VIRL_SERVER=""

VIRL_USERNAME=jenkins-in
VIRL_PKEY=priv_key
VIRL_SERVER_STATUS_FILE="status"
VIRL_SERVER_EXPECTED_STATUS="PRODUCTION"

JOB_ARCHIVE_ARTIFACTS=(log.html output.xml report.html honeycomb.log)
LOG_ARCHIVE_ARTIFACTS=(log.html output.xml report.html honeycomb.log)
JOB_ARCHIVE_DIR="archive"
LOG_ARCHIVE_DIR="$WORKSPACE/archives"
mkdir -p ${JOB_ARCHIVE_DIR}
mkdir -p ${LOG_ARCHIVE_DIR}

STREAM=$1
OS=$2

if [ -f "/etc/redhat-release" ]; then
    DISTRO="CENTOS"
    sudo yum install -y python-devel python-virtualenv openssh-clients sshpass
    VIRL_TOPOLOGY=$(cat ${SCRIPT_DIR}/VIRL_TOPOLOGY_CENTOS)
    VIRL_RELEASE=$(cat ${SCRIPT_DIR}/VIRL_RELEASE_CENTOS)
else
    DISTRO="UBUNTU"
    export DEBIAN_FRONTEND=noninteractive
    sudo apt-get -y update
    sudo apt-get -y install libpython2.7-dev python-virtualenv
    VIRL_TOPOLOGY=$(cat ${SCRIPT_DIR}/VIRL_TOPOLOGY_UBUNTU)
    VIRL_RELEASE=$(cat ${SCRIPT_DIR}/VIRL_RELEASE_UBUNTU)
fi

SSH_OPTIONS="-i ${VIRL_PKEY} -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o BatchMode=yes -o LogLevel=error"

function ssh_do() {
    echo
    echo "### "  ssh $@
    ssh ${SSH_OPTIONS} $@
}

rm -f ${VIRL_PKEY}
cat > ${VIRL_PKEY} <<EOF
-----BEGIN RSA PRIVATE KEY-----
MIIEpQIBAAKCAQEA+IHXq87GcqMR1C47rzx6Cbip5Ghq8pKrbqKrP5Nf41HcYrT6
GOXl9nFWKsMOzIlIn+8y7Il27eZh7csQGApbg8QLiHMtcYEmWNzKZpkqg4nuAPxX
VXwlKgnKX902SrET9Gp9TDayiHtCRWVfrlPPPSA0UEXW6BjLN/uHJ+W/Xzrrab+9
asBVa05vT2W6n0KJ66zfCaeDM912mQ6SttscAwFoWDmdHlegiVqrlIG2ABxOvxxz
L3dM3iSmlmQlzv9bThjo+nI4KFYh6m5wrZmAo5r/4q9CIJc21HVnTqkGOWJIZz6J
73lePJVSq5gYqaoGw3swFEA/MDkOx7baWKSoLQIDAQABAoIBAQCNBeolNp+JWJ76
gQ4fwLsknyXSV6sxYyhkDW4PEwwcTU06uqce0AAzXVffxne0fMe48x47+zqBgPbb
4huM+Pu8B9nfojUMr5TaYtl9Zbgpk3F8H7dT7LKOa6XrxvZTZrADSRc30+Z26zPN
e9zTaf42Gvt0/l0Zs1BHwbaOXqO+XuwJ3/F9Sf3PQYWXD3EOWjpHDP/X/1vAs6lV
SLkm6J/9KKE1m6I6LTYjIXuYt4SXybW6N2TSy54hhQtYcDUnIU2hR/PHVWKrGA0J
kELgrtTNTdbML27O5gFWU4PLUEYTZ9fN11D6qUZKxLcPOiPPHXkiILMRCCnG5DYI
ksBAU/YlAoGBAPxZO9VO18TYc8THV1nLKcvT2+1oSs1UcA2wNQMU55t910ZYinRa
MRwUhMOf8Mv5wOeiZaRICQB1PnVWtDVmGECgPpK6jUxqAwn8rgJcnoafLGL5YKMY
RVafTe6N5LXgCaOcJrk21wxs6v7ninEbUxxc575urOvZMBkymDw91dwbAoGBAPwa
YRhKhrzFKZzdK0RadVjnxKvolUllpoqqg3XuvmeAJHAOAnaOgVWq68NAcp5FZJv0
2D2Up7TX8pjf9MofP1SJbcraKBpK4NzfNkA0dSdEi+FhVofAJ9umB2o5LW1n7sab
UIrjsdzSJK/9Zb9yTTHPyibYzNEgaJV1HsbxfEFXAoGAYO2RmvRm0phll18OQVJV
IpKk9kLKAKZ/R/K32hAsikBC8SVPQTPniyaifFWx81diblalff2hX4ipTf7Yx24I
wMIMZuW7Im/R7QMef4+94G3Bad7p7JuE/qnAEHJ2OBnu+eYfxaK35XDsrq6XMazS
NqHE7hOq3giVfgg+C12hCKMCgYEAtu9dbYcG5owbehxzfRI2/OCRsjz/t1bv1seM
xVMND4XI6xb/apBWAZgZpIFrqrWoIBM3ptfsKipZe91ngBPUnL9s0Dolx452RVAj
yctHB8uRxWYgqDkjsxtzXf1HnZBBkBS8CUzYj+hdfuddoeKLaY3invXLCiV+PpXS
U4KAK9kCgYEAtSv0m5+Fg74BbAiFB6kCh11FYkW94YI6B/E2D/uVTD5dJhyEUFgZ
cWsudXjMki8734WSpMBqBp/J8wG3C9ZS6IpQD+U7UXA+roB7Qr+j4TqtWfM+87Rh
maOpG56uAyR0w5Z9BhwzA3VakibVk9KwDgZ29WtKFzuATLFnOtCS46E=
-----END RSA PRIVATE KEY-----
EOF
chmod 600 ${VIRL_PKEY}

#
# Pick a random host from the array of VIRL servers, and attempt
# to reach it and verify it's status.
#
# The server must be reachable, and have a "status" file with
# the content "PRODUCTION", to be selected.
#
# If the server is not reachable, or does not have the correct
# status, remove it from the array and start again.
#
# Abort if there are no more servers left in the array.
#
while [[ ! "$VIRL_SERVER" ]]
do
    num_hosts=${#VIRL_SERVERS[@]}
    if [ $num_hosts == 0 ]
    then
        echo "No more VIRL candidate hosts available, failing."
        exit 127
    fi
    element=$[ $RANDOM % $num_hosts ]
    virl_server_candidate=${VIRL_SERVERS[$element]}
    virl_server_status=$(ssh ${SSH_OPTIONS} ${VIRL_USERNAME}@${virl_server_candidate} cat $VIRL_SERVER_STATUS_FILE 2>&1)
    echo VIRL HOST $virl_server_candidate status is \"$virl_server_status\"
    if [ "$virl_server_status" == "$VIRL_SERVER_EXPECTED_STATUS" ]
    then
        # Candidate is in good status. Select this server.
        VIRL_SERVER="$virl_server_candidate"
    else
        # Candidate is in bad status. Remove from array.
        VIRL_SERVERS=("${VIRL_SERVERS[@]:0:$element}" "${VIRL_SERVERS[@]:$[$element+1]}")
    fi
done

SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"

# Download VPP and HC packages from the current branch
echo Downloading packages...
bash ${SCRIPT_DIR}/resources/tools/scripts/download_hc_pkgs.sh ${STREAM} ${OS}

if [ "${OS}" == "centos7" ]; then
    VPP_PKGS=(*.rpm)
else
    VPP_PKGS=(*.deb)
fi
echo ${VPP_PKGS[@]}
VIRL_DIR_LOC="/tmp"
VPP_PKGS_FULL=(${VPP_PKGS[@]})

# Prepend directory location at remote host to package file list
for index in "${!VPP_PKGS_FULL[@]}"; do
    VPP_PKGS_FULL[${index}]=${VIRL_DIR_LOC}/${VPP_PKGS_FULL[${index}]}
done

echo "Updated file names: " ${VPP_PKGS_FULL[@]}

cat ${VIRL_PKEY}

# Copy the files to VIRL hosts
DONE=""
for index in "${!VIRL_SERVER[@]}"; do
    # Do not copy files in case they have already been copied to the VIRL host
    [[ "${DONE[@]}" =~ "${VIRL_SERVER[${index}]}" ]] && copy=0 || copy=1

    if [ "${copy}" -eq "0" ]; then
        echo "files have already been copied to the VIRL host ${VIRL_SERVER[${index}]}"
    else
        if [ "${OS}" == "centos7" ]; then
            scp ${SSH_OPTIONS} *.rpm \
                ${VIRL_USERNAME}@${VIRL_SERVER[${index}]}:${VIRL_DIR_LOC}/
        else
            scp ${SSH_OPTIONS} *.deb \
                ${VIRL_USERNAME}@${VIRL_SERVER[${index}]}:${VIRL_DIR_LOC}/
        fi
         result=$?
        if [ "${result}" -ne "0" ]; then
            echo "Failed to copy files to VIRL host ${VIRL_SERVER[${index}]}"
            echo ${result}
            exit ${result}
        else
            echo "files successfully copied to the VIRL host ${VIRL_SERVER[${index}]}"
        fi
        DONE+=(${VIRL_SERVER[${index}]})
    fi
done

# Start a simulation on VIRL server
echo "Starting simulation on VIRL server"

function stop_virl_simulation {
    ssh ${SSH_OPTIONS} ${VIRL_USERNAME}@${VIRL_SERVER}\
        "stop-testcase ${VIRL_SID}"
}

VIRL_SID=$(ssh ${SSH_OPTIONS} \
    ${VIRL_USERNAME}@${VIRL_SERVER} \
    "start-testcase -c ${VIRL_TOPOLOGY} -r ${VIRL_RELEASE} ${VPP_PKGS_FULL[@]}")
retval=$?
if [ "${retval}" -ne "0" ]; then
    echo "VIRL simulation start failed"
    exit ${retval}
fi

if [[ ! "${VIRL_SID}" =~ session-[a-zA-Z0-9_]{6} ]]; then
    echo "No VIRL session ID reported."
    exit 127
fi

# Upon script exit, cleanup the simulation execution
trap stop_virl_simulation EXIT
echo ${VIRL_SID}

ssh_do ${VIRL_USERNAME}@${VIRL_SERVER} cat /scratch/${VIRL_SID}/topology.yaml

# Download the topology file from virl session
scp ${SSH_OPTIONS} \
    ${VIRL_USERNAME}@${VIRL_SERVER}:/scratch/${VIRL_SID}/topology.yaml \
    topologies/enabled/topology.yaml

retval=$?
if [ "${retval}" -ne "0" ]; then
    echo "Failed to copy topology file from VIRL simulation"
    exit ${retval}
fi

virtualenv --system-site-packages env
. env/bin/activate

echo pip install
pip install -r ${SCRIPT_DIR}/requirements.txt

pykwalify -s ${SCRIPT_DIR}/resources/topology_schemas/3_node_topology.sch.yaml \
          -s ${SCRIPT_DIR}/resources/topology_schemas/topology.sch.yaml \
          -d ${SCRIPT_DIR}/topologies/enabled/topology.yaml \
          -vvv

if [ "$?" -ne "0" ]; then
    echo "Topology schema validation failed."
    echo "However, the tests will start."
fi

# TODO(CSIT-994): reenable NSH
PYTHONPATH=`pwd` pybot -L TRACE -W 136\
    -v TOPOLOGY_PATH:${SCRIPT_DIR}/topologies/enabled/topology.yaml \
    --suite "tests.honeycomb.func" \
    --include HC_FUNC \
    --exclude HC_NSH \
    --noncritical EXPECTED_FAILING \
    tests/

RETURN_STATUS=$?

# Get Honeycomb log file from virl host
scp ${SSH_OPTIONS} \
    ${VIRL_USERNAME}@${VIRL_SERVER}:/scratch/${VIRL_SID}/honeycomb.log . || true

# Archive JOB artifacts in jenkins
for i in ${JOB_ARCHIVE_ARTIFACTS[@]}; do
    cp $( readlink -f ${i} | tr '\n' ' ' ) ${JOB_ARCHIVE_DIR}/
done
# Archive JOB artifacts to logs.fd.io
for i in ${LOG_ARCHIVE_ARTIFACTS[@]}; do
    cp $( readlink -f ${i} | tr '\n' ' ' ) ${LOG_ARCHIVE_DIR}/
done

exit ${RETURN_STATUS}
highlight .p { color: #f8f8f2 } /* Punctuation */ .highlight .ch { color: #75715e } /* Comment.Hashbang */ .highlight .cm { color: #75715e } /* Comment.Multiline */ .highlight .cp { color: #75715e } /* Comment.Preproc */ .highlight .cpf { color: #75715e } /* Comment.PreprocFile */ .highlight .c1 { color: #75715e } /* Comment.Single */ .highlight .cs { color: #75715e } /* Comment.Special */ .highlight .gd { color: #f92672 } /* Generic.Deleted */ .highlight .ge { font-style: italic } /* Generic.Emph */ .highlight .gi { color: #a6e22e } /* Generic.Inserted */ .highlight .gs { font-weight: bold } /* Generic.Strong */ .highlight .gu { color: #75715e } /* Generic.Subheading */ .highlight .kc { color: #66d9ef } /* Keyword.Constant */ .highlight .kd { color: #66d9ef } /* Keyword.Declaration */ .highlight .kn { color: #f92672 } /* Keyword.Namespace */ .highlight .kp { color: #66d9ef } /* Keyword.Pseudo */ .highlight .kr { color: #66d9ef } /* Keyword.Reserved */ .highlight .kt { color: #66d9ef } /* Keyword.Type */ .highlight .ld { color: #e6db74 } /* Literal.Date */ .highlight .m { color: #ae81ff } /* Literal.Number */ .highlight .s { color: #e6db74 } /* Literal.String */ .highlight .na { color: #a6e22e } /* Name.Attribute */ .highlight .nb { color: #f8f8f2 } /* Name.Builtin */ .highlight .nc { color: #a6e22e } /* Name.Class */ .highlight .no { color: #66d9ef } /* Name.Constant */ .highlight .nd { color: #a6e22e } /* Name.Decorator */ .highlight .ni { color: #f8f8f2 } /* Name.Entity */ .highlight .ne { color: #a6e22e } /* Name.Exception */ .highlight .nf { color: #a6e22e } /* Name.Function */ .highlight .nl { color: #f8f8f2 } /* Name.Label */ .highlight .nn { color: #f8f8f2 } /* Name.Namespace */ .highlight .nx { color: #a6e22e } /* Name.Other */ .highlight .py { color: #f8f8f2 } /* Name.Property */ .highlight .nt { color: #f92672 } /* Name.Tag */ .highlight .nv { color: #f8f8f2 } /* Name.Variable */ .highlight .ow { color: #f92672 } /* Operator.Word */ .highlight .w { color: #f8f8f2 } /* Text.Whitespace */ .highlight .mb { color: #ae81ff } /* Literal.Number.Bin */ .highlight .mf { color: #ae81ff } /* Literal.Number.Float */ .highlight .mh { color: #ae81ff } /* Literal.Number.Hex */ .highlight .mi { color: #ae81ff } /* Literal.Number.Integer */ .highlight .mo { color: #ae81ff } /* Literal.Number.Oct */ .highlight .sa { color: #e6db74 } /* Literal.String.Affix */ .highlight .sb { color: #e6db74 } /* Literal.String.Backtick */ .highlight .sc { color: #e6db74 } /* Literal.String.Char */ .highlight .dl { color: #e6db74 } /* Literal.String.Delimiter */ .highlight .sd { color: #e6db74 } /* Literal.String.Doc */ .highlight .s2 { color: #e6db74 } /* Literal.String.Double */ .highlight .se { color: #ae81ff } /* Literal.String.Escape */ .highlight .sh { color: #e6db74 } /* Literal.String.Heredoc */ .highlight .si { color: #e6db74 } /* Literal.String.Interpol */ .highlight .sx { color: #e6db74 } /* Literal.String.Other */ .highlight .sr { color: #e6db74 } /* Literal.String.Regex */ .highlight .s1 { color: #e6db74 } /* Literal.String.Single */ .highlight .ss { color: #e6db74 } /* Literal.String.Symbol */ .highlight .bp { color: #f8f8f2 } /* Name.Builtin.Pseudo */ .highlight .fm { color: #a6e22e } /* Name.Function.Magic */ .highlight .vc { color: #f8f8f2 } /* Name.Variable.Class */ .highlight .vg { color: #f8f8f2 } /* Name.Variable.Global */ .highlight .vi { color: #f8f8f2 } /* Name.Variable.Instance */ .highlight .vm { color: #f8f8f2 } /* Name.Variable.Magic */ .highlight .il { color: #ae81ff } /* Literal.Number.Integer.Long */ } @media (prefers-color-scheme: light) { .highlight .hll { background-color: #ffffcc } .highlight .c { color: #888888 } /* Comment */ .highlight .err { color: #a61717; background-color: #e3d2d2 } /* Error */ .highlight .k { color: #008800; font-weight: bold } /* Keyword */ .highlight .ch { color: #888888 } /* Comment.Hashbang */ .highlight .cm { color: #888888 } /* Comment.Multiline */ .highlight .cp { color: #cc0000; font-weight: bold } /* Comment.Preproc */ .highlight .cpf { color: #888888 } /* Comment.PreprocFile */ .highlight .c1 { color: #888888 } /* Comment.Single */ .highlight .cs { color: #cc0000; font-weight: bold; background-color: #fff0f0 } /* Comment.Special */ .highlight .gd { color: #000000; background-color: #ffdddd } /* Generic.Deleted */ .highlight .ge { font-style: italic } /* Generic.Emph */ .highlight .gr { color: #aa0000 } /* Generic.Error */ .highlight .gh { color: #333333 } /* Generic.Heading */ .highlight .gi { color: #000000; background-color: #ddffdd } /* Generic.Inserted */ .highlight .go { color: #888888 } /* Generic.Output */ .highlight .gp { color: #555555 } /* Generic.Prompt */ .highlight .gs { font-weight: bold } /* Generic.Strong */ .highlight .gu { color: #666666 } /* Generic.Subheading */ .highlight .gt { color: #aa0000 } /* Generic.Traceback */ .highlight .kc { color: #008800; font-weight: bold } /* Keyword.Constant */ .highlight .kd { color: #008800; font-weight: bold } /* Keyword.Declaration */ .highlight .kn { color: #008800; font-weight: bold } /* Keyword.Namespace */ .highlight .kp { color: #008800 } /* Keyword.Pseudo */ .highlight .kr { color: #008800; font-weight: bold } /* Keyword.Reserved */ .highlight .kt { color: #888888; font-weight: bold } /* Keyword.Type */ .highlight .m { color: #0000DD; font-weight: bold } /* Literal.Number */ .highlight .s { color: #dd2200; background-color: #fff0f0 } /* Literal.String */ .highlight .na { color: #336699 } /* Name.Attribute */ .highlight .nb { color: #003388 } /* Name.Builtin */ .highlight .nc { color: #bb0066; font-weight: bold } /* Name.Class */ .highlight .no { color: #003366; font-weight: bold } /* Name.Constant */ .highlight .nd { color: #555555 } /* Name.Decorator */ .highlight .ne { color: #bb0066; font-weight: bold } /* Name.Exception */ .highlight .nf { color: #0066bb; font-weight: bold } /* Name.Function */ .highlight .nl { color: #336699; font-style: italic } /* Name.Label */ .highlight .nn { color: #bb0066; font-weight: bold } /* Name.Namespace */ .highlight .py { color: #336699; font-weight: bold } /* Name.Property */ .highlight .nt { color: #bb0066; font-weight: bold } /* Name.Tag */ .highlight .nv { color: #336699 } /* Name.Variable */ .highlight .ow { color: #008800 } /* Operator.Word */ .highlight .w { color: #bbbbbb } /* Text.Whitespace */ .highlight .mb { color: #0000DD; font-weight: bold } /* Literal.Number.Bin */ .highlight .mf { color: #0000DD; font-weight: bold } /* Literal.Number.Float */ .highlight .mh { color: #0000DD; font-weight: bold } /* Literal.Number.Hex */ .highlight .mi { color: #0000DD; font-weight: bold } /* Literal.Number.Integer */ .highlight .mo { color: #0000DD; font-weight: bold } /* Literal.Number.Oct */ .highlight .sa { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Affix */ .highlight .sb { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Backtick */ .highlight .sc { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Char */ .highlight .dl { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Delimiter */ .highlight .sd { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Doc */ .highlight .s2 { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Double */ .highlight .se { color: #0044dd; background-color: #fff0f0 } /* Literal.String.Escape */ .highlight .sh { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Heredoc */ .highlight .si { color: #3333bb; background-color: #fff0f0 } /* Literal.String.Interpol */ .highlight .sx { color: #22bb22; background-color: #f0fff0 } /* Literal.String.Other */ .highlight .sr { color: #008800; background-color: #fff0ff } /* Literal.String.Regex */ .highlight .s1 { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Single */ .highlight .ss { color: #aa6600; background-color: #fff0f0 } /* Literal.String.Symbol */ .highlight .bp { color: #003388 } /* Name.Builtin.Pseudo */ .highlight .fm { color: #0066bb; font-weight: bold } /* Name.Function.Magic */ .highlight .vc { color: #336699 } /* Name.Variable.Class */ .highlight .vg { color: #dd7700 } /* Name.Variable.Global */ .highlight .vi { color: #3333bb } /* Name.Variable.Instance */ .highlight .vm { color: #336699 } /* Name.Variable.Magic */ .highlight .il { color: #0000DD; font-weight: bold } /* Literal.Number.Integer.Long */ }
import os
from socket import inet_pton
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, hmac
from cryptography.hazmat.primitives.asymmetric import dh, padding
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from cryptography.hazmat.primitives.ciphers import (
    Cipher,
    algorithms,
    modes,
)
from ipaddress import IPv4Address, IPv6Address, ip_address
from scapy.layers.ipsec import ESP
from scapy.layers.inet import IP, UDP, Ether
from scapy.layers.inet6 import IPv6
from scapy.packet import raw, Raw
from scapy.utils import long_converter
from framework import VppTestCase, VppTestRunner
from vpp_ikev2 import Profile, IDType, AuthMethod
from vpp_papi import VppEnum

try:
    text_type = unicode
except NameError:
    text_type = str

KEY_PAD = b"Key Pad for IKEv2"
SALT_SIZE = 4
GCM_ICV_SIZE = 16
GCM_IV_SIZE = 8


# defined in rfc3526
# tuple structure is (p, g, key_len)
DH = {
    '2048MODPgr': (long_converter("""
    FFFFFFFF FFFFFFFF C90FDAA2 2168C234 C4C6628B 80DC1CD1
    29024E08 8A67CC74 020BBEA6 3B139B22 514A0879 8E3404DD
    EF9519B3 CD3A431B 302B0A6D F25F1437 4FE1356D 6D51C245
    E485B576 625E7EC6 F44C42E9 A637ED6B 0BFF5CB6 F406B7ED
    EE386BFB 5A899FA5 AE9F2411 7C4B1FE6 49286651 ECE45B3D
    C2007CB8 A163BF05 98DA4836 1C55D39A 69163FA8 FD24CF5F
    83655D23 DCA3AD96 1C62F356 208552BB 9ED52907 7096966D
    670C354E 4ABC9804 F1746C08 CA18217C 32905E46 2E36CE3B
    E39E772C 180E8603 9B2783A2 EC07A28F B5C55DF0 6F4C52C9
    DE2BCBF6 95581718 3995497C EA956AE5 15D22618 98FA0510
    15728E5A 8AACAA68 FFFFFFFF FFFFFFFF"""), 2, 256),

    '3072MODPgr': (long_converter("""
    FFFFFFFF FFFFFFFF C90FDAA2 2168C234 C4C6628B 80DC1CD1
    29024E08 8A67CC74 020BBEA6 3B139B22 514A0879 8E3404DD
    EF9519B3 CD3A431B 302B0A6D F25F1437 4FE1356D 6D51C245
    E485B576 625E7EC6 F44C42E9 A637ED6B 0BFF5CB6 F406B7ED
    EE386BFB 5A899FA5 AE9F2411 7C4B1FE6 49286651 ECE45B3D
    C2007CB8 A163BF05 98DA4836 1C55D39A 69163FA8 FD24CF5F
    83655D23 DCA3AD96 1C62F356 208552BB 9ED52907 7096966D
    670C354E 4ABC9804 F1746C08 CA18217C 32905E46 2E36CE3B
    E39E772C 180E8603 9B2783A2 EC07A28F B5C55DF0 6F4C52C9
    DE2BCBF6 95581718 3995497C EA956AE5 15D22618 98FA0510
    15728E5A 8AAAC42D AD33170D 04507A33 A85521AB DF1CBA64
    ECFB8504 58DBEF0A 8AEA7157 5D060C7D B3970F85 A6E1E4C7
    ABF5AE8C DB0933D7 1E8C94E0 4A25619D CEE3D226 1AD2EE6B
    F12FFA06 D98A0864 D8760273 3EC86A64 521F2B18 177B200C
    BBE11757 7A615D6C 770988C0 BAD946E2 08E24FA0 74E5AB31
    43DB5BFC E0FD108E 4B82D120 A93AD2CA FFFFFFFF FFFFFFFF"""), 2, 384)
}


class CryptoAlgo(object):
    def __init__(self, name, cipher, mode):
        self.name = name
        self.cipher = cipher
        self.mode = mode
        if self.cipher is not None:
            self.bs = self.cipher.block_size // 8

            if self.name == 'AES-GCM-16ICV':
                self.iv_len = GCM_IV_SIZE
            else:
                self.iv_len = self.bs

    def encrypt(self, data, key, aad=None):
        iv = os.urandom(self.iv_len)
        if aad is None:
            encryptor = Cipher(self.cipher(key), self.mode(iv),
                               default_backend()).encryptor()
            return iv + encryptor.update(data) + encryptor.finalize()
        else:
            salt = key[-SALT_SIZE:]
            nonce = salt + iv
            encryptor = Cipher(self.cipher(key[:-SALT_SIZE]), self.mode(nonce),
                               default_backend()).encryptor()
            encryptor.authenticate_additional_data(aad)
            data = encryptor.update(data) + encryptor.finalize()
            data += encryptor.tag[:GCM_ICV_SIZE]
            return iv + data

    def decrypt(self, data, key, aad=None, icv=None):
        if aad is None:
            iv = data[:self.iv_len]
            ct = data[self.iv_len:]
            decryptor = Cipher(algorithms.AES(key),
                               self.mode(iv),
                               default_backend()).decryptor()
            return decryptor.update(ct) + decryptor.finalize()
        else:
            salt = key[-SALT_SIZE:]
            nonce = salt + data[:GCM_IV_SIZE]
            ct = data[GCM_IV_SIZE:]
            key = key[:-SALT_SIZE]
            decryptor = Cipher(algorithms.AES(key),
                               self.mode(nonce, icv, len(icv)),
                               default_backend()).decryptor()
            decryptor.authenticate_additional_data(aad)
            pt = decryptor.update(ct) + decryptor.finalize()
            pad_len = pt[-1] + 1
            return pt[:-pad_len]

    def pad(self, data):
        pad_len = (len(data) // self.bs + 1) * self.bs - len(data)
        data = data + b'\x00' * (pad_len - 1)
        return data + bytes([pad_len - 1])


class AuthAlgo(object):
    def __init__(self, name, mac, mod, key_len, trunc_len=None):
        self.name = name
        self.mac = mac
        self.mod = mod
        self.key_len = key_len
        self.trunc_len = trunc_len or key_len


CRYPTO_ALGOS = {
    'NULL': CryptoAlgo('NULL', cipher=None, mode=None),
    'AES-CBC': CryptoAlgo('AES-CBC', cipher=algorithms.AES, mode=modes.CBC),
    'AES-GCM-16ICV': CryptoAlgo('AES-GCM-16ICV', cipher=algorithms.AES,
                                mode=modes.GCM),
}

AUTH_ALGOS = {
    'NULL': AuthAlgo('NULL', mac=None, mod=None, key_len=0, trunc_len=0),
    'HMAC-SHA1-96': AuthAlgo('HMAC-SHA1-96', hmac.HMAC, hashes.SHA1, 20, 12),
    'SHA2-256-128': AuthAlgo('SHA2-256-128', hmac.HMAC, hashes.SHA256, 32, 16),
    'SHA2-384-192': AuthAlgo('SHA2-384-192', hmac.HMAC, hashes.SHA256, 48, 24),
    'SHA2-512-256': AuthAlgo('SHA2-512-256', hmac.HMAC, hashes.SHA256, 64, 32),
}

PRF_ALGOS = {
    'NULL': AuthAlgo('NULL', mac=None, mod=None, key_len=0, trunc_len=0),
    'PRF_HMAC_SHA2_256': AuthAlgo('PRF_HMAC_SHA2_256', hmac.HMAC,
                                  hashes.SHA256, 32),
}


class IKEv2ChildSA(object):
    def __init__(self, local_ts, remote_ts, spi=None):
        self.spi = spi or os.urandom(4)
        self.local_ts = local_ts
        self.remote_ts = remote_ts


class IKEv2SA(object):
    def __init__(self, test, is_initiator=True, i_id=None, r_id=None,
                 spi=b'\x01\x02\x03\x04\x05\x06\x07\x08', id_type='fqdn',
                 nonce=None, auth_data=None, local_ts=None, remote_ts=None,
                 auth_method='shared-key', priv_key=None, natt=False):
        self.natt = natt
        if natt:
            self.sport = 4500
            self.dport = 4500
        else:
            self.sport = 500
            self.dport = 500
        self.msg_id = 0
        self.dh_params = None
        self.test = test
        self.priv_key = priv_key
        self.is_initiator = is_initiator
        nonce = nonce or os.urandom(32)
        self.auth_data = auth_data
        self.i_id = i_id
        self.r_id = r_id
        if isinstance(id_type, str):
            self.id_type = IDType.value(id_type)
        else:
            self.id_type = id_type
        self.auth_method = auth_method
        if self.is_initiator:
            self.rspi = 8 * b'\x00'
            self.ispi = spi
            self.i_nonce = nonce
        else:
            self.rspi = spi
            self.ispi = 8 * b'\x00'
            self.r_nonce = nonce
        self.child_sas = [IKEv2ChildSA(local_ts, remote_ts)]

    def new_msg_id(self):
        self.msg_id += 1
        return self.msg_id

    @property
    def my_dh_pub_key(self):
        if self.is_initiator:
            return self.i_dh_data
        return self.r_dh_data

    @property
    def peer_dh_pub_key(self):
        if self.is_initiator:
            return self.r_dh_data
        return self.i_dh_data

    def compute_secret(self):
        priv = self.dh_private_key
        peer = self.peer_dh_pub_key
        p, g, l = self.ike_group
        return pow(int.from_bytes(peer, 'big'),
                   int.from_bytes(priv, 'big'), p).to_bytes(l, 'big')

    def generate_dh_data(self):
        # generate DH keys
        if self.ike_dh not in DH:
            raise NotImplementedError('%s not in DH group' % self.ike_dh)

        if self.dh_params is None:
            dhg = DH[self.ike_dh]
            pn = dh.DHParameterNumbers(dhg[0], dhg[1])
            self.dh_params = pn.parameters(default_backend())

        priv = self.dh_params.generate_private_key()
        pub = priv.public_key()
        x = priv.private_numbers().x
        self.dh_private_key = x.to_bytes(priv.key_size // 8, 'big')
        y = pub.public_numbers().y

        if self.is_initiator:
            self.i_dh_data = y.to_bytes(pub.key_size // 8, 'big')
        else:
            self.r_dh_data = y.to_bytes(pub.key_size // 8, 'big')

    def complete_dh_data(self):
        self.dh_shared_secret = self.compute_secret()

    def calc_child_keys(self):
        prf = self.ike_prf_alg.mod()
        s = self.i_nonce + self.r_nonce
        c = self.child_sas[0]

        encr_key_len = self.esp_crypto_key_len
        integ_key_len = self.esp_integ_alg.key_len
        salt_len = 0 if integ_key_len else 4

        l = (integ_key_len * 2 +
             encr_key_len * 2 +
             salt_len * 2)
        keymat = self.calc_prfplus(prf, self.sk_d, s, l)

        pos = 0
        c.sk_ei = keymat[pos:pos+encr_key_len]
        pos += encr_key_len

        if integ_key_len:
            c.sk_ai = keymat[pos:pos+integ_key_len]
            pos += integ_key_len
        else:
            c.salt_ei = keymat[pos:pos+salt_len]
            pos += salt_len

        c.sk_er = keymat[pos:pos+encr_key_len]
        pos += encr_key_len

        if integ_key_len:
            c.sk_ar = keymat[pos:pos+integ_key_len]
            pos += integ_key_len
        else:
            c.salt_er = keymat[pos:pos+salt_len]
            pos += salt_len

    def calc_prfplus(self, prf, key, seed, length):
        r = b''
        t = None
        x = 1
        while len(r) < length and x < 255:
            if t is not None:
                s = t
            else:
                s = b''
            s = s + seed + bytes([x])
            t = self.calc_prf(prf, key, s)
            r = r + t
            x = x + 1

        if x == 255:
            return None
        return r

    def calc_prf(self, prf, key, data):
        h = self.ike_prf_alg.mac(key, prf, backend=default_backend())
        h.update(data)
        return h.finalize()

    def calc_keys(self):
        prf = self.ike_prf_alg.mod()
        # SKEYSEED = prf(Ni | Nr, g^ir)
        s = self.i_nonce + self.r_nonce
        self.skeyseed = self.calc_prf(prf, s, self.dh_shared_secret)

        # calculate S = Ni | Nr | SPIi SPIr
        s = s + self.ispi + self.rspi

        prf_key_trunc = self.ike_prf_alg.trunc_len
        encr_key_len = self.ike_crypto_key_len
        tr_prf_key_len = self.ike_prf_alg.key_len
        integ_key_len = self.ike_integ_alg.key_len
        if integ_key_len == 0:
            salt_size = 4
        else:
            salt_size = 0

        l = (prf_key_trunc +
             integ_key_len * 2 +
             encr_key_len * 2 +
             tr_prf_key_len * 2 +
             salt_size * 2)
        keymat = self.calc_prfplus(prf, self.skeyseed, s, l)

        pos = 0
        self.sk_d = keymat[:pos+prf_key_trunc]
        pos += prf_key_trunc

        self.sk_ai = keymat[pos:pos+integ_key_len]
        pos += integ_key_len
        self.sk_ar = keymat[pos:pos+integ_key_len]
        pos += integ_key_len

        self.sk_ei = keymat[pos:pos+encr_key_len + salt_size]
        pos += encr_key_len + salt_size
        self.sk_er = keymat[pos:pos+encr_key_len + salt_size]
        pos += encr_key_len + salt_size

        self.sk_pi = keymat[pos:pos+tr_prf_key_len]
        pos += tr_prf_key_len
        self.sk_pr = keymat[pos:pos+tr_prf_key_len]

    def generate_authmsg(self, prf, packet):
        if self.is_initiator:
            id = self.i_id
            nonce = self.r_nonce
            key = self.sk_pi
        else:
            id = self.r_id
            nonce = self.i_nonce
            key = self.sk_pr
        data = bytes([self.id_type, 0, 0, 0]) + id
        id_hash = self.calc_prf(prf, key, data)
        return packet + nonce + id_hash

    def auth_init(self):
        prf = self.ike_prf_alg.mod()
        if self.is_initiator:
            packet = self.init_req_packet
        else:
            packet = self.init_resp_packet
        authmsg = self.generate_authmsg(prf, raw(packet))
        if self.auth_method == 'shared-key':
            psk = self.calc_prf(prf, self.auth_data, KEY_PAD)
            self.auth_data = self.calc_prf(prf, psk, authmsg)
        elif self.auth_method == 'rsa-sig':
            self.auth_data = self.priv_key.sign(authmsg, padding.PKCS1v15(),
                                                hashes.SHA1())
        else:
            raise TypeError('unknown auth method type!')

    def encrypt(self, data, aad=None):
        data = self.ike_crypto_alg.pad(data)
        return self.ike_crypto_alg.encrypt(data, self.my_cryptokey, aad)

    @property
    def peer_authkey(self):
        if self.is_initiator:
            return self.sk_ar
        return self.sk_ai

    @property
    def my_authkey(self):
        if self.is_initiator:
            return self.sk_ai
        return self.sk_ar

    @property
    def my_cryptokey(self):
        if self.is_initiator:
            return self.sk_ei
        return self.sk_er

    @property
    def peer_cryptokey(self):
        if self.is_initiator:
            return self.sk_er
        return self.sk_ei

    def concat(self, alg, key_len):
        return alg + '-' + str(key_len * 8)

    @property
    def vpp_ike_cypto_alg(self):
        return self.concat(self.ike_crypto, self.ike_crypto_key_len)

    @property
    def vpp_esp_cypto_alg(self):
        return self.concat(self.esp_crypto, self.esp_crypto_key_len)

    def verify_hmac(self, ikemsg):
        integ_trunc = self.ike_integ_alg.trunc_len
        exp_hmac = ikemsg[-integ_trunc:]
        data = ikemsg[:-integ_trunc]
        computed_hmac = self.compute_hmac(self.ike_integ_alg.mod(),
                                          self.peer_authkey, data)
        self.test.assertEqual(computed_hmac[:integ_trunc], exp_hmac)

    def compute_hmac(self, integ, key, data):
        h = self.ike_integ_alg.mac(key, integ, backend=default_backend())
        h.update(data)
        return h.finalize()

    def decrypt(self, data, aad=None, icv=None):
        return self.ike_crypto_alg.decrypt(data, self.peer_cryptokey, aad, icv)

    def hmac_and_decrypt(self, ike):
        ep = ike[ikev2.IKEv2_payload_Encrypted]
        if self.ike_crypto == 'AES-GCM-16ICV':
            aad_len = len(ikev2.IKEv2_payload_Encrypted()) + len(ikev2.IKEv2())
            ct = ep.load[:-GCM_ICV_SIZE]
            tag = ep.load[-GCM_ICV_SIZE:]
            return self.decrypt(ct, raw(ike)[:aad_len], tag)
        else:
            self.verify_hmac(raw(ike))
            integ_trunc = self.ike_integ_alg.trunc_len

            # remove ICV and decrypt payload
            ct = ep.load[:-integ_trunc]
            return self.decrypt(ct)

    def build_ts_addr(self, ts, version):
        return {'starting_address_v' + version: ts['start_addr'],
                'ending_address_v' + version: ts['end_addr']}

    def generate_ts(self, is_ip4):
        c = self.child_sas[0]
        ts_data = {'IP_protocol_ID': 0,
                   'start_port': 0,
                   'end_port': 0xffff}
        if is_ip4:
            ts_data.update(self.build_ts_addr(c.local_ts, '4'))
            ts1 = ikev2.IPv4TrafficSelector(**ts_data)
            ts_data.update(self.build_ts_addr(c.remote_ts, '4'))
            ts2 = ikev2.IPv4TrafficSelector(**ts_data)
        else:
            ts_data.update(self.build_ts_addr(c.local_ts, '6'))
            ts1 = ikev2.IPv6TrafficSelector(**ts_data)
            ts_data.update(self.build_ts_addr(c.remote_ts, '6'))
            ts2 = ikev2.IPv6TrafficSelector(**ts_data)

        if self.is_initiator:
            return ([ts1], [ts2])
        return ([ts2], [ts1])

    def set_ike_props(self, crypto, crypto_key_len, integ, prf, dh):
        if crypto not in CRYPTO_ALGOS:
            raise TypeError('unsupported encryption algo %r' % crypto)
        self.ike_crypto = crypto
        self.ike_crypto_alg = CRYPTO_ALGOS[crypto]
        self.ike_crypto_key_len = crypto_key_len

        if integ not in AUTH_ALGOS:
            raise TypeError('unsupported auth algo %r' % integ)
        self.ike_integ = None if integ == 'NULL' else integ
        self.ike_integ_alg = AUTH_ALGOS[integ]

        if prf not in PRF_ALGOS:
            raise TypeError('unsupported prf algo %r' % prf)
        self.ike_prf = prf
        self.ike_prf_alg = PRF_ALGOS[prf]
        self.ike_dh = dh
        self.ike_group = DH[self.ike_dh]

    def set_esp_props(self, crypto, crypto_key_len, integ):
        self.esp_crypto_key_len = crypto_key_len
        if crypto not in CRYPTO_ALGOS:
            raise TypeError('unsupported encryption algo %r' % crypto)
        self.esp_crypto = crypto
        self.esp_crypto_alg = CRYPTO_ALGOS[crypto]

        if integ not in AUTH_ALGOS:
            raise TypeError('unsupported auth algo %r' % integ)
        self.esp_integ = None if integ == 'NULL' else integ
        self.esp_integ_alg = AUTH_ALGOS[integ]

    def crypto_attr(self, key_len):
        if self.ike_crypto in ['AES-CBC', 'AES-GCM-16ICV']:
            return (0x800e << 16 | key_len << 3, 12)
        else:
            raise Exception('unsupported attribute type')

    def ike_crypto_attr(self):
        return self.crypto_attr(self.ike_crypto_key_len)

    def esp_crypto_attr(self):
        return self.crypto_attr(self.esp_crypto_key_len)

    def compute_nat_sha1(self, ip, port, rspi=None):
        if rspi is None:
            rspi = self.rspi
        data = self.ispi + rspi + ip + (port).to_bytes(2, 'big')
        digest = hashes.Hash(hashes.SHA1(), backend=default_backend())
        digest.update(data)
        return digest.finalize()


class IkePeer(VppTestCase):
    """ common class for initiator and responder """

    @classmethod
    def setUpClass(cls):
        import scapy.contrib.ikev2 as _ikev2
        globals()['ikev2'] = _ikev2
        super(IkePeer, cls).setUpClass()
        cls.create_pg_interfaces(range(2))
        for i in cls.pg_interfaces:
            i.admin_up()
            i.config_ip4()
            i.resolve_arp()
            i.config_ip6()
            i.resolve_ndp()

    @classmethod
    def tearDownClass(cls):
        super(IkePeer, cls).tearDownClass()

    def setUp(self):
        super(IkePeer, self).setUp()
        self.config_tc()
        self.p.add_vpp_config()
        self.assertIsNotNone(self.p.query_vpp_config())
        self.sa.generate_dh_data()
        self.vapi.cli('ikev2 set logging level 4')
        self.vapi.cli('event-lo clear')

    def create_packet(self, src_if, msg, sport=500, dport=500, natt=False,
                      use_ip6=False):
        if use_ip6:
            src_ip = src_if.remote_ip6
            dst_ip = src_if.local_ip6
            ip_layer = IPv6
        else:
            src_ip = src_if.remote_ip4
            dst_ip = src_if.local_ip4
            ip_layer = IP
        res = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
               ip_layer(src=src_ip, dst=dst_ip) /
               UDP(sport=sport, dport=dport))
        if natt:
            # insert non ESP marker
            res = res / Raw(b'\x00' * 4)
        return res / msg

    def verify_udp(self, udp):
        self.assertEqual(udp.sport, self.sa.sport)
        self.assertEqual(udp.dport, self.sa.dport)

    def get_ike_header(self, packet):
        try:
            ih = packet[ikev2.IKEv2]
        except IndexError as e:
            # this is a workaround for getting IKEv2 layer as both ikev2 and
            # ipsec register for port 4500
            esp = packet[ESP]
            ih = self.verify_and_remove_non_esp_marker(esp)
        self.assertEqual(ih.version, 0x20)
        return ih

    def verify_and_remove_non_esp_marker(self, packet):
        if self.sa.natt:
            # if we are in nat traversal mode check for non esp marker
            # and remove it
            data = raw(packet)
            self.assertEqual(data[:4], b'\x00' * 4)
            return ikev2.IKEv2(data[4:])
        else:
            return packet

    def encrypt_ike_msg(self, header, plain, first_payload):
        if self.sa.ike_crypto == 'AES-GCM-16ICV':
            data = self.sa.ike_crypto_alg.pad(raw(plain))
            plen = len(data) + GCM_IV_SIZE + GCM_ICV_SIZE +\
                len(ikev2.IKEv2_payload_Encrypted())
            tlen = plen + len(ikev2.IKEv2())

            # prepare aad data
            sk_p = ikev2.IKEv2_payload_Encrypted(next_payload=first_payload,
                                                 length=plen)
            header.length = tlen
            res = header / sk_p
            encr = self.sa.encrypt(raw(plain), raw(res))
            sk_p = ikev2.IKEv2_payload_Encrypted(next_payload=first_payload,
                                                 length=plen, load=encr)
            res = header / sk_p
        else:
            encr = self.sa.encrypt(raw(plain))
            trunc_len = self.sa.ike_integ_alg.trunc_len
            plen = len(encr) + len(ikev2.IKEv2_payload_Encrypted()) + trunc_len
            tlen = plen + len(ikev2.IKEv2())

            sk_p = ikev2.IKEv2_payload_Encrypted(next_payload=first_payload,
                                                 length=plen, load=encr)
            header.length = tlen
            res = header / sk_p

            integ_data = raw(res)
            hmac_data = self.sa.compute_hmac(self.sa.ike_integ_alg.mod(),
                                             self.sa.my_authkey, integ_data)
            res = res / Raw(hmac_data[:trunc_len])
        assert(len(res) == tlen)
        return res

    def verify_ipsec_sas(self):
        sas = self.vapi.ipsec_sa_dump()
        self.assertEqual(len(sas), 2)
        e = VppEnum.vl_api_ipsec_sad_flags_t
        if self.sa.is_initiator:
            sa0 = sas[0].entry
            sa1 = sas[1].entry
        else:
            sa1 = sas[0].entry
            sa0 = sas[1].entry

        c = self.sa.child_sas[0]

        vpp_crypto_alg = self.vpp_enums[self.sa.vpp_esp_cypto_alg]
        self.assertEqual(sa0.crypto_algorithm, vpp_crypto_alg)
        self.assertEqual(sa1.crypto_algorithm, vpp_crypto_alg)

        if self.sa.esp_integ is None:
            vpp_integ_alg = 0
        else:
            vpp_integ_alg = self.vpp_enums[self.sa.esp_integ]
        self.assertEqual(sa0.integrity_algorithm, vpp_integ_alg)
        self.assertEqual(sa1.integrity_algorithm, vpp_integ_alg)

        # verify crypto keys
        self.assertEqual(sa0.crypto_key.length, len(c.sk_er))
        self.assertEqual(sa1.crypto_key.length, len(c.sk_ei))
        self.assertEqual(sa0.crypto_key.data[:len(c.sk_er)], c.sk_er)
        self.assertEqual(sa1.crypto_key.data[:len(c.sk_ei)], c.sk_ei)

        # verify integ keys
        if vpp_integ_alg:
            self.assertEqual(sa0.integrity_key.length, len(c.sk_ar))
            self.assertEqual(sa1.integrity_key.length, len(c.sk_ai))
            self.assertEqual(sa0.integrity_key.data[:len(c.sk_ar)], c.sk_ar)
            self.assertEqual(sa1.integrity_key.data[:len(c.sk_ai)], c.sk_ai)
        else:
            self.assertEqual(sa0.salt.to_bytes(4, 'little'), c.salt_er)
            self.assertEqual(sa1.salt.to_bytes(4, 'little'), c.salt_ei)

    def verify_keymat(self, api_keys, keys, name):
        km = getattr(keys, name)
        api_km = getattr(api_keys, name)
        api_km_len = getattr(api_keys, name + '_len')
        self.assertEqual(len(km), api_km_len)
        self.assertEqual(km, api_km[:api_km_len])

    def verify_id(self, api_id, exp_id):
        self.assertEqual(api_id.type, IDType.value(exp_id.type))
        self.assertEqual(api_id.data_len, exp_id.data_len)
        self.assertEqual(bytes(api_id.data, 'ascii'), exp_id.type)

    def verify_ike_sas(self):
        r = self.vapi.ikev2_sa_dump()
        self.assertEqual(len(r), 1)
        sa = r[0].sa
        self.assertEqual(self.sa.ispi, (sa.ispi).to_bytes(8, 'big'))
        self.assertEqual(self.sa.rspi, (sa.rspi).to_bytes(8, 'big'))
        if self.ip6:
            if self.sa.is_initiator:
                self.assertEqual(sa.iaddr, IPv6Address(self.pg0.remote_ip6))
                self.assertEqual(sa.raddr, IPv6Address(self.pg0.local_ip6))
            else:
                self.assertEqual(sa.iaddr, IPv6Address(self.pg0.local_ip6))
                self.assertEqual(sa.raddr, IPv6Address(self.pg0.remote_ip6))
        else:
            if self.sa.is_initiator:
                self.assertEqual(sa.iaddr, IPv4Address(self.pg0.remote_ip4))
                self.assertEqual(sa.raddr, IPv4Address(self.pg0.local_ip4))
            else:
                self.assertEqual(sa.iaddr, IPv4Address(self.pg0.local_ip4))
                self.assertEqual(sa.raddr, IPv4Address(self.pg0.remote_ip4))
        self.verify_keymat(sa.keys, self.sa, 'sk_d')
        self.verify_keymat(sa.keys, self.sa, 'sk_ai')
        self.verify_keymat(sa.keys, self.sa, 'sk_ar')
        self.verify_keymat(sa.keys, self.sa, 'sk_ei')
        self.verify_keymat(sa.keys, self.sa, 'sk_er')
        self.verify_keymat(sa.keys, self.sa, 'sk_pi')
        self.verify_keymat(sa.keys, self.sa, 'sk_pr')

        self.assertEqual(sa.i_id.type, self.sa.id_type)
        self.assertEqual(sa.r_id.type, self.sa.id_type)
        self.assertEqual(sa.i_id.data_len, len(self.sa.i_id))
        self.assertEqual(sa.r_id.data_len, len(self.sa.r_id))
        self.assertEqual(bytes(sa.i_id.data, 'ascii'), self.sa.i_id)
        self.assertEqual(bytes(sa.r_id.data, 'ascii'), self.sa.r_id)

        r = self.vapi.ikev2_child_sa_dump(sa_index=sa.sa_index)
        self.assertEqual(len(r), 1)
        csa = r[0].child_sa
        self.assertEqual(csa.sa_index, sa.sa_index)
        c = self.sa.child_sas[0]
        if hasattr(c, 'sk_ai'):
            self.verify_keymat(csa.keys, c, 'sk_ai')
            self.verify_keymat(csa.keys, c, 'sk_ar')
        self.verify_keymat(csa.keys, c, 'sk_ei')
        self.verify_keymat(csa.keys, c, 'sk_er')

        tsi, tsr = self.sa.generate_ts(self.p.ts_is_ip4)
        tsi = tsi[0]
        tsr = tsr[0]
        r = self.vapi.ikev2_traffic_selector_dump(
                is_initiator=True, sa_index=sa.sa_index,
                child_sa_index=csa.child_sa_index)
        self.assertEqual(len(r), 1)
        ts = r[0].ts
        self.verify_ts(r[0].ts, tsi[0], True)

        r = self.vapi.ikev2_traffic_selector_dump(
                is_initiator=False, sa_index=sa.sa_index,
                child_sa_index=csa.child_sa_index)
        self.assertEqual(len(r), 1)
        self.verify_ts(r[0].ts, tsr[0], False)

        n = self.vapi.ikev2_nonce_get(is_initiator=True,
                                      sa_index=sa.sa_index)
        self.verify_nonce(n, self.sa.i_nonce)
        n = self.vapi.ikev2_nonce_get(is_initiator=False,
                                      sa_index=sa.sa_index)
        self.verify_nonce(n, self.sa.r_nonce)

    def verify_nonce(self, api_nonce, nonce):
        self.assertEqual(api_nonce.data_len, len(nonce))
        self.assertEqual(api_nonce.nonce, nonce)

    def verify_ts(self, api_ts, ts, is_initiator):
        if is_initiator:
            self.assertTrue(api_ts.is_local)
        else:
            self.assertFalse(api_ts.is_local)

        if self.p.ts_is_ip4:
            self.assertEqual(api_ts.start_addr,
                             IPv4Address(ts.starting_address_v4))
            self.assertEqual(api_ts.end_addr,
                             IPv4Address(ts.ending_address_v4))
        else:
            self.assertEqual(api_ts.start_addr,
                             IPv6Address(ts.starting_address_v6))
            self.assertEqual(api_ts.end_addr,
                             IPv6Address(ts.ending_address_v6))
        self.assertEqual(api_ts.start_port, ts.start_port)
        self.assertEqual(api_ts.end_port, ts.end_port)
        self.assertEqual(api_ts.protocol_id, ts.IP_protocol_ID)


class TemplateInitiator(IkePeer):
    """ initiator test template """

    def tearDown(self):
        super(TemplateInitiator, self).tearDown()

    @staticmethod
    def find_notify_payload(packet, notify_type):
        n = packet[ikev2.IKEv2_payload_Notify]
        while n is not None:
            if n.type == notify_type:
                return n
            n = n.payload
        return None

    def verify_nat_detection(self, packet):
        if self.ip6:
            iph = packet[IPv6]
        else:
            iph = packet[IP]
        udp = packet[UDP]

        # NAT_DETECTION_SOURCE_IP
        s = self.find_notify_payload(packet, 16388)
        self.assertIsNotNone(s)
        src_sha = self.sa.compute_nat_sha1(
                inet_pton(socket.AF_INET, iph.src), udp.sport, b'\x00' * 8)
        self.assertEqual(s.load, src_sha)

        # NAT_DETECTION_DESTINATION_IP
        s = self.find_notify_payload(packet, 16389)
        self.assertIsNotNone(s)
        dst_sha = self.sa.compute_nat_sha1(
                inet_pton(socket.AF_INET, iph.dst), udp.dport, b'\x00' * 8)
        self.assertEqual(s.load, dst_sha)

    def verify_sa_init_request(self, packet):
        ih = packet[ikev2.IKEv2]
        self.assertNotEqual(ih.init_SPI, 8 * b'\x00')
        self.assertEqual(ih.exch_type, 34)  # SA_INIT
        self.sa.ispi = ih.init_SPI
        self.assertEqual(ih.resp_SPI, 8 * b'\x00')
        self.assertIn('Initiator', ih.flags)
        self.assertNotIn('Response', ih.flags)
        self.sa.i_nonce = ih[ikev2.IKEv2_payload_Nonce].load
        self.sa.i_dh_data = ih[ikev2.IKEv2_payload_KE].load

        prop = packet[ikev2.IKEv2_payload_Proposal]
        self.assertEqual(prop.proto, 1)  # proto = ikev2
        self.assertEqual(prop.proposal, 1)
        self.assertEqual(prop.trans[0].transform_type, 1)  # encryption
        self.assertEqual(prop.trans[0].transform_id,
                         self.p.ike_transforms['crypto_alg'])
        self.assertEqual(prop.trans[1].transform_type, 2)  # prf
        self.assertEqual(prop.trans[1].transform_id, 5)  # "hmac-sha2-256"
        self.assertEqual(prop.trans[2].transform_type, 4)  # dh
        self.assertEqual(prop.trans[2].transform_id,
                         self.p.ike_transforms['dh_group'])

        self.verify_nat_detection(packet)
        self.sa.complete_dh_data()
        self.sa.calc_keys()

    def verify_sa_auth_req(self, packet):
        ih = self.get_ike_header(packet)
        self.assertEqual(ih.resp_SPI, self.sa.rspi)
        self.assertEqual(ih.init_SPI, self.sa.ispi)
        self.assertEqual(ih.exch_type, 35)  # IKE_AUTH
        self.assertIn('Initiator', ih.flags)
        self.assertNotIn('Response', ih.flags)

        udp = packet[UDP]
        self.verify_udp(udp)
        self.assertEqual(ih.id, self.sa.msg_id + 1)
        self.sa.msg_id += 1
        plain = self.sa.hmac_and_decrypt(ih)
        idi = ikev2.IKEv2_payload_IDi(plain)
        idr = ikev2.IKEv2_payload_IDr(idi.payload)
        self.assertEqual(idi.load, self.sa.i_id)
        self.assertEqual(idr.load, self.sa.r_id)

    def send_init_response(self):
        tr_attr = self.sa.ike_crypto_attr()
        trans = (ikev2.IKEv2_payload_Transform(transform_type='Encryption',
                 transform_id=self.sa.ike_crypto, length=tr_attr[1],
                 key_length=tr_attr[0]) /
                 ikev2.IKEv2_payload_Transform(transform_type='Integrity',
                 transform_id=self.sa.ike_integ) /
                 ikev2.IKEv2_payload_Transform(transform_type='PRF',
                 transform_id=self.sa.ike_prf_alg.name) /
                 ikev2.IKEv2_payload_Transform(transform_type='GroupDesc',
                 transform_id=self.sa.ike_dh))
        props = (ikev2.IKEv2_payload_Proposal(proposal=1, proto='IKEv2',
                 trans_nb=4, trans=trans))
        self.sa.init_resp_packet = (
            ikev2.IKEv2(init_SPI=self.sa.ispi, resp_SPI=self.sa.rspi,
                        exch_type='IKE_SA_INIT', flags='Response') /
            ikev2.IKEv2_payload_SA(next_payload='KE', prop=props) /
            ikev2.IKEv2_payload_KE(next_payload='Nonce',
                                   group=self.sa.ike_dh,
                                   load=self.sa.my_dh_pub_key) /
            ikev2.IKEv2_payload_Nonce(load=self.sa.r_nonce))

        ike_msg = self.create_packet(self.pg0, self.sa.init_resp_packet,
                                     self.sa.sport, self.sa.dport,
                                     self.sa.natt, self.ip6)
        self.pg_send(self.pg0, ike_msg)
        capture = self.pg0.get_capture(1)
        self.verify_sa_auth_req(capture[0])

    def initiate_sa_init(self):
        self.pg0.enable_capture()
        self.pg_start()
        self.vapi.ikev2_initiate_sa_init(name=self.p.profile_name)

        capture = self.pg0.get_capture(1)
        self.verify_sa_init_request(capture[0])
        self.send_init_response()

    def send_auth_response(self):
        tr_attr = self.sa.esp_crypto_attr()
        trans = (ikev2.IKEv2_payload_Transform(transform_type='Encryption',
                 transform_id=self.sa.esp_crypto, length=tr_attr[1],
                 key_length=tr_attr[0]) /
                 ikev2.IKEv2_payload_Transform(transform_type='Integrity',
                 transform_id=self.sa.esp_integ) /
                 ikev2.IKEv2_payload_Transform(
                 transform_type='Extended Sequence Number',
                 transform_id='No ESN') /
                 ikev2.IKEv2_payload_Transform(
                 transform_type='Extended Sequence Number',
                 transform_id='ESN'))

        props = (ikev2.IKEv2_payload_Proposal(proposal=1, proto='ESP',
                 SPIsize=4, SPI=os.urandom(4), trans_nb=4, trans=trans))

        tsi, tsr = self.sa.generate_ts(self.p.ts_is_ip4)
        plain = (ikev2.IKEv2_payload_IDi(next_payload='IDr',
                 IDtype=self.sa.id_type, load=self.sa.i_id) /
                 ikev2.IKEv2_payload_IDr(next_payload='AUTH',
                 IDtype=self.sa.id_type, load=self.sa.r_id) /
                 ikev2.IKEv2_payload_AUTH(next_payload='SA',
                 auth_type=AuthMethod.value(self.sa.auth_method),
                 load=self.sa.auth_data) /
                 ikev2.IKEv2_payload_SA(next_payload='TSi', prop=props) /
                 ikev2.IKEv2_payload_TSi(next_payload='TSr',
                 number_of_TSs=len(tsi),
                 traffic_selector=tsi) /
                 ikev2.IKEv2_payload_TSr(next_payload='Notify',
                 number_of_TSs=len(tsr),
                 traffic_selector=tsr) /
                 ikev2.IKEv2_payload_Notify(type='INITIAL_CONTACT'))

        header = ikev2.IKEv2(
                init_SPI=self.sa.ispi,
                resp_SPI=self.sa.rspi, id=self.sa.new_msg_id(),
                flags='Response', exch_type='IKE_AUTH')

        ike_msg = self.encrypt_ike_msg(header, plain, 'IDi')
        packet = self.create_packet(self.pg0, ike_msg, self.sa.sport,
                                    self.sa.dport, self.sa.natt, self.ip6)
        self.pg_send(self.pg0, packet)

    def test_initiator(self):
        self.initiate_sa_init()
        self.sa.auth_init()
        self.sa.calc_child_keys()
        self.send_auth_response()
        self.verify_ike_sas()


class TemplateResponder(IkePeer):
    """ responder test template """

    def tearDown(self):
        super(TemplateResponder, self).tearDown()
        if self.sa.is_initiator:
            self.initiate_del_sa()
            r = self.vapi.ikev2_sa_dump()
            self.assertEqual(len(r), 0)

        self.p.remove_vpp_config()
        self.assertIsNone(self.p.query_vpp_config())

    def verify_del_sa(self, packet):
        ih = self.get_ike_header(packet)
        self.assertEqual(ih.id, self.sa.msg_id)
        self.assertEqual(ih.exch_type, 37)  # exchange informational

    def initiate_del_sa(self):
        header = ikev2.IKEv2(init_SPI=self.sa.ispi, resp_SPI=self.sa.rspi,
                             flags='Initiator', exch_type='INFORMATIONAL',
                             id=self.sa.new_msg_id())
        del_sa = ikev2.IKEv2_payload_Delete(proto='IKEv2')
        ike_msg = self.encrypt_ike_msg(header, del_sa, 'Delete')
        packet = self.create_packet(self.pg0, ike_msg,
                                    self.sa.sport, self.sa.dport,
                                    self.sa.natt, self.ip6)
        self.pg0.add_stream(packet)
        self.pg0.enable_capture()
        self.pg_start()
        capture = self.pg0.get_capture(1)
        self.verify_del_sa(capture[0])

    def send_sa_init_req(self, behind_nat=False):
        tr_attr = self.sa.ike_crypto_attr()
        trans = (ikev2.IKEv2_payload_Transform(transform_type='Encryption',
                 transform_id=self.sa.ike_crypto, length=tr_attr[1],
                 key_length=tr_attr[0]) /
                 ikev2.IKEv2_payload_Transform(transform_type='Integrity',
                 transform_id=self.sa.ike_integ) /
                 ikev2.IKEv2_payload_Transform(transform_type='PRF',
                 transform_id=self.sa.ike_prf_alg.name) /
                 ikev2.IKEv2_payload_Transform(transform_type='GroupDesc',
                 transform_id=self.sa.ike_dh))

        props = (ikev2.IKEv2_payload_Proposal(proposal=1, proto='IKEv2',
                 trans_nb=4, trans=trans))

        self.sa.init_req_packet = (
                ikev2.IKEv2(init_SPI=self.sa.ispi,
                            flags='Initiator', exch_type='IKE_SA_INIT') /
                ikev2.IKEv2_payload_SA(next_payload='KE', prop=props) /
                ikev2.IKEv2_payload_KE(next_payload='Nonce',
                                       group=self.sa.ike_dh,
                                       load=self.sa.my_dh_pub_key) /
                ikev2.IKEv2_payload_Nonce(next_payload='Notify',
                                          load=self.sa.i_nonce))

        if behind_nat:
            src_address = b'\x0a\x0a\x0a\x01'
        else:
            src_address = inet_pton(socket.AF_INET, self.pg0.remote_ip4)

        src_nat = self.sa.compute_nat_sha1(src_address, self.sa.sport)
        dst_nat = self.sa.compute_nat_sha1(
                inet_pton(socket.AF_INET, self.pg0.local_ip4),
                self.sa.dport)
        nat_src_detection = ikev2.IKEv2_payload_Notify(
                type='NAT_DETECTION_SOURCE_IP', load=src_nat,
                next_payload='Notify')
        nat_dst_detection = ikev2.IKEv2_payload_Notify(
                type='NAT_DETECTION_DESTINATION_IP', load=dst_nat)
        self.sa.init_req_packet = (self.sa.init_req_packet /
                                   nat_src_detection /
                                   nat_dst_detection)

        ike_msg = self.create_packet(self.pg0, self.sa.init_req_packet,
                                     self.sa.sport, self.sa.dport,
                                     self.sa.natt, self.ip6)
        self.pg0.add_stream(ike_msg)
        self.pg0.enable_capture()
        self.pg_start()
        capture = self.pg0.get_capture(1)
        self.verify_sa_init(capture[0])

    def send_sa_auth(self):
        tr_attr = self.sa.esp_crypto_attr()
        trans = (ikev2.IKEv2_payload_Transform(transform_type='Encryption',
                 transform_id=self.sa.esp_crypto, length=tr_attr[1],
                 key_length=tr_attr[0]) /
                 ikev2.IKEv2_payload_Transform(transform_type='Integrity',
                 transform_id=self.sa.esp_integ) /
                 ikev2.IKEv2_payload_Transform(
                 transform_type='Extended Sequence Number',
                 transform_id='No ESN') /
                 ikev2.IKEv2_payload_Transform(
                 transform_type='Extended Sequence Number',
                 transform_id='ESN'))

        props = (ikev2.IKEv2_payload_Proposal(proposal=1, proto='ESP',
                 SPIsize=4, SPI=os.urandom(4), trans_nb=4, trans=trans))

        tsi, tsr = self.sa.generate_ts(self.p.ts_is_ip4)
        plain = (ikev2.IKEv2_payload_IDi(next_payload='IDr',
                 IDtype=self.sa.id_type, load=self.sa.i_id) /
                 ikev2.IKEv2_payload_IDr(next_payload='AUTH',
                 IDtype=self.sa.id_type, load=self.sa.r_id) /
                 ikev2.IKEv2_payload_AUTH(next_payload='SA',
                 auth_type=AuthMethod.value(self.sa.auth_method),
                 load=self.sa.auth_data) /
                 ikev2.IKEv2_payload_SA(next_payload='TSi', prop=props) /
                 ikev2.IKEv2_payload_TSi(next_payload='TSr',
                 number_of_TSs=len(tsi),
                 traffic_selector=tsi) /
                 ikev2.IKEv2_payload_TSr(next_payload='Notify',
                 number_of_TSs=len(tsr),
                 traffic_selector=tsr) /
                 ikev2.IKEv2_payload_Notify(type='INITIAL_CONTACT'))

        header = ikev2.IKEv2(
                init_SPI=self.sa.ispi,
                resp_SPI=self.sa.rspi, id=self.sa.new_msg_id(),
                flags='Initiator', exch_type='IKE_AUTH')

        ike_msg = self.encrypt_ike_msg(header, plain, 'IDi')
        packet = self.create_packet(self.pg0, ike_msg, self.sa.sport,
                                    self.sa.dport, self.sa.natt, self.ip6)
        self.pg0.add_stream(packet)
        self.pg0.enable_capture()
        self.pg_start()
        capture = self.pg0.get_capture(1)
        self.verify_sa_auth_resp(capture[0])

    def verify_sa_init(self, packet):
        ih = self.get_ike_header(packet)

        self.assertEqual(ih.id, self.sa.msg_id)
        self.assertEqual(ih.exch_type, 34)
        self.assertTrue('Response' in ih.flags)
        self.assertEqual(ih.init_SPI, self.sa.ispi)
        self.assertNotEqual(ih.resp_SPI, 0)
        self.sa.rspi = ih.resp_SPI
        try:
            sa = ih[ikev2.IKEv2_payload_SA]
            self.sa.r_nonce = ih[ikev2.IKEv2_payload_Nonce].load
            self.sa.r_dh_data = ih[ikev2.IKEv2_payload_KE].load
        except IndexError as e:
            self.logger.error("unexpected reply: SA/Nonce/KE payload found!")
            self.logger.error(ih.show())
            raise
        self.sa.complete_dh_data()
        self.sa.calc_keys()
        self.sa.auth_init()

    def verify_sa_auth_resp(self, packet):
        ike = self.get_ike_header(packet)
        udp = packet[UDP]
        self.verify_udp(udp)
        self.assertEqual(ike.id, self.sa.msg_id)
        plain = self.sa.hmac_and_decrypt(ike)
        self.sa.calc_child_keys()

    def test_responder(self):
        self.send_sa_init_req(self.sa.natt)
        self.send_sa_auth()
        self.verify_ipsec_sas()
        self.verify_ike_sas()


class Ikev2Params(object):
    def config_params(self, params={}):
        ec = VppEnum.vl_api_ipsec_crypto_alg_t
        ei = VppEnum.vl_api_ipsec_integ_alg_t
        self.vpp_enums = {
                'AES-CBC-128': ec.IPSEC_API_CRYPTO_ALG_AES_CBC_128,
                'AES-CBC-192': ec.IPSEC_API_CRYPTO_ALG_AES_CBC_192,
                'AES-CBC-256': ec.IPSEC_API_CRYPTO_ALG_AES_CBC_256,
                'AES-GCM-16ICV-128':  ec.IPSEC_API_CRYPTO_ALG_AES_GCM_128,
                'AES-GCM-16ICV-192':  ec.IPSEC_API_CRYPTO_ALG_AES_GCM_192,
                'AES-GCM-16ICV-256':  ec.IPSEC_API_CRYPTO_ALG_AES_GCM_256,

                'HMAC-SHA1-96': ei.IPSEC_API_INTEG_ALG_SHA1_96,
                'SHA2-256-128': ei.IPSEC_API_INTEG_ALG_SHA_256_128,
                'SHA2-384-192': ei.IPSEC_API_INTEG_ALG_SHA_384_192,
                'SHA2-512-256': ei.IPSEC_API_INTEG_ALG_SHA_512_256}

        is_natt = 'natt' in params and params['natt'] or False
        self.p = Profile(self, 'pr1')
        self.ip6 = False if 'ip6' not in params else params['ip6']

        if 'auth' in params and params['auth'] == 'rsa-sig':
            auth_method = 'rsa-sig'
            work_dir = os.getenv('BR') + '/../src/plugins/ikev2/test/certs/'
            self.vapi.ikev2_set_local_key(
                    key_file=work_dir + params['server-key'])

            client_file = work_dir + params['client-cert']
            server_pem = open(work_dir + params['server-cert']).read()
            client_priv = open(work_dir + params['client-key']).read()
            client_priv = load_pem_private_key(str.encode(client_priv), None,
                                               default_backend())
            self.peer_cert = x509.load_pem_x509_certificate(
                    str.encode(server_pem),
                    default_backend())
            self.p.add_auth(method='rsa-sig', data=str.encode(client_file))
            auth_data = None
        else:
            auth_data = b'$3cr3tpa$$w0rd'
            self.p.add_auth(method='shared-key', data=auth_data)
            auth_method = 'shared-key'
            client_priv = None

        is_init = True if 'is_initiator' not in params else\
            params['is_initiator']

        idr = {'id_type': 'fqdn', 'data': b'vpp.home'}
        idi = {'id_type': 'fqdn', 'data': b'roadwarrior.example.com'}
        if is_init:
            self.p.add_local_id(**idr)
            self.p.add_remote_id(**idi)
        else:
            self.p.add_local_id(**idi)
            self.p.add_remote_id(**idr)

        loc_ts = {'start_addr': '10.10.10.0', 'end_addr': '10.10.10.255'} if\
            'loc_ts' not in params else params['loc_ts']
        rem_ts = {'start_addr': '10.0.0.0', 'end_addr': '10.0.0.255'} if\
            'rem_ts' not in params else params['rem_ts']
        self.p.add_local_ts(**loc_ts)
        self.p.add_remote_ts(**rem_ts)
        if 'responder' in params:
            self.p.add_responder(params['responder'])
        if 'ike_transforms' in params:
            self.p.add_ike_transforms(params['ike_transforms'])
        if 'esp_transforms' in params:
            self.p.add_esp_transforms(params['esp_transforms'])

        self.sa = IKEv2SA(self, i_id=idi['data'], r_id=idr['data'],
                          is_initiator=is_init,
                          id_type=self.p.local_id['id_type'], natt=is_natt,
                          priv_key=client_priv, auth_method=auth_method,
                          auth_data=auth_data,
                          local_ts=self.p.remote_ts, remote_ts=self.p.local_ts)

        ike_crypto = ('AES-CBC', 32) if 'ike-crypto' not in params else\
            params['ike-crypto']
        ike_integ = 'HMAC-SHA1-96' if 'ike-integ' not in params else\
            params['ike-integ']
        ike_dh = '2048MODPgr' if 'ike-dh' not in params else\
            params['ike-dh']

        esp_crypto = ('AES-CBC', 32) if 'esp-crypto' not in params else\
            params['esp-crypto']
        esp_integ = 'HMAC-SHA1-96' if 'esp-integ' not in params else\
            params['esp-integ']

        self.sa.set_ike_props(
                crypto=ike_crypto[0], crypto_key_len=ike_crypto[1],
                integ=ike_integ, prf='PRF_HMAC_SHA2_256', dh=ike_dh)
        self.sa.set_esp_props(
                crypto=esp_crypto[0], crypto_key_len=esp_crypto[1],
                integ=esp_integ)


class TestApi(VppTestCase):
    """ Test IKEV2 API """
    @classmethod
    def setUpClass(cls):
        super(TestApi, cls).setUpClass()

    @classmethod
    def tearDownClass(cls):
        super(TestApi, cls).tearDownClass()

    def tearDown(self):
        super(TestApi, self).tearDown()
        self.p1.remove_vpp_config()
        self.p2.remove_vpp_config()
        r = self.vapi.ikev2_profile_dump()
        self.assertEqual(len(r), 0)

    def configure_profile(self, cfg):
        p = Profile(self, cfg['name'])
        p.add_local_id(id_type=cfg['loc_id'][0], data=cfg['loc_id'][1])
        p.add_remote_id(id_type=cfg['rem_id'][0], data=cfg['rem_id'][1])
        p.add_local_ts(**cfg['loc_ts'])
        p.add_remote_ts(**cfg['rem_ts'])
        p.add_responder(cfg['responder'])
        p.add_ike_transforms(cfg['ike_ts'])
        p.add_esp_transforms(cfg['esp_ts'])
        p.add_auth(**cfg['auth'])
        p.set_udp_encap(cfg['udp_encap'])
        p.set_ipsec_over_udp_port(cfg['ipsec_over_udp_port'])
        if 'lifetime_data' in cfg:
            p.set_lifetime_data(cfg['lifetime_data'])
        if 'tun_itf' in cfg:
            p.set_tunnel_interface(cfg['tun_itf'])
        p.add_vpp_config()
        return p

    def test_profile_api(self):
        """ test profile dump API """
        loc_ts4 = {
                    'proto': 8,
                    'start_port': 1,
                    'end_port': 19,
                    'start_addr': '3.3.3.2',
                    'end_addr': '3.3.3.3',
                }
        rem_ts4 = {
                    'proto': 9,
                    'start_port': 10,
                    'end_port': 119,
                    'start_addr': '4.5.76.80',
                    'end_addr': '2.3.4.6',
                }

        loc_ts6 = {
                    'proto': 8,
                    'start_port': 1,
                    'end_port': 19,
                    'start_addr': 'ab::1',
                    'end_addr': 'ab::4',
                }
        rem_ts6 = {
                    'proto': 9,
                    'start_port': 10,
                    'end_port': 119,
                    'start_addr': 'cd::12',
                    'end_addr': 'cd::13',
                }

        conf = {
            'p1': {
                'name': 'p1',
                'loc_id': ('fqdn', b'vpp.home'),
                'rem_id': ('fqdn', b'roadwarrior.example.com'),
                'loc_ts': loc_ts4,
                'rem_ts': rem_ts4,
                'responder': {'sw_if_index': 0, 'addr': '5.6.7.8'},
                'ike_ts': {
                        'crypto_alg': 20,
                        'crypto_key_size': 32,
                        'integ_alg': 1,
                        'dh_group': 1},
                'esp_ts': {
                        'crypto_alg': 13,
                        'crypto_key_size': 24,
                        'integ_alg': 2},
                'auth': {'method': 'shared-key', 'data': b'sharedkeydata'},
                'udp_encap': True,
                'ipsec_over_udp_port': 4501,
                'lifetime_data': {
                    'lifetime': 123,
                    'lifetime_maxdata': 20192,
                    'lifetime_jitter': 9,
                    'handover': 132},
            },
            'p2': {
                'name': 'p2',
                'loc_id': ('ip4-addr', b'192.168.2.1'),
                'rem_id': ('ip6-addr', b'abcd::1'),
                'loc_ts': loc_ts6,
                'rem_ts': rem_ts6,
                'responder': {'sw_if_index': 4, 'addr': 'def::10'},
                'ike_ts': {
                        'crypto_alg': 12,
                        'crypto_key_size': 16,
                        'integ_alg': 3,
                        'dh_group': 3},
                'esp_ts': {
                        'crypto_alg': 9,
                        'crypto_key_size': 24,
                        'integ_alg': 4},
                'auth': {'method': 'shared-key', 'data': b'sharedkeydata'},
                'udp_encap': False,
                'ipsec_over_udp_port': 4600,
                'tun_itf': 0}
        }
        self.p1 = self.configure_profile(conf['p1'])
        self.p2 = self.configure_profile(conf['p2'])

        r = self.vapi.ikev2_profile_dump()
        self.assertEqual(len(r), 2)
        self.verify_profile(r[0].profile, conf['p1'])
        self.verify_profile(r[1].profile, conf['p2'])

    def verify_id(self, api_id, cfg_id):
        self.assertEqual(api_id.type, IDType.value(cfg_id[0]))
        self.assertEqual(bytes(api_id.data, 'ascii'), cfg_id[1])

    def verify_ts(self, api_ts, cfg_ts):
        self.assertEqual(api_ts.protocol_id, cfg_ts['proto'])
        self.assertEqual(api_ts.start_port, cfg_ts['start_port'])
        self.assertEqual(api_ts.end_port, cfg_ts['end_port'])
        self.assertEqual(api_ts.start_addr,
                         ip_address(text_type(cfg_ts['start_addr'])))
        self.assertEqual(api_ts.end_addr,
                         ip_address(text_type(cfg_ts['end_addr'])))

    def verify_responder(self, api_r, cfg_r):
        self.assertEqual(api_r.sw_if_index, cfg_r['sw_if_index'])
        self.assertEqual(api_r.addr, ip_address(cfg_r['addr']))

    def verify_transforms(self, api_ts, cfg_ts):
        self.assertEqual(api_ts.crypto_alg, cfg_ts['crypto_alg'])
        self.assertEqual(api_ts.crypto_key_size, cfg_ts['crypto_key_size'])
        self.assertEqual(api_ts.integ_alg, cfg_ts['integ_alg'])

    def verify_ike_transforms(self, api_ts, cfg_ts):
        self.verify_transforms(api_ts, cfg_ts)
        self.assertEqual(api_ts.dh_group, cfg_ts['dh_group'])

    def verify_esp_transforms(self, api_ts, cfg_ts):
        self.verify_transforms(api_ts, cfg_ts)

    def verify_auth(self, api_auth, cfg_auth):
        self.assertEqual(api_auth.method, AuthMethod.value(cfg_auth['method']))
        self.assertEqual(api_auth.data, cfg_auth['data'])
        self.assertEqual(api_auth.data_len, len(cfg_auth['data']))

    def verify_lifetime_data(self, p, ld):
        self.assertEqual(p.lifetime, ld['lifetime'])
        self.assertEqual(p.lifetime_maxdata, ld['lifetime_maxdata'])
        self.assertEqual(p.lifetime_jitter, ld['lifetime_jitter'])
        self.assertEqual(p.handover, ld['handover'])

    def verify_profile(self, ap, cp):
        self.assertEqual(ap.name, cp['name'])
        self.assertEqual(ap.udp_encap, cp['udp_encap'])
        self.verify_id(ap.loc_id, cp['loc_id'])
        self.verify_id(ap.rem_id, cp['rem_id'])
        self.verify_ts(ap.loc_ts, cp['loc_ts'])
        self.verify_ts(ap.rem_ts, cp['rem_ts'])
        self.verify_responder(ap.responder, cp['responder'])
        self.verify_ike_transforms(ap.ike_ts, cp['ike_ts'])
        self.verify_esp_transforms(ap.esp_ts, cp['esp_ts'])
        self.verify_auth(ap.auth, cp['auth'])
        if 'lifetime_data' in cp:
            self.verify_lifetime_data(ap, cp['lifetime_data'])
        self.assertEqual(ap.ipsec_over_udp_port, cp['ipsec_over_udp_port'])
        if 'tun_itf' in cp:
            self.assertEqual(ap.tun_itf, cp['tun_itf'])
        else:
            self.assertEqual(ap.tun_itf, 0xffffffff)


class TestInitiatorPsk(TemplateInitiator, Ikev2Params):
    """ test ikev2 initiator - pre shared key auth """
    def config_tc(self):
        self.config_params({
            'is_initiator': False,  # seen from test case perspective
                                    # thus vpp is initiator
            'responder': {'sw_if_index': self.pg0.sw_if_index,
                           'addr': self.pg0.remote_ip4},
            'ike-crypto': ('AES-GCM-16ICV', 32),
            'ike-integ': 'NULL',
            'ike-dh': '3072MODPgr',
            'ike_transforms': {
                'crypto_alg': 20,  # "aes-gcm-16"
                'crypto_key_size': 256,
                'dh_group': 15,  # "modp-3072"
            },
            'esp_transforms': {
                'crypto_alg': 12,  # "aes-cbc"
                'crypto_key_size': 256,
                # "hmac-sha2-256-128"
                'integ_alg': 12}})


class TestResponderNATT(TemplateResponder, Ikev2Params):
    """ test ikev2 responder - nat traversal """
    def config_tc(self):
        self.config_params(
                {'natt': True})


class TestResponderPsk(TemplateResponder, Ikev2Params):
    """ test ikev2 responder - pre shared key auth """
    def config_tc(self):
        self.config_params()


class TestResponderRsaSign(TemplateResponder, Ikev2Params):
    """ test ikev2 responder - cert based auth """
    def config_tc(self):
        self.config_params({
            'auth': 'rsa-sig',
            'server-key': 'server-key.pem',
            'client-key': 'client-key.pem',
            'client-cert': 'client-cert.pem',
            'server-cert': 'server-cert.pem'})


class Test_IKE_AES_CBC_128_SHA256_128_MODP2048_ESP_AES_CBC_192_SHA_384_192\
        (TemplateResponder, Ikev2Params):
    """
    IKE:AES_CBC_128_SHA256_128,DH=modp2048 ESP:AES_CBC_192_SHA_384_192
    """
    def config_tc(self):
        self.config_params({
            'ike-crypto': ('AES-CBC', 16),
            'ike-integ': 'SHA2-256-128',
            'esp-crypto': ('AES-CBC', 24),
            'esp-integ': 'SHA2-384-192',
            'ike-dh': '2048MODPgr'})


class TestAES_CBC_128_SHA256_128_MODP3072_ESP_AES_GCM_16\
        (TemplateResponder, Ikev2Params):
    """
    IKE:AES_CBC_128_SHA256_128,DH=modp3072 ESP:AES_GCM_16
    """
    def config_tc(self):
        self.config_params({
            'ike-crypto': ('AES-CBC', 32),
            'ike-integ': 'SHA2-256-128',
            'esp-crypto': ('AES-GCM-16ICV', 32),
            'esp-integ': 'NULL',
            'ike-dh': '3072MODPgr'})


class Test_IKE_AES_GCM_16_256(TemplateResponder, Ikev2Params):
    """
    IKE:AES_GCM_16_256
    """
    def config_tc(self):
        self.config_params({
            'ip6': True,
            'natt': True,
            'ike-crypto': ('AES-GCM-16ICV', 32),
            'ike-integ': 'NULL',
            'ike-dh': '2048MODPgr',
            'loc_ts': {'start_addr': 'ab:cd::0',
                       'end_addr': 'ab:cd::10'},
            'rem_ts': {'start_addr': '11::0',
                       'end_addr': '11::100'}})


class TestMalformedMessages(TemplateResponder, Ikev2Params):
    """ malformed packet test """

    def tearDown(self):
        pass

    def config_tc(self):
        self.config_params()

    def assert_counter(self, count, name, version='ip4'):
        node_name = '/err/ikev2-%s/' % version + name
        self.assertEqual(count, self.statistics.get_err_counter(node_name))

    def create_ike_init_msg(self, length=None, payload=None):
        msg = ikev2.IKEv2(length=length, init_SPI='\x11' * 8,
                          flags='Initiator', exch_type='IKE_SA_INIT')
        if payload is not None:
            msg /= payload
        return self.create_packet(self.pg0, msg, self.sa.sport,
                                  self.sa.dport)

    def verify_bad_packet_length(self):
        ike_msg = self.create_ike_init_msg(length=0xdead)
        self.send_and_assert_no_replies(self.pg0, ike_msg * self.pkt_count)
        self.assert_counter(self.pkt_count, 'Bad packet length')

    def verify_bad_sa_payload_length(self):
        p = ikev2.IKEv2_payload_SA(length=0xdead)
        ike_msg = self.create_ike_init_msg(payload=p)
        self.send_and_assert_no_replies(self.pg0, ike_msg * self.pkt_count)
        self.assert_counter(self.pkt_count, 'Malformed packet')

    def test_responder(self):
        self.pkt_count = 254
        self.verify_bad_packet_length()
        self.verify_bad_sa_payload_length()


if __name__ == '__main__':
    unittest.main(testRunner=VppTestRunner)