我们从Python开源项目中，提取了以下41个代码示例，用于说明如何使用socket.SCM_RIGHTS。
def flush(self): """Send buffered requests to the display server. Will send as many requests as possible to the display server. Will not block; if sendmsg() would block, will leave events in the queue. Returns True if the queue was emptied. """ while self._send_queue: b, fds = self._send_queue.pop(0) try: self._f.sendmsg([b], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", fds))]) for fd in fds: os.close(fd) except socket.error as e: if socket.errno == 11: # Would block. Return the data to the head of the queue # and try again later! self.log.debug("flush would block; returning data to queue") self._send_queue.insert(0, (b, fds)) return raise return True
def testFDPassPartialIntInMiddle(self): # Try to pass two FD arrays, the first of which is truncated. msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock, len(MSG), 10240) self.assertEqual(msg, MSG) self.checkRecvmsgAddress(addr, self.cli_addr) self.checkFlags(flags, eor=True, ignore=socket.MSG_CTRUNC) self.assertLessEqual(len(ancdata), 2) fds = array.array("i") # Arrays may have been combined in a single control message for cmsg_level, cmsg_type, cmsg_data in ancdata: self.assertEqual(cmsg_level, socket.SOL_SOCKET) self.assertEqual(cmsg_type, socket.SCM_RIGHTS) fds.frombytes(cmsg_data[: len(cmsg_data) - (len(cmsg_data) % fds.itemsize)]) self.assertLessEqual(len(fds), 2) self.checkFDs(fds)
def recvfds(sock, size): '''Receive an array of fds over an AF_UNIX socket.''' a = array.array('i') bytes_size = a.itemsize * size msg, ancdata, flags, addr = sock.recvmsg(1, socket.CMSG_LEN(bytes_size)) if not msg and not ancdata: raise EOFError try: if ACKNOWLEDGE: sock.send(b'A') if len(ancdata) != 1: raise RuntimeError('received %d items of ancdata' % len(ancdata)) cmsg_level, cmsg_type, cmsg_data = ancdata[0] if (cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS): if len(cmsg_data) % a.itemsize != 0: raise ValueError a.frombytes(cmsg_data) assert len(a) % 256 == msg[0] return list(a) except (ValueError, IndexError): pass raise RuntimeError('Invalid data received')
def checkTruncatedArray(self, ancbuf, maxdata, mindata=0): # Check that file descriptor data is truncated to between # mindata and maxdata bytes when received with buffer size # ancbuf, and that any complete file descriptor numbers are # valid. msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock, len(MSG), ancbuf) self.assertEqual(msg, MSG) self.checkRecvmsgAddress(addr, self.cli_addr) self.checkFlags(flags, eor=True, checkset=socket.MSG_CTRUNC) if mindata == 0 and ancdata == []: return self.assertEqual(len(ancdata), 1) cmsg_level, cmsg_type, cmsg_data = ancdata[0] self.assertEqual(cmsg_level, socket.SOL_SOCKET) self.assertEqual(cmsg_type, socket.SCM_RIGHTS) self.assertGreaterEqual(len(cmsg_data), mindata) self.assertLessEqual(len(cmsg_data), maxdata) fds = array.array("i") fds.frombytes(cmsg_data[: len(cmsg_data) - (len(cmsg_data) % fds.itemsize)]) self.checkFDs(fds)
def recv_fd(sock): ''' Receive a single file descriptor ''' msg, ancdata, flags, addr = sock.recvmsg(1, socket.CMSG_LEN(struct.calcsize('i'))) cmsg_level, cmsg_type, cmsg_data = ancdata[0] assert cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS sock.sendall(b'OK') return struct.unpack('i', cmsg_data)[0]
def send_fd(sock, fd): ''' Send a single file descriptor. ''' sock.sendmsg([b'x'], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, struct.pack('i', fd))]) ack = sock.recv(2) assert ack == b'OK'
def recv(self): """Receive as much data as is available. Returns True if any data was received. Will not block. """ data = None try: fds = array.array("i") data, ancdata, msg_flags, address = self._f.recvmsg( 1024, socket.CMSG_SPACE(16 * fds.itemsize)) for cmsg_level, cmsg_type, cmsg_data in ancdata: if (cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS): fds.fromstring(cmsg_data[ :len(cmsg_data) - (len(cmsg_data) % fds.itemsize)]) self._incoming_fds.extend(fds) if data: self._decode(data) return True else: raise ServerDisconnected() except socket.error as e: if e.errno == 11: # No data available; would otherwise block return raise
def closeRecvmsgFDs(self, recvmsg_result): # Close all file descriptors specified in the ancillary data # of the given return value from recvmsg() or recvmsg_into(). for cmsg_level, cmsg_type, cmsg_data in recvmsg_result[1]: if (cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS): fds = array.array("i") fds.frombytes(cmsg_data[: len(cmsg_data) - (len(cmsg_data) % fds.itemsize)]) for fd in fds: os.close(fd)
def createAndSendFDs(self, n): # Send n new file descriptors created by newFDs() to the # server, with the constant MSG as the non-ancillary data. self.assertEqual( self.sendmsgToServer([MSG], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", self.newFDs(n)))]), len(MSG))
def checkRecvmsgFDs(self, numfds, result, maxcmsgs=1, ignoreflags=0): # Check that constant MSG was received with numfds file # descriptors in a maximum of maxcmsgs control messages (which # must contain only complete integers). By default, check # that MSG_CTRUNC is unset, but ignore any flags in # ignoreflags. msg, ancdata, flags, addr = result self.assertEqual(msg, MSG) self.checkRecvmsgAddress(addr, self.cli_addr) self.checkFlags(flags, eor=True, checkunset=socket.MSG_CTRUNC, ignore=ignoreflags) self.assertIsInstance(ancdata, list) self.assertLessEqual(len(ancdata), maxcmsgs) fds = array.array("i") for item in ancdata: self.assertIsInstance(item, tuple) cmsg_level, cmsg_type, cmsg_data = item self.assertEqual(cmsg_level, socket.SOL_SOCKET) self.assertEqual(cmsg_type, socket.SCM_RIGHTS) self.assertIsInstance(cmsg_data, bytes) self.assertEqual(len(cmsg_data) % SIZEOF_INT, 0) fds.frombytes(cmsg_data) self.assertEqual(len(fds), numfds) self.checkFDs(fds)
def _testFDPassSimple(self): self.assertEqual( self.sendmsgToServer( [MSG], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", self.newFDs(1)).tobytes())]), len(MSG))
def _testFDPassSeparate(self): fd0, fd1 = self.newFDs(2) self.assertEqual( self.sendmsgToServer([MSG], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", [fd0])), (socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", [fd1]))]), len(MSG))
def _testFDPassEmpty(self): self.sendAncillaryIfPossible(MSG, [(socket.SOL_SOCKET, socket.SCM_RIGHTS, b"")])
def testFDPassPartialInt(self): # Try to pass a truncated FD array. msg, ancdata, flags, addr = self.doRecvmsg(self.serv_sock, len(MSG), 10240) self.assertEqual(msg, MSG) self.checkRecvmsgAddress(addr, self.cli_addr) self.checkFlags(flags, eor=True, ignore=socket.MSG_CTRUNC) self.assertLessEqual(len(ancdata), 1) for cmsg_level, cmsg_type, cmsg_data in ancdata: self.assertEqual(cmsg_level, socket.SOL_SOCKET) self.assertEqual(cmsg_type, socket.SCM_RIGHTS) self.assertLess(len(cmsg_data), SIZEOF_INT)
def _testFDPassPartialInt(self): self.sendAncillaryIfPossible( MSG, [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", [self.badfd]).tobytes()[:-1])])
def _testFDPassPartialIntInMiddle(self): fd0, fd1 = self.newFDs(2) self.sendAncillaryIfPossible( MSG, [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", [fd0, self.badfd]).tobytes()[:-1]), (socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", [fd1]))])
def sendfds(sock, fds): '''Send an array of fds over an AF_UNIX socket.''' fds = array.array('i', fds) msg = bytes([len(fds) % 256]) sock.sendmsg([msg], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fds)]) if ACKNOWLEDGE and sock.recv(1) != b'A': raise RuntimeError('did not receive acknowledgement of fd')
def DupFd(fd): '''Return a wrapper for an fd.''' popen_obj = context.get_spawning_popen() if popen_obj is not None: return popen_obj.DupFd(popen_obj.duplicate_for_child(fd)) elif HAVE_SEND_HANDLE: from . import resource_sharer return resource_sharer.DupFd(fd) else: raise ValueError('SCM_RIGHTS appears not to be available') # # Try making some callable types picklable #
def _testFDPassSeparateMinSpace(self): fd0, fd1 = self.newFDs(2) self.assertEqual( self.sendmsgToServer([MSG], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", [fd0])), (socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", [fd1]))]), len(MSG))
def run_frontend(self, tcp_address, backlog=10, timeout=0.1): log.info('Running frontend loop') address, port = tcp_address.split(':') with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) sock.bind((address, int(port))) sock.listen(backlog) sock.settimeout(timeout) while not self.stop_event.is_set(): try: conn, address = await self.loop.sock_accept(sock) except socket.timeout: await asyncio.sleep(timeout) continue service = await self.loop.sock_recv(conn, BUFFER_SIZE) if not service: break if service not in self.services: self.loop.sock_sendall(conn, b'-Service not found\r\n') continue # detach and pack FD into a array fd = conn.detach() fds = array.array("I", [fd]) try: # Send FD to server connection server_conn = self.services[service] server_conn.sendmsg([b'1'], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fds)]) except BrokenPipeError: # NOQA # If connections is broken, the server is gone # so we need to remove it from services del self.services[service] conn = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM) conn.sendall(b'-Service not found\r\n') conn.close()
