1 # Copyright (c) 2020 Cisco and/or its affiliates.
2 # Licensed under the Apache License, Version 2.0 (the "License");
3 # you may not use this file except in compliance with the License.
4 # You may obtain a copy of the License at:
6 # http://www.apache.org/licenses/LICENSE-2.0
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS,
10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11 # See the License for the specific language governing permissions and
12 # limitations under the License.
14 """Segment Routing over IPv6 data plane utilities library."""
16 from enum import IntEnum
18 from ipaddress import ip_address, IPv6Address
20 from resources.libraries.python.Constants import Constants
21 from resources.libraries.python.InterfaceUtil import InterfaceUtil
22 from resources.libraries.python.IPAddress import IPAddress
23 from resources.libraries.python.IPUtil import IPUtil
24 from resources.libraries.python.PapiExecutor import PapiSocketExecutor
27 class SRv6Behavior(IntEnum):
28 """SRv6 LocalSID supported functions."""
31 # Endpoint function with Layer-3 cross-connect
33 # Endpoint with decapsulation and Layer-2 cross-connect
35 # Endpoint with decapsulation and IPv6 cross-connect
37 # Endpoint with decapsulation and IPv4 cross-connect
39 # Endpoint with decapsulation and IPv6 table lookup
41 # Endpoint with decapsulation and IPv4 table lookup
43 # Endpoint to SR-unaware appliance via static proxy
45 # Endpoint to SR-unaware appliance via dynamic proxy
47 # Endpoint to SR-unaware appliance via masquerading
51 class SRv6PolicySteeringTypes(IntEnum):
52 """SRv6 steering types."""
62 def create_srv6_sid_list(sids):
63 """Create SRv6 SID list object.
65 :param sids: SID IPv6 addresses.
67 :returns: SRv6 SID list object.
70 sid_list = [IPv6Address(sid).packed for sid in sids]
73 num_sids=len(sid_list),
75 sids=sid_list + (16 - len(sid_list)) * [IPv6Address(0).packed]
79 def configure_sr_localsid(
80 node, local_sid, behavior, interface=None, next_hop=None,
81 fib_table=None, out_if=None, in_if=None, src_addr=None,
83 """Create SRv6 LocalSID and binds it to a particular behaviour on
86 :param node: Given node to create localSID on.
87 :param local_sid: LocalSID IPv6 address.
88 :param behavior: SRv6 LocalSID function.
89 :param interface: Interface name (Optional, required for
91 :param next_hop: Next hop IPv4/IPv6 address (Optional, required for L3
93 :param fib_table: FIB table for IPv4/IPv6 lookup (Optional, required for
95 :param out_if: Interface name of local interface for sending traffic
96 towards the Service Function (Optional, required for SRv6 endpoint
97 to SR-unaware appliance).
98 :param in_if: Interface name of local interface receiving the traffic
99 coming back from the Service Function (Optional, required for SRv6
100 endpoint to SR-unaware appliance).
101 :param src_addr: Source address on the packets coming back on in_if
102 interface (Optional, required for SRv6 endpoint to SR-unaware
103 appliance via static proxy).
104 :param sid_list: SID list (Optional, required for SRv6 endpoint to
105 SR-unaware appliance via static proxy).
116 :raises ValueError: If required parameter is missing.
118 beh = behavior.replace(u".", u"_").upper()
119 # There is no SRv6Behaviour enum defined for functions from SRv6 plugins
120 # so we need to use CLI command to configure it.
121 if beh in (getattr(SRv6Behavior, u"END_AD").name,
122 getattr(SRv6Behavior, u"END_AS").name,
123 getattr(SRv6Behavior, u"END_AM").name):
124 if beh == getattr(SRv6Behavior, u"END_AS").name:
125 if next_hop is None or out_if is None or in_if is None or \
126 src_addr is None or sid_list is None:
128 f"Required parameter(s) missing.\n"
129 f"next_hop:{next_hop}\n "
132 f"src_addr:{src_addr}\n"
133 f"sid_list:{sid_list}"
135 sid_conf = f"next {u' next '.join(sid_list)}"
136 params = f"nh {next_hop} oif {out_if} iif {in_if} " \
137 f"src {src_addr} {sid_conf}"
139 if next_hop is None or out_if is None or in_if is None:
141 f"Required parameter(s) missing.\n"
142 f"next_hop:{next_hop}\n"
146 params = f"nh {next_hop} oif {out_if} iif {in_if}"
148 cli_cmd = f"sr localsid address {local_sid} behavior {behavior} " \
151 PapiSocketExecutor.run_cli_cmd(node, cli_cmd)
154 cmd = u"sr_localsid_add_del"
157 localsid=IPv6Address(local_sid).packed,
159 behavior=getattr(SRv6Behavior, beh).value,
160 sw_if_index=Constants.BITWISE_NON_ZERO,
165 err_msg = f"Failed to add SR localSID {local_sid} " \
166 f"host {node[u'host']}"
167 if beh in (getattr(SRv6Behavior, u"END_X").name,
168 getattr(SRv6Behavior, u"END_DX4").name,
169 getattr(SRv6Behavior, u"END_DX6").name):
170 if interface is None or next_hop is None:
172 f"Required parameter(s) missing.\n"
173 f"interface:{interface}\n"
174 f"next_hop:{next_hop}"
176 args[u"sw_if_index"] = InterfaceUtil.get_interface_index(
179 args[u"nh_addr"] = IPAddress.create_ip_address_object(
182 elif beh == getattr(SRv6Behavior, u"END_DX2").name:
183 if interface is None:
185 f"Required parameter missing.\ninterface: {interface}"
187 args[u"sw_if_index"] = InterfaceUtil.get_interface_index(
190 elif beh in (getattr(SRv6Behavior, u"END_DT4").name,
191 getattr(SRv6Behavior, u"END_DT6").name):
192 if fib_table is None:
194 f"Required parameter missing.\n"
195 f"fib_table: {fib_table}"
197 args[u"fib_table"] = fib_table
199 with PapiSocketExecutor(node) as papi_exec:
200 papi_exec.add(cmd, **args).get_reply(err_msg)
203 def show_sr_localsids(node):
204 """Show SRv6 LocalSIDs on the given node.
206 :param node: Given node to show localSIDs on.
209 cmd = u"sr_localsids_dump"
210 PapiSocketExecutor.dump_and_log(node, (cmd,))
213 def configure_sr_policy(node, bsid, sid_list, mode=u"encap"):
214 """Create SRv6 policy on the given node.
216 :param node: Given node to create SRv6 policy on.
217 :param bsid: BindingSID - local SID IPv6 address.
218 :param sid_list: SID list.
219 :param mode: Encapsulation / insertion mode.
225 cmd = u"sr_policy_add"
227 bsid_addr=IPv6Address(bsid).packed,
229 is_encap=bool(mode == u"encap"),
231 sids=SRv6.create_srv6_sid_list(sid_list)
233 err_msg = f"Failed to add SR policy for BindingSID {bsid} " \
234 f"on host {node[u'host']}"
236 with PapiSocketExecutor(node) as papi_exec:
237 papi_exec.add(cmd, **args).get_reply(err_msg)
240 def show_sr_policies(node):
241 """Show SRv6 policies on the given node.
243 :param node: Given node to show SRv6 policies on.
246 cmd = u"sr_policies_dump"
247 PapiSocketExecutor.dump_and_log(node, (cmd,))
250 def _get_sr_steer_policy_args(
251 node, mode, interface=None, ip_addr=None, prefix=None):
252 """Return values of sw_if_index, mask_width, prefix_addr and
253 traffic_type for sr_steering_add_del API.
255 :param node: Given node to create/delete steering policy on.
256 :param mode: Mode of operation - L2 or L3.
257 :param interface: Interface name (Optional, required in case of
259 :param ip_addr: IPv4/IPv6 address (Optional, required in case of L3
261 :param prefix: IP address prefix (Optional, required in case of L3
268 :returns: Values for sw_if_index, prefix and traffic_type
270 :raises ValueError: If unsupported mode used or required parameter
273 if mode.lower() == u"l2":
274 if interface is None:
276 f"Required data missing.\n"
277 f"interface: {interface}"
279 sw_if_index = InterfaceUtil.get_interface_index(node, interface)
281 traffic_type = getattr(
282 SRv6PolicySteeringTypes, u"SR_STEER_L2"
284 elif mode.lower() == u"l3":
285 if ip_addr is None or prefix is None:
287 f"Required data missing.\n"
288 f"IP address:{ip_addr}\n"
291 sw_if_index = Constants.BITWISE_NON_ZERO
292 ip_addr = ip_address(ip_addr)
293 prefix = IPUtil.create_prefix_object(ip_addr, int(prefix))
294 traffic_type = getattr(
295 SRv6PolicySteeringTypes, u"SR_STEER_IPV4"
296 ).value if ip_addr.version == 4 else getattr(
297 SRv6PolicySteeringTypes, u"SR_STEER_IPV6"
300 raise ValueError(f"Unsupported mode: {mode}")
302 return sw_if_index, prefix, traffic_type
304 # TODO: Bring L1 names, arguments and defaults closer to PAPI ones.
306 def configure_sr_steer(
307 node, mode, bsid, interface=None, ip_addr=None, prefix=None):
308 """Create SRv6 steering policy on the given node.
310 :param node: Given node to create steering policy on.
311 :param mode: Mode of operation - L2 or L3.
312 :param bsid: BindingSID - local SID IPv6 address.
313 :param interface: Interface name (Optional, required in case of
315 :param ip_addr: IPv4/IPv6 address (Optional, required in case of L3
317 :param prefix: IP address prefix (Optional, required in case of L3
325 :raises ValueError: If unsupported mode used or required parameter
328 sw_if_index, prefix, traffic_type = SRv6._get_sr_steer_policy_args(
329 node, mode, interface, ip_addr, prefix
332 cmd = u"sr_steering_add_del"
335 bsid_addr=IPv6Address(str(bsid)).packed,
339 sw_if_index=sw_if_index,
340 traffic_type=traffic_type
342 err_msg = f"Failed to add SRv6 steering policy for BindingSID {bsid} " \
343 f"on host {node[u'host']}"
345 with PapiSocketExecutor(node) as papi_exec:
346 papi_exec.add(cmd, **args).get_reply(err_msg)
349 def show_sr_steering_policies(node):
350 """Show SRv6 steering policies on the given node.
352 :param node: Given node to show SRv6 steering policies on.
355 cmd = u"sr_steering_pol_dump"
356 PapiSocketExecutor.dump_and_log(node, (cmd,))
359 def set_sr_encaps_source_address(node, ip6_addr):
360 """Set SRv6 encapsulation source address on the given node.
362 :param node: Given node to set SRv6 encapsulation source address on.
363 :param ip6_addr: Local SID IPv6 address.
367 cmd = u"sr_set_encap_source"
369 encaps_source=IPv6Address(ip6_addr).packed
371 err_msg = f"Failed to set SRv6 encapsulation source address " \
372 f"{ip6_addr} on host {node[u'host']}"
374 with PapiSocketExecutor(node) as papi_exec:
375 papi_exec.add(cmd, **args).get_reply(err_msg)