client/deps/lansim/src/py/tuntap.py - platform/external/autotest - Git at Google

 # Copyright (c) 2013 The Chromium OS Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.

 import fcntl
 import os
 import struct
 import socket

 from lansim import pyiftun
 from lansim import tools


 # Export some constants used by callers to pass to the |mode| argument while
 # create a TunTap() object.
 from lansim.pyiftun import IFF_TAP, IFF_TUN


 class TunTapError(Exception):
     """TunTap specific error."""


 ETHERNET_HEADER_SIZE = 18


 STRUCT_IFREQ_FMT = {
     "ifr_flags": "h", # short ifr_flags
     "ifr_mtu": "i", # int ifr_flags
     "ifr_addr": "HH12s", # struct sockaddr_in ifr_addr
     "ifr_hwaddr": "H14s", # struct sockaddr ifru_hwaddr
 }


 IFNAMSIZ_FMT = str(pyiftun.IFNAMSIZ) + 's'


 def pack_struct_ifreq(if_name, argname, *args):
     """Packs a binary string representing a struct ifreq.

     The struct ifreq is used to call ioctl() on network devices. The argument
     type and size depends on the operation performed and is represented as the
     union of different types. This function packs the struct according to the
     provided |argname| which defines the type of |arg|. See netdevice(7) for a
     list of possible arguments and ioctl() commands.

     @param if_name: The interface name.
     @param argname: The name of the member used for the union in struct ifreq.
     @param args: The values used to pack the requested |argname|.
     @raises ValueError: if |argname| isn't a supported union's member name.
     """
     if argname not in STRUCT_IFREQ_FMT:
       raise ValueError()
     return struct.pack(IFNAMSIZ_FMT + STRUCT_IFREQ_FMT[argname], if_name, *args)


 def unpack_struct_ifreq(data, argname):
     """Returns a tuple with the interpreted contents of the a struct ifreq.

     The result returned from a ioctl() on network devices has the same format
     than the passed struct ifreq request. This function decodes this result into
     a python tuple, which depends on the |argname| passed to

     @param data: The packed representation of the struct ifreq.
     @param argname: The name of the member used for the union in struct ifreq.
     @raises ValueError: if |argname| isn't a supported union's member name.
     """
     if argname not in STRUCT_IFREQ_FMT:
       raise ValueError()
     return struct.unpack(IFNAMSIZ_FMT + STRUCT_IFREQ_FMT[argname], data)


 class TunTap(object):
     """TUN/TAP network interface manipulation class."""


     DEFAULT_DEV_NAME = {
         IFF_TUN: "tun%d",
         IFF_TAP: "tap%d",
     }


     def __init__(self, mode=pyiftun.IFF_TUN, name=None, tundev='/dev/net/tun'):
         """Creates or re-opens a TUN/TAP interface.

         @param mode: This argument is passed to the TUNSETIFF ioctl() to create
         the interface. It says whether the interface created is a TAP (IFF_TAP)
         or TUN (IFF_TUN) interface and some related constant flags found on
         pyiftun.IFF_*.
         @param name: The name of the created interface. If the name ends in '%d'
         that value will be replaced by the kernel with a given number, otherwise
         the name will be appended with '%d'.
         @param tundev: The path to the kerner interface to the tun driver which
         defaults to the standard '/dev/net/tun' if not specified.
         """
         tun_type = mode & pyiftun.TUN_TYPE_MASK
         if tun_type not in self.DEFAULT_DEV_NAME:
             raise TunTapError("mode (%r) not supported" % mode)

         self.mode = mode

         # The interface name can have a "%d" that the kernel will replace with
         # a number.
         if name is None:
             name = self.DEFAULT_DEV_NAME[tun_type]
         elif not name.endswith('%d'):
             name += "%d"

         # Create the TUN/TAP interface.
         fd = os.open(tundev, os.O_RDWR)
         self._fd = fd

         ifs = fcntl.ioctl(fd, pyiftun.TUNSETIFF,
             pack_struct_ifreq(name, 'ifr_flags', mode))
         ifs_name, ifs_mode = struct.unpack(IFNAMSIZ_FMT + "H", ifs)
         self.name = ifs_name.rstrip('\0')

         # Socket used for ioctl() operations over the network device.
         self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

         self.mtu = self._get_mtu()


     def __del__(self):
         if hasattr(self, '_fd'):
             os.close(self._fd)


     def _get_mtu(self):
         ifs = fcntl.ioctl(self._sock, pyiftun.SIOCGIFMTU,
             pack_struct_ifreq(self.name, 'ifr_mtu', 0))
         ifr_name, ifr_mtu = unpack_struct_ifreq(ifs, 'ifr_mtu')
         return ifr_mtu


     def _get_flags(self):
         ifs = fcntl.ioctl(self._sock, pyiftun.SIOCGIFFLAGS,
             pack_struct_ifreq(self.name, 'ifr_flags', 0))
         ifr_name, ifr_flags = unpack_struct_ifreq(ifs, 'ifr_flags')
         return ifr_flags


     def _set_flags(self, flags):
         ifs = fcntl.ioctl(self._sock, pyiftun.SIOCSIFFLAGS,
             pack_struct_ifreq(self.name, 'ifr_flags', flags))
         ifr_name, ifr_flags = unpack_struct_ifreq(ifs, 'ifr_flags')
         return ifr_flags


     def get_addr(self):
         """Return the address of the interface.

         @param string addr: The IPv4 address for the interface.
         """
         ifs = fcntl.ioctl(self._sock, pyiftun.SIOCGIFADDR,
             pack_struct_ifreq(self.name, 'ifr_addr', socket.AF_INET, 0, ''))
         ifr_name, ifr_family, ifr_type, ifr_addr = unpack_struct_ifreq(
                 ifs, 'ifr_addr')
         if ifr_type != 0:
             return None
         # ifr_addr contains up to 12 bytes (see STRUCT_IFREQ_FMT).
         return socket.inet_ntoa(ifr_addr[:4])


     def set_addr(self, addr, mask=None):
         """Sets the address and network mask of the interface.

         @param string addr: The IPv4 address for the interface.
         """
         str_addr = socket.inet_aton(addr)
         ifs = fcntl.ioctl(self._sock, pyiftun.SIOCSIFADDR,
             pack_struct_ifreq(self.name, 'ifr_addr',
                 socket.AF_INET, 0, str_addr))

         if mask != None:
           net_mask = (1 << 32) - (1 << (32 - mask))
           str_mask = struct.pack('!I', net_mask)
           ifs = fcntl.ioctl(self._sock, pyiftun.SIOCSIFNETMASK,
               pack_struct_ifreq(self.name, 'ifr_addr',
                   socket.AF_INET, 0, str_mask))


     """The interface IPv4 address in plain text as in '192.168.0.1'."""
     addr = property(get_addr, set_addr)


     def get_hwaddr(self):
         ifs = fcntl.ioctl(self._sock, pyiftun.SIOCGIFHWADDR,
             pack_struct_ifreq(self.name, 'ifr_hwaddr', 0, ''))
         ifr_name, ifr_family, ifr_hwaddr = unpack_struct_ifreq(
             ifs, 'ifr_hwaddr')
         return (ifr_family, tools.inet_ntohw(ifr_hwaddr[:6]))


     def set_hwaddr(self, hwaddr):
         """Sets the hardware ethernet address of the interface.

         The interface needs to be down in order to set this hardware (MAC)
         address.

         @param string hwaddr: The address in hex format: 'aa:bb:cc:DD:EE:FF'.
         """
         ifs = fcntl.ioctl(self._sock, pyiftun.SIOCSIFHWADDR,
             pack_struct_ifreq(self.name, 'ifr_hwaddr', 1, # 1 for Ethernet
                               tools.inet_hwton(hwaddr)))
         ifr_name, ifr_family, ifr_hwaddr = unpack_struct_ifreq(
             ifs, 'ifr_hwaddr')
         return (ifr_family, tools.inet_ntohw(ifr_hwaddr[:6]))


     """The interface Ethernet address as in '00:11:22:AA:BB:CC'."""
     hwaddr = property(get_hwaddr, set_hwaddr)


     def up(self):
         """Brings up the interface."""
         self._set_flags(self._get_flags() | pyiftun.IFF_UP)


     def down(self):
         """Brings down the interface."""
         self._set_flags(self._get_flags() & ~pyiftun.IFF_UP)


     def is_up(self):
         """Returns whether the interface is up."""
         return (self._get_flags() & pyiftun.IFF_UP) != 0


     def read(self):
         """Reads a 'sent' frame from the interface.

         The frame format depends on the interface type: Ethernet frame for TAP
         interfaces and IP frame for TUN interfaces. This function blocks until
         a new frame is available.

         @return string: A single frame sent to the interface.
         """
         return os.read(self._fd, self.mtu + ETHERNET_HEADER_SIZE)


     def write(self, data):
         """Write a 'received' frame from the interface.

         The frame format depends on the interface type: Ethernet frame for TAP
         interfaces and IP frame for TUN interfaces. This function does not
         block.

         @param data: A single frame received from the interface.
         """
         os.write(self._fd, data)


     def fileno(self):
         """Returns a file descriptor suitable to be used with select()."""
         return self._fd
	# Copyright (c) 2013 The Chromium OS Authors. All rights reserved.
	# Use of this source code is governed by a BSD-style license that can be
	# found in the LICENSE file.

	import fcntl
	import os
	import struct
	import socket

	from lansim import pyiftun
	from lansim import tools


	# Export some constants used by callers to pass to the \|mode\| argument while
	# create a TunTap() object.
	from lansim.pyiftun import IFF_TAP, IFF_TUN


	class TunTapError(Exception):
	"""TunTap specific error."""


	ETHERNET_HEADER_SIZE = 18


	STRUCT_IFREQ_FMT = {
	"ifr_flags": "h", # short ifr_flags
	"ifr_mtu": "i", # int ifr_flags
	"ifr_addr": "HH12s", # struct sockaddr_in ifr_addr
	"ifr_hwaddr": "H14s", # struct sockaddr ifru_hwaddr
	}


	IFNAMSIZ_FMT = str(pyiftun.IFNAMSIZ) + 's'


	def pack_struct_ifreq(if_name, argname, *args):
	"""Packs a binary string representing a struct ifreq.

	The struct ifreq is used to call ioctl() on network devices. The argument
	type and size depends on the operation performed and is represented as the
	union of different types. This function packs the struct according to the
	provided \|argname\| which defines the type of \|arg\|. See netdevice(7) for a
	list of possible arguments and ioctl() commands.

	@param if_name: The interface name.
	@param argname: The name of the member used for the union in struct ifreq.
	@param args: The values used to pack the requested \|argname\|.
	@raises ValueError: if \|argname\| isn't a supported union's member name.
	"""
	if argname not in STRUCT_IFREQ_FMT:
	raise ValueError()
	return struct.pack(IFNAMSIZ_FMT + STRUCT_IFREQ_FMT[argname], if_name, *args)


	def unpack_struct_ifreq(data, argname):
	"""Returns a tuple with the interpreted contents of the a struct ifreq.

	The result returned from a ioctl() on network devices has the same format
	than the passed struct ifreq request. This function decodes this result into
	a python tuple, which depends on the \|argname\| passed to

	@param data: The packed representation of the struct ifreq.
	@param argname: The name of the member used for the union in struct ifreq.
	@raises ValueError: if \|argname\| isn't a supported union's member name.
	"""
	if argname not in STRUCT_IFREQ_FMT:
	raise ValueError()
	return struct.unpack(IFNAMSIZ_FMT + STRUCT_IFREQ_FMT[argname], data)


	class TunTap(object):
	"""TUN/TAP network interface manipulation class."""


	DEFAULT_DEV_NAME = {
	IFF_TUN: "tun%d",
	IFF_TAP: "tap%d",
	}


	def __init__(self, mode=pyiftun.IFF_TUN, name=None, tundev='/dev/net/tun'):
	"""Creates or re-opens a TUN/TAP interface.

	@param mode: This argument is passed to the TUNSETIFF ioctl() to create
	the interface. It says whether the interface created is a TAP (IFF_TAP)
	or TUN (IFF_TUN) interface and some related constant flags found on
	pyiftun.IFF_*.
	@param name: The name of the created interface. If the name ends in '%d'
	that value will be replaced by the kernel with a given number, otherwise
	the name will be appended with '%d'.
	@param tundev: The path to the kerner interface to the tun driver which
	defaults to the standard '/dev/net/tun' if not specified.
	"""
	tun_type = mode & pyiftun.TUN_TYPE_MASK
	if tun_type not in self.DEFAULT_DEV_NAME:
	raise TunTapError("mode (%r) not supported" % mode)

	self.mode = mode

	# The interface name can have a "%d" that the kernel will replace with
	# a number.
	if name is None:
	name = self.DEFAULT_DEV_NAME[tun_type]
	elif not name.endswith('%d'):
	name += "%d"

	# Create the TUN/TAP interface.
	fd = os.open(tundev, os.O_RDWR)
	self._fd = fd

	ifs = fcntl.ioctl(fd, pyiftun.TUNSETIFF,
	pack_struct_ifreq(name, 'ifr_flags', mode))
	ifs_name, ifs_mode = struct.unpack(IFNAMSIZ_FMT + "H", ifs)
	self.name = ifs_name.rstrip('\0')

	# Socket used for ioctl() operations over the network device.
	self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

	self.mtu = self._get_mtu()


	def __del__(self):
	if hasattr(self, '_fd'):
	os.close(self._fd)


	def _get_mtu(self):
	ifs = fcntl.ioctl(self._sock, pyiftun.SIOCGIFMTU,
	pack_struct_ifreq(self.name, 'ifr_mtu', 0))
	ifr_name, ifr_mtu = unpack_struct_ifreq(ifs, 'ifr_mtu')
	return ifr_mtu


	def _get_flags(self):
	ifs = fcntl.ioctl(self._sock, pyiftun.SIOCGIFFLAGS,
	pack_struct_ifreq(self.name, 'ifr_flags', 0))
	ifr_name, ifr_flags = unpack_struct_ifreq(ifs, 'ifr_flags')
	return ifr_flags


	def _set_flags(self, flags):
	ifs = fcntl.ioctl(self._sock, pyiftun.SIOCSIFFLAGS,
	pack_struct_ifreq(self.name, 'ifr_flags', flags))
	ifr_name, ifr_flags = unpack_struct_ifreq(ifs, 'ifr_flags')
	return ifr_flags


	def get_addr(self):
	"""Return the address of the interface.

	@param string addr: The IPv4 address for the interface.
	"""
	ifs = fcntl.ioctl(self._sock, pyiftun.SIOCGIFADDR,
	pack_struct_ifreq(self.name, 'ifr_addr', socket.AF_INET, 0, ''))
	ifr_name, ifr_family, ifr_type, ifr_addr = unpack_struct_ifreq(
	ifs, 'ifr_addr')
	if ifr_type != 0:
	return None
	# ifr_addr contains up to 12 bytes (see STRUCT_IFREQ_FMT).
	return socket.inet_ntoa(ifr_addr[:4])


	def set_addr(self, addr, mask=None):
	"""Sets the address and network mask of the interface.

	@param string addr: The IPv4 address for the interface.
	"""
	str_addr = socket.inet_aton(addr)
	ifs = fcntl.ioctl(self._sock, pyiftun.SIOCSIFADDR,
	pack_struct_ifreq(self.name, 'ifr_addr',
	socket.AF_INET, 0, str_addr))

	if mask != None:
	net_mask = (1 << 32) - (1 << (32 - mask))
	str_mask = struct.pack('!I', net_mask)
	ifs = fcntl.ioctl(self._sock, pyiftun.SIOCSIFNETMASK,
	pack_struct_ifreq(self.name, 'ifr_addr',
	socket.AF_INET, 0, str_mask))


	"""The interface IPv4 address in plain text as in '192.168.0.1'."""
	addr = property(get_addr, set_addr)


	def get_hwaddr(self):
	ifs = fcntl.ioctl(self._sock, pyiftun.SIOCGIFHWADDR,
	pack_struct_ifreq(self.name, 'ifr_hwaddr', 0, ''))
	ifr_name, ifr_family, ifr_hwaddr = unpack_struct_ifreq(
	ifs, 'ifr_hwaddr')
	return (ifr_family, tools.inet_ntohw(ifr_hwaddr[:6]))


	def set_hwaddr(self, hwaddr):
	"""Sets the hardware ethernet address of the interface.

	The interface needs to be down in order to set this hardware (MAC)
	address.

	@param string hwaddr: The address in hex format: 'aa:bb:cc:DD:EE:FF'.
	"""
	ifs = fcntl.ioctl(self._sock, pyiftun.SIOCSIFHWADDR,
	pack_struct_ifreq(self.name, 'ifr_hwaddr', 1, # 1 for Ethernet
	tools.inet_hwton(hwaddr)))
	ifr_name, ifr_family, ifr_hwaddr = unpack_struct_ifreq(
	ifs, 'ifr_hwaddr')
	return (ifr_family, tools.inet_ntohw(ifr_hwaddr[:6]))


	"""The interface Ethernet address as in '00:11:22:AA:BB:CC'."""
	hwaddr = property(get_hwaddr, set_hwaddr)


	def up(self):
	"""Brings up the interface."""
	self._set_flags(self._get_flags() \| pyiftun.IFF_UP)


	def down(self):
	"""Brings down the interface."""
	self._set_flags(self._get_flags() & ~pyiftun.IFF_UP)


	def is_up(self):
	"""Returns whether the interface is up."""
	return (self._get_flags() & pyiftun.IFF_UP) != 0


	def read(self):
	"""Reads a 'sent' frame from the interface.

	The frame format depends on the interface type: Ethernet frame for TAP
	interfaces and IP frame for TUN interfaces. This function blocks until
	a new frame is available.

	@return string: A single frame sent to the interface.
	"""
	return os.read(self._fd, self.mtu + ETHERNET_HEADER_SIZE)


	def write(self, data):
	"""Write a 'received' frame from the interface.

	The frame format depends on the interface type: Ethernet frame for TAP
	interfaces and IP frame for TUN interfaces. This function does not
	block.

	@param data: A single frame received from the interface.
	"""
	os.write(self._fd, data)


	def fileno(self):
	"""Returns a file descriptor suitable to be used with select()."""
	return self._fd