``'can0'``. The network interface name ``''`` can be used to receive packets
from all network interfaces of this family.
+ - :const:`CAN_ISOTP` protocol require a tuple ``(interface, rx_addr, tx_addr)``
+ where both additional parameters are unsigned long integer that represent a
+ CAN identifier (standard or extended).
+
- A string or a tuple ``(id, unit)`` is used for the :const:`SYSPROTO_CONTROL`
protocol of the :const:`PF_SYSTEM` family. The string is the name of a
kernel control using a dynamically-assigned ID. The tuple can be used if ID
.. versionadded:: 3.5
+.. data:: CAN_ISOTP
+
+ CAN_ISOTP, in the CAN protocol family, is the ISO-TP (ISO 15765-2) protocol.
+ ISO-TP constants, documented in the Linux documentation.
+
+ Availability: Linux >= 2.6.25
+
+ .. versionadded:: 3.7
+
+
.. data:: AF_RDS
PF_RDS
SOL_RDS
:const:`SOCK_DGRAM`, :const:`SOCK_RAW` or perhaps one of the other ``SOCK_``
constants. The protocol number is usually zero and may be omitted or in the
case where the address family is :const:`AF_CAN` the protocol should be one
- of :const:`CAN_RAW` or :const:`CAN_BCM`. If *fileno* is specified, the other
+ of :const:`CAN_RAW`, :const:`CAN_BCM` or :const:`CAN_ISOTP`. If *fileno* is specified, the other
arguments are ignored, causing the socket with the specified file descriptor
to return. Unlike :func:`socket.fromfd`, *fileno* will return the same
socket and not a duplicate. This may help close a detached socket using
.. versionchanged:: 3.4
The returned socket is now non-inheritable.
+ .. versionchanged:: 3.7
+ The CAN_ISOTP protocol was added.
.. function:: socketpair([family[, type[, proto]]])
# disabled promiscuous mode
s.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)
-The last example shows how to use the socket interface to communicate to a CAN
+The next example shows how to use the socket interface to communicate to a CAN
network using the raw socket protocol. To use CAN with the broadcast
manager protocol instead, open a socket with::
can use the :meth:`socket.send`, and the :meth:`socket.recv` operations (and
their counterparts) on the socket object as usual.
-This example might require special privileges::
+This last example might require special privileges::
import socket
import struct
s.close()
return True
+def _have_socket_can_isotp():
+ """Check whether CAN ISOTP sockets are supported on this host."""
+ try:
+ s = socket.socket(socket.PF_CAN, socket.SOCK_DGRAM, socket.CAN_ISOTP)
+ except (AttributeError, OSError):
+ return False
+ else:
+ s.close()
+ return True
+
def _have_socket_rds():
"""Check whether RDS sockets are supported on this host."""
try:
HAVE_SOCKET_CAN = _have_socket_can()
+HAVE_SOCKET_CAN_ISOTP = _have_socket_can_isotp()
+
HAVE_SOCKET_RDS = _have_socket_rds()
HAVE_SOCKET_ALG = _have_socket_alg()
self.assertEqual(bytes_sent, len(header_plus_frame))
+@unittest.skipUnless(HAVE_SOCKET_CAN_ISOTP, 'CAN ISOTP required for this test.')
+class ISOTPTest(unittest.TestCase):
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.interface = "vcan0"
+
+ def testCrucialConstants(self):
+ socket.AF_CAN
+ socket.PF_CAN
+ socket.CAN_ISOTP
+ socket.SOCK_DGRAM
+
+ def testCreateSocket(self):
+ with socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW) as s:
+ pass
+
+ @unittest.skipUnless(hasattr(socket, "CAN_ISOTP"),
+ 'socket.CAN_ISOTP required for this test.')
+ def testCreateISOTPSocket(self):
+ with socket.socket(socket.PF_CAN, socket.SOCK_DGRAM, socket.CAN_ISOTP) as s:
+ pass
+
+ def testTooLongInterfaceName(self):
+ # most systems limit IFNAMSIZ to 16, take 1024 to be sure
+ with socket.socket(socket.PF_CAN, socket.SOCK_DGRAM, socket.CAN_ISOTP) as s:
+ with self.assertRaisesRegex(OSError, 'interface name too long'):
+ s.bind(('x' * 1024, 1, 2))
+
+ def testBind(self):
+ try:
+ with socket.socket(socket.PF_CAN, socket.SOCK_DGRAM, socket.CAN_ISOTP) as s:
+ addr = self.interface, 0x123, 0x456
+ s.bind(addr)
+ self.assertEqual(s.getsockname(), addr)
+ except OSError as e:
+ if e.errno == errno.ENODEV:
+ self.skipTest('network interface `%s` does not exist' %
+ self.interface)
+ else:
+ raise
+
+
@unittest.skipUnless(HAVE_SOCKET_RDS, 'RDS sockets required for this test.')
class BasicRDSTest(unittest.TestCase):
Kostyantyn Leschenko
Benno Leslie
Christopher Tur Lesniewski-Laas
+Pier-Yves Lessard
Alain Leufroy
Mark Levinson
Mark Levitt
--- /dev/null
+Added support for CAN ISO-TP protocol in the socket module.
ifname = ifr.ifr_name;
}
- return Py_BuildValue("O&h", PyUnicode_DecodeFSDefault,
- ifname,
- a->can_family);
+ switch (proto) {
+#ifdef CAN_ISOTP
+ case CAN_ISOTP:
+ {
+ return Py_BuildValue("O&kk", PyUnicode_DecodeFSDefault,
+ ifname,
+ a->can_addr.tp.rx_id,
+ a->can_addr.tp.tx_id);
+ }
+#endif
+ default:
+ {
+ return Py_BuildValue("O&", PyUnicode_DecodeFSDefault,
+ ifname);
+ }
+ }
}
#endif
}
#endif
-#if defined(AF_CAN) && defined(CAN_RAW) && defined(CAN_BCM)
+#ifdef AF_CAN
+
+#if defined(CAN_RAW) && defined(CAN_BCM)
case AF_CAN:
switch (s->sock_proto) {
case CAN_RAW:
PyObject *interfaceName;
struct ifreq ifr;
Py_ssize_t len;
-
addr = (struct sockaddr_can *)addr_ret;
if (!PyArg_ParseTuple(args, "O&", PyUnicode_FSConverter,
Py_DECREF(interfaceName);
return 1;
}
+#endif
+
+#ifdef CAN_ISOTP
+ case CAN_ISOTP:
+ {
+ struct sockaddr_can *addr;
+ PyObject *interfaceName;
+ struct ifreq ifr;
+ Py_ssize_t len;
+ unsigned long int rx_id, tx_id;
+
+ addr = (struct sockaddr_can *)addr_ret;
+
+ if (!PyArg_ParseTuple(args, "O&kk", PyUnicode_FSConverter,
+ &interfaceName,
+ &rx_id,
+ &tx_id))
+ return 0;
+
+ len = PyBytes_GET_SIZE(interfaceName);
+
+ if (len == 0) {
+ ifr.ifr_ifindex = 0;
+ } else if ((size_t)len < sizeof(ifr.ifr_name)) {
+ strncpy(ifr.ifr_name, PyBytes_AS_STRING(interfaceName), sizeof(ifr.ifr_name));
+ ifr.ifr_name[(sizeof(ifr.ifr_name))-1] = '\0';
+ if (ioctl(s->sock_fd, SIOCGIFINDEX, &ifr) < 0) {
+ s->errorhandler();
+ Py_DECREF(interfaceName);
+ return 0;
+ }
+ } else {
+ PyErr_SetString(PyExc_OSError,
+ "AF_CAN interface name too long");
+ Py_DECREF(interfaceName);
+ return 0;
+ }
+
+ addr->can_family = AF_CAN;
+ addr->can_ifindex = ifr.ifr_ifindex;
+ addr->can_addr.tp.rx_id = rx_id;
+ addr->can_addr.tp.tx_id = tx_id;
+
+ *len_ret = sizeof(*addr);
+ Py_DECREF(interfaceName);
+ return 1;
+ }
+#endif
default:
PyErr_SetString(PyExc_OSError,
"getsockaddrarg: unsupported CAN protocol");
PyModule_AddIntMacro(m, CAN_SFF_MASK);
PyModule_AddIntMacro(m, CAN_EFF_MASK);
PyModule_AddIntMacro(m, CAN_ERR_MASK);
+#ifdef CAN_ISOTP
+ PyModule_AddIntMacro(m, CAN_ISOTP);
+#endif
#endif
#ifdef HAVE_LINUX_CAN_RAW_H
PyModule_AddIntMacro(m, CAN_RAW_FILTER);
projects / python / commitdiff