blob: e5aadf36db85ad6139298d837f537961a24942b0 [file] [log] [blame]
# Copyright 2017 The Android Open Source Project
# 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
from socket import * # pylint: disable=wildcard-import
from scapy import all as scapy
import binascii
import struct
import csocket
import cstruct
import multinetwork_base
import net_test
import util
import xfrm
_ENCRYPTION_KEY_256 = binascii.unhexlify("308146eb3bd84b044573d60f5a5fd159"
_AUTHENTICATION_KEY_128 = binascii.unhexlify("af442892cdcd0ef650e9c299f9a8436a")
_ALGO_AUTH_NULL = (xfrm.XfrmAlgoAuth((b"digest_null", 0, 0)), b"")
_ALGO_HMAC_SHA1 = (xfrm.XfrmAlgoAuth((xfrm.XFRM_AALG_HMAC_SHA1, 128, 96)),
_ALGO_CRYPT_NULL = (xfrm.XfrmAlgo((b"ecb(cipher_null)", 0)), b"")
_ALGO_CBC_AES_256 = (xfrm.XfrmAlgo((xfrm.XFRM_EALG_CBC_AES, 256)),
# Match all bits of the mark
MARK_MASK_ALL = 0xffffffff
def SetPolicySockopt(sock, family, opt_data):
optlen = len(opt_data) if opt_data is not None else 0
if family == AF_INET:
csocket.Setsockopt(sock, IPPROTO_IP, xfrm.IP_XFRM_POLICY, opt_data, optlen)
csocket.Setsockopt(sock, IPPROTO_IPV6, xfrm.IPV6_XFRM_POLICY, opt_data,
def ApplySocketPolicy(sock, family, direction, spi, reqid, tun_addrs):
"""Create and apply an ESP policy to a socket.
A socket may have only one policy per direction, so applying a policy will
remove any policy that was previously applied in that direction.
sock: The socket that needs a policy
family: AF_INET or AF_INET6
spi: 32-bit SPI in host byte order
reqid: 32-bit ID matched against SAs
tun_addrs: A tuple of (local, remote) addresses for tunnel mode, or None
to request a transport mode SA.
# Create a selector that matches all packets of the specified address family.
selector = xfrm.EmptySelector(family)
# Create an XFRM policy and template.
policy = xfrm.UserPolicy(direction, selector)
template = xfrm.UserTemplate(family, spi, reqid, tun_addrs)
# Set the policy and template on our socket.
opt_data = policy.Pack() + template.Pack()
# The policy family might not match the socket family. For example, we might
# have an IPv4 policy on a dual-stack socket.
sockfamily = sock.getsockopt(SOL_SOCKET, net_test.SO_DOMAIN)
SetPolicySockopt(sock, sockfamily, opt_data)
def _GetCryptParameters(crypt_alg):
"""Looks up encryption algorithm's block and IV lengths.
crypt_alg: the encryption algorithm constant
A tuple of the block size, and IV length
if crypt_alg == _ALGO_CRYPT_NULL:
return (4, 0)
if crypt_alg == _ALGO_CBC_AES_256:
return (16, 16)
return (0, 0)
def GetEspPacketLength(mode, version, udp_encap, payload,
auth_alg, crypt_alg):
"""Calculates encrypted length of a UDP packet with the given payload.
version: IPPROTO_IP for IPv4, IPPROTO_IPV6 for IPv6. The inner header.
udp_encap: whether UDP encap overhead should be accounted for. Since the
outermost IP header is ignored (payload only), only add for udp
encap'd packets.
payload: UDP payload bytes.
auth_alg: The xfrm_base authentication algorithm used in the SA.
crypt_alg: The xfrm_base encryption algorithm used in the SA.
Return: the packet length.
crypt_iv_len, crypt_blk_size=_GetCryptParameters(crypt_alg)
auth_trunc_len = auth_alg[0].trunc_len
# Wrap in UDP payload
payload_len = len(payload) + net_test.UDP_HDR_LEN
# Size constants
esp_hdr_len = len(xfrm.EspHdr) # SPI + Seq number
icv_len = auth_trunc_len // 8
# Add inner IP header if tunnel mode
if mode == xfrm.XFRM_MODE_TUNNEL:
payload_len += net_test.GetIpHdrLength(version)
# Add ESP trailer
payload_len += 2 # Pad Length + Next Header fields
# Align to block size of encryption algorithm
payload_len += util.GetPadLength(crypt_blk_size, payload_len)
# Add initialization vector, header length and ICV length
payload_len += esp_hdr_len + crypt_iv_len + icv_len
# Add encap as needed
if udp_encap:
payload_len += net_test.UDP_HDR_LEN
return payload_len
def GetEspTrailer(length, nexthdr):
# ESP padding per RFC 4303 section 2.4.
# For a null cipher with a block size of 1, padding is only necessary to
# ensure that the 1-byte Pad Length and Next Header fields are right aligned
# on a 4-byte boundary.
esplen = length + 2 # Packet length plus Pad Length and Next Header.
padlen = util.GetPadLength(4, esplen)
# The pad bytes are consecutive integers starting from 0x01.
padding = "".join((chr(i) for i in range(1, padlen + 1))).encode("utf-8")
return padding + struct.pack("BB", padlen, nexthdr)
def EncryptPacketWithNull(packet, spi, seq, tun_addrs):
"""Apply null encryption to a packet.
This performs ESP encapsulation on the given packet. The returned packet will
be a tunnel mode packet if tun_addrs is provided.
The input packet is assumed to be a UDP packet. The input packet *MUST* have
its length and checksum fields in IP and UDP headers set appropriately. This
can be done by "rebuilding" the scapy object. e.g.,
ip6_packet = scapy.IPv6(bytes(ip6_packet))
TODO: Support TCP
packet: a scapy.IPv6 or scapy.IP packet
spi: security parameter index for ESP header in host byte order
seq: sequence number for ESP header
tun_addrs: A tuple of (local, remote) addresses for tunnel mode, or None
to request a transport mode packet.
The encrypted packet (scapy.IPv6 or scapy.IP)
# The top-level packet changes in tunnel mode, which would invalidate
# the passed-in packet pointer. For consistency, this function now returns
# a new packet and does not modify the user's original packet.
packet = packet.copy()
udp_layer = packet.getlayer(scapy.UDP)
if not udp_layer:
raise ValueError("Expected a UDP packet")
# Build an ESP header.
esp_packet = scapy.Raw(xfrm.EspHdr((spi, seq)).Pack())
if tun_addrs:
tsrc_addr, tdst_addr = tun_addrs
outer_version = net_test.GetAddressVersion(tsrc_addr)
ip_type = {4: scapy.IP, 6: scapy.IPv6}[outer_version]
new_ip_layer = ip_type(src=tsrc_addr, dst=tdst_addr)
inner_layer = packet
esp_nexthdr = {scapy.IPv6: IPPROTO_IPV6,
scapy.IP: IPPROTO_IPIP}[type(packet)]
new_ip_layer = None
inner_layer = udp_layer
esp_nexthdr = IPPROTO_UDP
trailer = GetEspTrailer(len(inner_layer), esp_nexthdr)
# Assemble the packet.
esp_packet.payload = scapy.Raw(inner_layer)
packet = new_ip_layer if new_ip_layer else packet
packet.payload = scapy.Raw(bytes(esp_packet) + trailer)
# TODO: Can we simplify this and avoid the initial copy()?
# Fix the IPv4/IPv6 headers.
if type(packet) is scapy.IPv6:
packet.nh = IPPROTO_ESP
# Recompute plen.
packet.plen = None
packet = scapy.IPv6(bytes(packet))
elif type(packet) is scapy.IP:
packet.proto = IPPROTO_ESP
# Recompute IPv4 len and checksum.
packet.len = None
packet.chksum = None
packet = scapy.IP(bytes(packet))
raise ValueError("First layer in packet should be IPv4 or IPv6: " + repr(packet))
return packet
def DecryptPacketWithNull(packet):
"""Apply null decryption to a packet.
This performs ESP decapsulation on the given packet. The input packet is
assumed to be a UDP packet. This function will remove the ESP header and
trailer bytes from an ESP packet.
TODO: Support TCP
packet: a scapy.IPv6 or scapy.IP packet
A tuple of decrypted packet (scapy.IPv6 or scapy.IP) and EspHdr
esp_hdr, esp_data = cstruct.Read(bytes(packet.payload), xfrm.EspHdr)
# Parse and strip ESP trailer.
pad_len, esp_nexthdr = struct.unpack("BB", esp_data[-2:])
trailer_len = pad_len + 2 # Add the size of the pad_len and next_hdr fields.
LayerType = {
IPPROTO_IPV6: scapy.IPv6,
IPPROTO_UDP: scapy.UDP}[esp_nexthdr]
next_layer = LayerType(esp_data[:-trailer_len])
if esp_nexthdr in [IPPROTO_IPIP, IPPROTO_IPV6]:
# Tunnel mode decap is simple. Return the inner packet.
return next_layer, esp_hdr
# Cut out the ESP header.
packet.payload = next_layer
# Fix the IPv4/IPv6 headers.
if type(packet) is scapy.IPv6:
packet.nh = IPPROTO_UDP
packet.plen = None # Recompute packet length.
packet = scapy.IPv6(bytes(packet))
elif type(packet) is scapy.IP:
packet.proto = IPPROTO_UDP
packet.len = None # Recompute packet length.
packet.chksum = None # Recompute IPv4 checksum.
packet = scapy.IP(bytes(packet))
raise ValueError("First layer in packet should be IPv4 or IPv6: " + repr(packet))
return packet, esp_hdr
class XfrmBaseTest(multinetwork_base.MultiNetworkBaseTest):
"""Base test class for all XFRM-related testing."""
def _isIcmpv6(self, payload):
if not isinstance(payload, scapy.IPv6):
return False
if payload.nh == IPPROTO_ICMPV6:
return True
return payload.nh == IPPROTO_HOPOPTS and payload.payload.nh == IPPROTO_ICMPV6
def _ExpectEspPacketOn(self, netid, spi, seq, length, src_addr, dst_addr):
"""Read a packet from a netid and verify its properties.
netid: netid from which to read an ESP packet
spi: SPI of the ESP packet in host byte order
seq: sequence number of the ESP packet
length: length of the packet's ESP payload or None to skip this check
src_addr: source address of the packet or None to skip this check
dst_addr: destination address of the packet or None to skip this check
scapy.IP/IPv6: the read packet
packets = []
for packet in self.ReadAllPacketsOn(netid):
if not self._isIcmpv6(packet):
self.assertEqual(1, len(packets))
packet = packets[0]
if length is not None:
self.assertEqual(length, len(packet.payload))
if dst_addr is not None:
self.assertEqual(dst_addr, packet.dst)
if src_addr is not None:
self.assertEqual(src_addr, packet.src)
# extract the ESP header
esp_hdr, _ = cstruct.Read(bytes(packet.payload), xfrm.EspHdr)
self.assertEqual(xfrm.EspHdr((spi, seq)), esp_hdr)
return packet
# TODO: delete this when we're more diligent about deleting our SAs.
class XfrmLazyTest(XfrmBaseTest):
"""Base test class Xfrm tests that cleans XFRM state on teardown."""
def setUp(self):
super(XfrmBaseTest, self).setUp()
self.xfrm = xfrm.Xfrm()
def tearDown(self):
super(XfrmBaseTest, self).tearDown()