DHCPv6 Proxy test update
[csit.git] / resources / traffic_scripts / dhcp / send_dhcp_v6_messages.py
1 #!/usr/bin/env python
2 # Copyright (c) 2016 Cisco and/or its affiliates.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at:
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14
15 """Traffic script that sends DHCPv6 proxy packets."""
16
17 from scapy.layers.dhcp6 import *
18 from scapy.layers.inet6 import IPv6, UDP, UDP_SERVICES
19
20 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
21 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
22
23
24 def _check_udp_checksum(pkt):
25     """Check udp checksum in ip packet.
26     Return true if checksum is correct."""
27     new = pkt.__class__(str(pkt))
28     del new['UDP'].chksum
29     new = new.__class__(str(new))
30     return new['UDP'].chksum == pkt['UDP'].chksum
31
32
33 def dhcpv6_solicit(tx_if, rx_if, dhcp_multicast_ip, link_local_ip, proxy_ip,
34                    server_ip, server_mac, client_duid, client_mac):
35     """Send and check DHCPv6 SOLICIT proxy packet.
36
37     :param tx_if: Client interface.
38     :param rx_if: DHCPv6 server interface.
39     :param dhcp_multicast_ip: Servers and relay agents multicast address.
40     :param link_local_ip: Client link-local address.
41     :param proxy_ip: IP address of DHCPv6 proxy server.
42     :param server_ip: IP address of DHCPv6 server.
43     :param server_mac: MAC address of DHCPv6 server.
44     :param client_duid: Client DHCP Unique Identifier.
45     :param client_mac: Client MAC address.
46     :type tx_if: str
47     :type rx_if: str
48     :type dhcp_multicast_ip: str
49     :type link_local_ip: str
50     :type proxy_ip: str
51     :type server_ip: str
52     :type server_mac: str
53     :type client_duid: str
54     :type client_mac: str
55     :return interface_id: ID of proxy interface.
56     :rtype interface_id: str
57     """
58
59     rxq = RxQueue(rx_if)
60     txq = TxQueue(tx_if)
61
62     sent_packets = []
63
64     dhcp6_solicit_pkt = Ether(src=client_mac, dst="33:33:00:01:00:02") / \
65                         IPv6(src=link_local_ip, dst=dhcp_multicast_ip) / \
66                         UDP(sport=UDP_SERVICES.dhcpv6_client,
67                             dport=UDP_SERVICES.dhcpv6_server) / \
68                         DHCP6_Solicit() / \
69                         DHCP6OptClientId(duid=client_duid)
70
71     sent_packets.append(dhcp6_solicit_pkt)
72     txq.send(dhcp6_solicit_pkt)
73
74     ether = rxq.recv(2)
75
76     if ether is None:
77         raise RuntimeError('DHCPv6 SOLICIT timeout')
78
79     if ether.dst != server_mac:
80         raise RuntimeError("Destination MAC address error!")
81     print "Destination MAC address: OK."
82
83     if ether['IPv6'].src != proxy_ip:
84         raise RuntimeError("Source IP address error!")
85     print "Source IP address: OK."
86
87     if ether['IPv6'].dst != server_ip:
88         raise RuntimeError("Destination IP address error!")
89     print "Destination IP address: OK."
90
91     if ether['IPv6']['UDP']\
92         ['DHCPv6 Relay Forward Message (Relay Agent/Server Message)']:
93         print "Relay Agent/Server Message: OK."
94     else:
95         raise RuntimeError("Relay Agent/Server Message error.")
96
97     if ether['IPv6']['UDP']\
98         ['DHCPv6 Relay Forward Message (Relay Agent/Server Message)']\
99         .linkaddr != proxy_ip:
100         raise RuntimeError("Proxy IP address error!")
101     print "Proxy IP address: OK."
102
103     try:
104         interface_id =  ether['IPv6']['UDP']\
105             ['DHCPv6 Relay Forward Message (Relay Agent/Server Message)']\
106             ['Unknown DHCPv6 OPtion']['DHCP6 Interface-Id Option'].ifaceid
107     except Exception:
108         raise RuntimeError("DHCP6 Interface-Id error!")
109
110     return interface_id
111
112
113 def dhcpv6_advertise(rx_if, tx_if, link_local_ip, proxy_ip,
114                      server_ip, server_mac, proxy_to_server_mac, interface_id):
115     """Send and check DHCPv6 ADVERTISE proxy packet.
116
117     :param rx_if: DHCPv6 server interface.
118     :param tx_if: Client interface.
119     :param link_local_ip: Client link-local address.
120     :param proxy_ip: IP address of DHCPv6 proxy server.
121     :param server_ip: IP address of DHCPv6 server.
122     :param server_mac: MAC address of DHCPv6 server.
123     :param proxy_to_server_mac: MAC address of DHCPv6 proxy interface.
124     :param interface_id: ID of proxy interface.
125     :type rx_if: str
126     :type tx_if: str
127     :type link_local_ip: str
128     :type proxy_ip: str
129     :type server_ip: str
130     :type server_mac: str
131     :type proxy_to_server_mac: str
132     :type interface_id: str
133     """
134
135     rxq = RxQueue(rx_if)
136     txq = TxQueue(tx_if)
137
138     sent_packets = []
139
140     dhcp6_advertise_pkt = Ether(src=server_mac, dst=proxy_to_server_mac) / \
141                           IPv6(src=server_ip, dst=proxy_ip) / \
142                           UDP(sport=UDP_SERVICES.dhcpv6_server,
143                               dport=UDP_SERVICES.dhcpv6_client) / \
144                           DHCP6_RelayReply(peeraddr=link_local_ip,
145                                            linkaddr=proxy_ip) / \
146                           DHCP6OptIfaceId(ifaceid=interface_id) / \
147                           DHCP6OptRelayMsg() / \
148                           DHCP6_Advertise()
149
150     sent_packets.append(dhcp6_advertise_pkt)
151     txq.send(dhcp6_advertise_pkt)
152
153     ether = rxq.recv(2)
154
155     if ether is None:
156         raise RuntimeError('DHCPv6 ADVERTISE timeout')
157
158     if ether['IPv6'].src != proxy_ip:
159         raise RuntimeError("Source IP address error!")
160     print "Source IP address: OK."
161
162     if not _check_udp_checksum(ether['IPv6']):
163         raise RuntimeError("Checksum error!")
164     print "Checksum: OK."
165
166     if ether['IPv6']['UDP']['DHCPv6 Advertise Message'].msgtype != 'ADVERTISE':
167         raise RuntimeError("Message type error!")
168     print "Message type: OK."
169
170
171 def dhcpv6_request(tx_if, rx_if, dhcp_multicast_ip, link_local_ip, proxy_ip,
172                    server_ip, client_duid, client_mac):
173     """Send and check DHCPv6 REQUEST proxy packet.
174
175     :param tx_if: Client interface.
176     :param rx_if: DHCPv6 server interface.
177     :param dhcp_multicast_ip: Servers and relay agents multicast address.
178     :param link_local_ip: Client link-local address.
179     :param proxy_ip: IP address of DHCPv6 proxy server.
180     :param server_ip: IP address of DHCPv6 server.
181     :param client_duid: Client DHCP Unique Identifier.
182     :param client_mac: Client MAC address.
183     :type tx_if: str
184     :type rx_if: str
185     :type dhcp_multicast_ip: str
186     :type link_local_ip: str
187     :type proxy_ip: str
188     :type server_ip: str
189     :type client_duid: str
190     :type client_mac: str
191     """
192
193     rxq = RxQueue(rx_if)
194     txq = TxQueue(tx_if)
195
196     sent_packets = []
197
198     dhcp6_request_pkt = Ether(src=client_mac, dst="33:33:00:01:00:02") / \
199                         IPv6(src=link_local_ip, dst=dhcp_multicast_ip) / \
200                         UDP(sport=UDP_SERVICES.dhcpv6_client,
201                             dport=UDP_SERVICES.dhcpv6_server) / \
202                         DHCP6_Request() / \
203                         DHCP6OptClientId(duid=client_duid)
204
205     sent_packets.append(dhcp6_request_pkt)
206     txq.send(dhcp6_request_pkt)
207
208     ether = rxq.recv(2)
209
210     if ether is None:
211         raise RuntimeError('DHCPv6 REQUEST timeout')
212
213     if ether['IPv6'].src != proxy_ip:
214         raise RuntimeError("Source IP address error!")
215     print "Source IP address: OK."
216
217     if ether['IPv6'].dst != server_ip:
218         raise RuntimeError("Destination IP address error!")
219     print "Destination IP address: OK."
220
221     if ether['IPv6']['UDP']\
222         ['DHCPv6 Relay Forward Message (Relay Agent/Server Message)']:
223         print "Relay Agent/Server Message: OK."
224     else:
225         raise RuntimeError("Relay Agent/Server Message error.")
226
227     if ether['IPv6']['UDP']\
228             ['DHCPv6 Relay Forward Message (Relay Agent/Server Message)']\
229             .linkaddr != proxy_ip:
230         raise RuntimeError("Proxy IP address error!")
231     print "Proxy IP address: OK."
232
233
234 def dhcpv6_reply(rx_if, tx_if, link_local_ip, proxy_ip, server_ip, server_mac,
235                  interface_id):
236     """Send and check DHCPv6 REPLY proxy packet.
237
238     :param rx_if: DHCPv6 server interface.
239     :param tx_if: Client interface.
240     :param link_local_ip: Client link-local address.
241     :param proxy_ip: IP address of DHCPv6 proxy server.
242     :param server_ip: IP address of DHCPv6 server.
243     :param server_mac: MAC address of DHCPv6 server.
244     :param interface_id: ID of proxy interface.
245     :type rx_if: str
246     :type tx_if: str
247     :type link_local_ip: str
248     :type proxy_ip: str
249     :type server_ip: str
250     :type server_mac: str
251     :type interface_id: str
252     """
253
254     rxq = RxQueue(rx_if)
255     txq = TxQueue(tx_if)
256
257     sent_packets = []
258
259     dhcp_reply_pkt = Ether(src=server_mac) / \
260                     IPv6(src=server_ip, dst=proxy_ip) / \
261                     UDP(sport=UDP_SERVICES.dhcpv6_server,
262                         dport=UDP_SERVICES.dhcpv6_client) / \
263                     DHCP6_RelayReply(peeraddr=link_local_ip,
264                                      linkaddr=proxy_ip) / \
265                     DHCP6OptIfaceId(ifaceid=interface_id) / \
266                     DHCP6OptRelayMsg() / \
267                     DHCP6_Reply()
268
269     sent_packets.append(dhcp_reply_pkt)
270     txq.send(dhcp_reply_pkt)
271
272     ether = rxq.recv(2)
273
274     if ether is None:
275         raise RuntimeError('DHCPv6 REPLY timeout')
276
277     if ether['IPv6'].src != proxy_ip:
278         raise RuntimeError("Source IP address error!")
279     print "Source IP address: OK."
280
281     if not _check_udp_checksum(ether['IPv6']):
282         raise RuntimeError("Checksum error!")
283     print "Checksum: OK."
284
285     if ether['IPv6']['UDP']['DHCPv6 Reply Message'].msgtype != 'REPLY':
286         raise RuntimeError("Message type error!")
287     print "Message type: OK."
288
289
290 def main():
291     """Send DHCPv6 proxy messages."""
292
293     args = TrafficScriptArg(['tx_src_ip', 'tx_dst_ip', 'proxy_ip', 'proxy_mac',
294                              'server_ip', 'client_mac', 'server_mac',
295                              'proxy_to_server_mac'])
296
297     client_if = args.get_arg('tx_if')
298     server_if = args.get_arg('rx_if')
299     proxy_ip = args.get_arg('proxy_ip')
300     proxy_mac = args.get_arg('proxy_mac')
301     proxy_to_server_mac = args.get_arg('proxy_to_server_mac')
302     server_ip = args.get_arg('server_ip')
303     client_mac = args.get_arg('client_mac')
304     server_mac = args.get_arg('server_mac')
305
306     link_local_ip = "fe80::1"
307     dhcp_multicast_ip = "ff02::1:2"
308     client_duid = str(random.randint(0, 9999))
309
310     # SOLICIT
311     interface_id = dhcpv6_solicit(client_if, server_if, dhcp_multicast_ip,
312                                   link_local_ip, proxy_ip, server_ip,
313                                   server_mac, client_duid, client_mac)
314
315     # ADVERTISE
316     dhcpv6_advertise(client_if, server_if, link_local_ip, proxy_ip,
317                      server_ip, server_mac, proxy_to_server_mac, interface_id)
318
319     # REQUEST
320     dhcpv6_request(client_if, server_if, dhcp_multicast_ip, link_local_ip,
321                    proxy_ip, server_ip, client_duid, client_mac)
322
323     # REPLY
324     dhcpv6_reply(client_if, server_if, link_local_ip, proxy_ip, server_ip,
325                  server_mac, interface_id)
326
327     sys.exit(0)
328
329 if __name__ == "__main__":
330     main()

©2016 FD.io a Linux Foundation Collaborative Project. All Rights Reserved.
Linux Foundation is a registered trademark of The Linux Foundation. Linux is a registered trademark of Linus Torvalds.
Please see our privacy policy and terms of use.