tcp.py 42 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187
  1. # -*- test-case-name: twisted.test.test_tcp -*-
  2. # Copyright (c) Twisted Matrix Laboratories.
  3. # See LICENSE for details.
  4. """
  5. Various asynchronous TCP/IP classes.
  6. End users shouldn't use this module directly - use the reactor APIs instead.
  7. """
  8. from __future__ import division, absolute_import
  9. # System Imports
  10. import socket
  11. import sys
  12. import operator
  13. import struct
  14. from zope.interface import implementer
  15. from twisted.python.compat import _PY3, lazyByteSlice
  16. from twisted.python.runtime import platformType
  17. from twisted.python import versions, deprecate
  18. try:
  19. # Try to get the memory BIO based startTLS implementation, available since
  20. # pyOpenSSL 0.10
  21. from twisted.internet._newtls import (
  22. ConnectionMixin as _TLSConnectionMixin,
  23. ClientMixin as _TLSClientMixin,
  24. ServerMixin as _TLSServerMixin)
  25. except ImportError:
  26. # There is no version of startTLS available
  27. class _TLSConnectionMixin(object):
  28. TLS = False
  29. class _TLSClientMixin(object):
  30. pass
  31. class _TLSServerMixin(object):
  32. pass
  33. if platformType == 'win32':
  34. # no such thing as WSAEPERM or error code 10001 according to winsock.h or MSDN
  35. EPERM = object()
  36. from errno import WSAEINVAL as EINVAL
  37. from errno import WSAEWOULDBLOCK as EWOULDBLOCK
  38. from errno import WSAEINPROGRESS as EINPROGRESS
  39. from errno import WSAEALREADY as EALREADY
  40. from errno import WSAEISCONN as EISCONN
  41. from errno import WSAENOBUFS as ENOBUFS
  42. from errno import WSAEMFILE as EMFILE
  43. # No such thing as WSAENFILE, either.
  44. ENFILE = object()
  45. # Nor ENOMEM
  46. ENOMEM = object()
  47. EAGAIN = EWOULDBLOCK
  48. from errno import WSAECONNRESET as ECONNABORTED
  49. from twisted.python.win32 import formatError as strerror
  50. else:
  51. from errno import EPERM
  52. from errno import EINVAL
  53. from errno import EWOULDBLOCK
  54. from errno import EINPROGRESS
  55. from errno import EALREADY
  56. from errno import EISCONN
  57. from errno import ENOBUFS
  58. from errno import EMFILE
  59. from errno import ENFILE
  60. from errno import ENOMEM
  61. from errno import EAGAIN
  62. from errno import ECONNABORTED
  63. from os import strerror
  64. from errno import errorcode
  65. # Twisted Imports
  66. from twisted.internet import base, address, fdesc
  67. from twisted.internet.task import deferLater
  68. from twisted.python import log, failure, reflect
  69. from twisted.python.util import untilConcludes
  70. from twisted.internet.error import CannotListenError
  71. from twisted.internet import abstract, main, interfaces, error
  72. from twisted.internet.protocol import Protocol
  73. # Not all platforms have, or support, this flag.
  74. _AI_NUMERICSERV = getattr(socket, "AI_NUMERICSERV", 0)
  75. # The type for service names passed to socket.getservbyname:
  76. if _PY3:
  77. _portNameType = str
  78. else:
  79. _portNameType = (str, unicode)
  80. class _SocketCloser(object):
  81. """
  82. @ivar _shouldShutdown: Set to C{True} if C{shutdown} should be called
  83. before calling C{close} on the underlying socket.
  84. @type _shouldShutdown: C{bool}
  85. """
  86. _shouldShutdown = True
  87. def _closeSocket(self, orderly):
  88. # The call to shutdown() before close() isn't really necessary, because
  89. # we set FD_CLOEXEC now, which will ensure this is the only process
  90. # holding the FD, thus ensuring close() really will shutdown the TCP
  91. # socket. However, do it anyways, just to be safe.
  92. skt = self.socket
  93. try:
  94. if orderly:
  95. if self._shouldShutdown:
  96. skt.shutdown(2)
  97. else:
  98. # Set SO_LINGER to 1,0 which, by convention, causes a
  99. # connection reset to be sent when close is called,
  100. # instead of the standard FIN shutdown sequence.
  101. self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
  102. struct.pack("ii", 1, 0))
  103. except socket.error:
  104. pass
  105. try:
  106. skt.close()
  107. except socket.error:
  108. pass
  109. class _AbortingMixin(object):
  110. """
  111. Common implementation of C{abortConnection}.
  112. @ivar _aborting: Set to C{True} when C{abortConnection} is called.
  113. @type _aborting: C{bool}
  114. """
  115. _aborting = False
  116. def abortConnection(self):
  117. """
  118. Aborts the connection immediately, dropping any buffered data.
  119. @since: 11.1
  120. """
  121. if self.disconnected or self._aborting:
  122. return
  123. self._aborting = True
  124. self.stopReading()
  125. self.stopWriting()
  126. self.doRead = lambda *args, **kwargs: None
  127. self.doWrite = lambda *args, **kwargs: None
  128. self.reactor.callLater(0, self.connectionLost,
  129. failure.Failure(error.ConnectionAborted()))
  130. @implementer(interfaces.ITCPTransport, interfaces.ISystemHandle)
  131. class Connection(_TLSConnectionMixin, abstract.FileDescriptor, _SocketCloser,
  132. _AbortingMixin):
  133. """
  134. Superclass of all socket-based FileDescriptors.
  135. This is an abstract superclass of all objects which represent a TCP/IP
  136. connection based socket.
  137. @ivar logstr: prefix used when logging events related to this connection.
  138. @type logstr: C{str}
  139. """
  140. def __init__(self, skt, protocol, reactor=None):
  141. abstract.FileDescriptor.__init__(self, reactor=reactor)
  142. self.socket = skt
  143. self.socket.setblocking(0)
  144. self.fileno = skt.fileno
  145. self.protocol = protocol
  146. def getHandle(self):
  147. """Return the socket for this connection."""
  148. return self.socket
  149. def doRead(self):
  150. """Calls self.protocol.dataReceived with all available data.
  151. This reads up to self.bufferSize bytes of data from its socket, then
  152. calls self.dataReceived(data) to process it. If the connection is not
  153. lost through an error in the physical recv(), this function will return
  154. the result of the dataReceived call.
  155. """
  156. try:
  157. data = self.socket.recv(self.bufferSize)
  158. except socket.error as se:
  159. if se.args[0] == EWOULDBLOCK:
  160. return
  161. else:
  162. return main.CONNECTION_LOST
  163. return self._dataReceived(data)
  164. def _dataReceived(self, data):
  165. if not data:
  166. return main.CONNECTION_DONE
  167. rval = self.protocol.dataReceived(data)
  168. if rval is not None:
  169. offender = self.protocol.dataReceived
  170. warningFormat = (
  171. 'Returning a value other than None from %(fqpn)s is '
  172. 'deprecated since %(version)s.')
  173. warningString = deprecate.getDeprecationWarningString(
  174. offender, versions.Version('Twisted', 11, 0, 0),
  175. format=warningFormat)
  176. deprecate.warnAboutFunction(offender, warningString)
  177. return rval
  178. def writeSomeData(self, data):
  179. """
  180. Write as much as possible of the given data to this TCP connection.
  181. This sends up to C{self.SEND_LIMIT} bytes from C{data}. If the
  182. connection is lost, an exception is returned. Otherwise, the number
  183. of bytes successfully written is returned.
  184. """
  185. # Limit length of buffer to try to send, because some OSes are too
  186. # stupid to do so themselves (ahem windows)
  187. limitedData = lazyByteSlice(data, 0, self.SEND_LIMIT)
  188. try:
  189. return untilConcludes(self.socket.send, limitedData)
  190. except socket.error as se:
  191. if se.args[0] in (EWOULDBLOCK, ENOBUFS):
  192. return 0
  193. else:
  194. return main.CONNECTION_LOST
  195. def _closeWriteConnection(self):
  196. try:
  197. self.socket.shutdown(1)
  198. except socket.error:
  199. pass
  200. p = interfaces.IHalfCloseableProtocol(self.protocol, None)
  201. if p:
  202. try:
  203. p.writeConnectionLost()
  204. except:
  205. f = failure.Failure()
  206. log.err()
  207. self.connectionLost(f)
  208. def readConnectionLost(self, reason):
  209. p = interfaces.IHalfCloseableProtocol(self.protocol, None)
  210. if p:
  211. try:
  212. p.readConnectionLost()
  213. except:
  214. log.err()
  215. self.connectionLost(failure.Failure())
  216. else:
  217. self.connectionLost(reason)
  218. def connectionLost(self, reason):
  219. """See abstract.FileDescriptor.connectionLost().
  220. """
  221. # Make sure we're not called twice, which can happen e.g. if
  222. # abortConnection() is called from protocol's dataReceived and then
  223. # code immediately after throws an exception that reaches the
  224. # reactor. We can't rely on "disconnected" attribute for this check
  225. # since twisted.internet._oldtls does evil things to it:
  226. if not hasattr(self, "socket"):
  227. return
  228. abstract.FileDescriptor.connectionLost(self, reason)
  229. self._closeSocket(not reason.check(error.ConnectionAborted))
  230. protocol = self.protocol
  231. del self.protocol
  232. del self.socket
  233. del self.fileno
  234. protocol.connectionLost(reason)
  235. logstr = "Uninitialized"
  236. def logPrefix(self):
  237. """Return the prefix to log with when I own the logging thread.
  238. """
  239. return self.logstr
  240. def getTcpNoDelay(self):
  241. return operator.truth(self.socket.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY))
  242. def setTcpNoDelay(self, enabled):
  243. self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, enabled)
  244. def getTcpKeepAlive(self):
  245. return operator.truth(self.socket.getsockopt(socket.SOL_SOCKET,
  246. socket.SO_KEEPALIVE))
  247. def setTcpKeepAlive(self, enabled):
  248. self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, enabled)
  249. class _BaseBaseClient(object):
  250. """
  251. Code shared with other (non-POSIX) reactors for management of general
  252. outgoing connections.
  253. Requirements upon subclasses are documented as instance variables rather
  254. than abstract methods, in order to avoid MRO confusion, since this base is
  255. mixed in to unfortunately weird and distinctive multiple-inheritance
  256. hierarchies and many of these attributes are provided by peer classes
  257. rather than descendant classes in those hierarchies.
  258. @ivar addressFamily: The address family constant (C{socket.AF_INET},
  259. C{socket.AF_INET6}, C{socket.AF_UNIX}) of the underlying socket of this
  260. client connection.
  261. @type addressFamily: C{int}
  262. @ivar socketType: The socket type constant (C{socket.SOCK_STREAM} or
  263. C{socket.SOCK_DGRAM}) of the underlying socket.
  264. @type socketType: C{int}
  265. @ivar _requiresResolution: A flag indicating whether the address of this
  266. client will require name resolution. C{True} if the hostname of said
  267. address indicates a name that must be resolved by hostname lookup,
  268. C{False} if it indicates an IP address literal.
  269. @type _requiresResolution: C{bool}
  270. @cvar _commonConnection: Subclasses must provide this attribute, which
  271. indicates the L{Connection}-alike class to invoke C{__init__} and
  272. C{connectionLost} on.
  273. @type _commonConnection: C{type}
  274. @ivar _stopReadingAndWriting: Subclasses must implement in order to remove
  275. this transport from its reactor's notifications in response to a
  276. terminated connection attempt.
  277. @type _stopReadingAndWriting: 0-argument callable returning L{None}
  278. @ivar _closeSocket: Subclasses must implement in order to close the socket
  279. in response to a terminated connection attempt.
  280. @type _closeSocket: 1-argument callable; see L{_SocketCloser._closeSocket}
  281. @ivar _collectSocketDetails: Clean up references to the attached socket in
  282. its underlying OS resource (such as a file descriptor or file handle),
  283. as part of post connection-failure cleanup.
  284. @type _collectSocketDetails: 0-argument callable returning L{None}.
  285. @ivar reactor: The class pointed to by C{_commonConnection} should set this
  286. attribute in its constructor.
  287. @type reactor: L{twisted.internet.interfaces.IReactorTime},
  288. L{twisted.internet.interfaces.IReactorCore},
  289. L{twisted.internet.interfaces.IReactorFDSet}
  290. """
  291. addressFamily = socket.AF_INET
  292. socketType = socket.SOCK_STREAM
  293. def _finishInit(self, whenDone, skt, error, reactor):
  294. """
  295. Called by subclasses to continue to the stage of initialization where
  296. the socket connect attempt is made.
  297. @param whenDone: A 0-argument callable to invoke once the connection is
  298. set up. This is L{None} if the connection could not be prepared
  299. due to a previous error.
  300. @param skt: The socket object to use to perform the connection.
  301. @type skt: C{socket._socketobject}
  302. @param error: The error to fail the connection with.
  303. @param reactor: The reactor to use for this client.
  304. @type reactor: L{twisted.internet.interfaces.IReactorTime}
  305. """
  306. if whenDone:
  307. self._commonConnection.__init__(self, skt, None, reactor)
  308. reactor.callLater(0, whenDone)
  309. else:
  310. reactor.callLater(0, self.failIfNotConnected, error)
  311. def resolveAddress(self):
  312. """
  313. Resolve the name that was passed to this L{_BaseBaseClient}, if
  314. necessary, and then move on to attempting the connection once an
  315. address has been determined. (The connection will be attempted
  316. immediately within this function if either name resolution can be
  317. synchronous or the address was an IP address literal.)
  318. @note: You don't want to call this method from outside, as it won't do
  319. anything useful; it's just part of the connection bootstrapping
  320. process. Also, although this method is on L{_BaseBaseClient} for
  321. historical reasons, it's not used anywhere except for L{Client}
  322. itself.
  323. @return: L{None}
  324. """
  325. if self._requiresResolution:
  326. d = self.reactor.resolve(self.addr[0])
  327. d.addCallback(lambda n: (n,) + self.addr[1:])
  328. d.addCallbacks(self._setRealAddress, self.failIfNotConnected)
  329. else:
  330. self._setRealAddress(self.addr)
  331. def _setRealAddress(self, address):
  332. """
  333. Set the resolved address of this L{_BaseBaseClient} and initiate the
  334. connection attempt.
  335. @param address: Depending on whether this is an IPv4 or IPv6 connection
  336. attempt, a 2-tuple of C{(host, port)} or a 4-tuple of C{(host,
  337. port, flow, scope)}. At this point it is a fully resolved address,
  338. and the 'host' portion will always be an IP address, not a DNS
  339. name.
  340. """
  341. self.realAddress = address
  342. self.doConnect()
  343. def failIfNotConnected(self, err):
  344. """
  345. Generic method called when the attempts to connect failed. It basically
  346. cleans everything it can: call connectionFailed, stop read and write,
  347. delete socket related members.
  348. """
  349. if (self.connected or self.disconnected or
  350. not hasattr(self, "connector")):
  351. return
  352. self._stopReadingAndWriting()
  353. try:
  354. self._closeSocket(True)
  355. except AttributeError:
  356. pass
  357. else:
  358. self._collectSocketDetails()
  359. self.connector.connectionFailed(failure.Failure(err))
  360. del self.connector
  361. def stopConnecting(self):
  362. """
  363. If a connection attempt is still outstanding (i.e. no connection is
  364. yet established), immediately stop attempting to connect.
  365. """
  366. self.failIfNotConnected(error.UserError())
  367. def connectionLost(self, reason):
  368. """
  369. Invoked by lower-level logic when it's time to clean the socket up.
  370. Depending on the state of the connection, either inform the attached
  371. L{Connector} that the connection attempt has failed, or inform the
  372. connected L{IProtocol} that the established connection has been lost.
  373. @param reason: the reason that the connection was terminated
  374. @type reason: L{Failure}
  375. """
  376. if not self.connected:
  377. self.failIfNotConnected(error.ConnectError(string=reason))
  378. else:
  379. self._commonConnection.connectionLost(self, reason)
  380. self.connector.connectionLost(reason)
  381. class BaseClient(_BaseBaseClient, _TLSClientMixin, Connection):
  382. """
  383. A base class for client TCP (and similar) sockets.
  384. @ivar realAddress: The address object that will be used for socket.connect;
  385. this address is an address tuple (the number of elements dependent upon
  386. the address family) which does not contain any names which need to be
  387. resolved.
  388. @type realAddress: C{tuple}
  389. @ivar _base: L{Connection}, which is the base class of this class which has
  390. all of the useful file descriptor methods. This is used by
  391. L{_TLSServerMixin} to call the right methods to directly manipulate the
  392. transport, as is necessary for writing TLS-encrypted bytes (whereas
  393. those methods on L{Server} will go through another layer of TLS if it
  394. has been enabled).
  395. """
  396. _base = Connection
  397. _commonConnection = Connection
  398. def _stopReadingAndWriting(self):
  399. """
  400. Implement the POSIX-ish (i.e.
  401. L{twisted.internet.interfaces.IReactorFDSet}) method of detaching this
  402. socket from the reactor for L{_BaseBaseClient}.
  403. """
  404. if hasattr(self, "reactor"):
  405. # this doesn't happen if we failed in __init__
  406. self.stopReading()
  407. self.stopWriting()
  408. def _collectSocketDetails(self):
  409. """
  410. Clean up references to the socket and its file descriptor.
  411. @see: L{_BaseBaseClient}
  412. """
  413. del self.socket, self.fileno
  414. def createInternetSocket(self):
  415. """(internal) Create a non-blocking socket using
  416. self.addressFamily, self.socketType.
  417. """
  418. s = socket.socket(self.addressFamily, self.socketType)
  419. s.setblocking(0)
  420. fdesc._setCloseOnExec(s.fileno())
  421. return s
  422. def doConnect(self):
  423. """
  424. Initiate the outgoing connection attempt.
  425. @note: Applications do not need to call this method; it will be invoked
  426. internally as part of L{IReactorTCP.connectTCP}.
  427. """
  428. self.doWrite = self.doConnect
  429. self.doRead = self.doConnect
  430. if not hasattr(self, "connector"):
  431. # this happens when connection failed but doConnect
  432. # was scheduled via a callLater in self._finishInit
  433. return
  434. err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
  435. if err:
  436. self.failIfNotConnected(error.getConnectError((err, strerror(err))))
  437. return
  438. # doConnect gets called twice. The first time we actually need to
  439. # start the connection attempt. The second time we don't really
  440. # want to (SO_ERROR above will have taken care of any errors, and if
  441. # it reported none, the mere fact that doConnect was called again is
  442. # sufficient to indicate that the connection has succeeded), but it
  443. # is not /particularly/ detrimental to do so. This should get
  444. # cleaned up some day, though.
  445. try:
  446. connectResult = self.socket.connect_ex(self.realAddress)
  447. except socket.error as se:
  448. connectResult = se.args[0]
  449. if connectResult:
  450. if connectResult == EISCONN:
  451. pass
  452. # on Windows EINVAL means sometimes that we should keep trying:
  453. # http://msdn.microsoft.com/library/default.asp?url=/library/en-us/winsock/winsock/connect_2.asp
  454. elif ((connectResult in (EWOULDBLOCK, EINPROGRESS, EALREADY)) or
  455. (connectResult == EINVAL and platformType == "win32")):
  456. self.startReading()
  457. self.startWriting()
  458. return
  459. else:
  460. self.failIfNotConnected(error.getConnectError((connectResult, strerror(connectResult))))
  461. return
  462. # If I have reached this point without raising or returning, that means
  463. # that the socket is connected.
  464. del self.doWrite
  465. del self.doRead
  466. # we first stop and then start, to reset any references to the old doRead
  467. self.stopReading()
  468. self.stopWriting()
  469. self._connectDone()
  470. def _connectDone(self):
  471. """
  472. This is a hook for when a connection attempt has succeeded.
  473. Here, we build the protocol from the
  474. L{twisted.internet.protocol.ClientFactory} that was passed in, compute
  475. a log string, begin reading so as to send traffic to the newly built
  476. protocol, and finally hook up the protocol itself.
  477. This hook is overridden by L{ssl.Client} to initiate the TLS protocol.
  478. """
  479. self.protocol = self.connector.buildProtocol(self.getPeer())
  480. self.connected = 1
  481. logPrefix = self._getLogPrefix(self.protocol)
  482. self.logstr = "%s,client" % logPrefix
  483. if self.protocol is None:
  484. # Factory.buildProtocol is allowed to return None. In that case,
  485. # make up a protocol to satisfy the rest of the implementation;
  486. # connectionLost is going to be called on something, for example.
  487. # This is easier than adding special case support for a None
  488. # protocol throughout the rest of the transport implementation.
  489. self.protocol = Protocol()
  490. # But dispose of the connection quickly.
  491. self.loseConnection()
  492. else:
  493. self.startReading()
  494. self.protocol.makeConnection(self)
  495. _NUMERIC_ONLY = socket.AI_NUMERICHOST | _AI_NUMERICSERV
  496. def _resolveIPv6(ip, port):
  497. """
  498. Resolve an IPv6 literal into an IPv6 address.
  499. This is necessary to resolve any embedded scope identifiers to the relevant
  500. C{sin6_scope_id} for use with C{socket.connect()}, C{socket.listen()}, or
  501. C{socket.bind()}; see U{RFC 3493 <https://tools.ietf.org/html/rfc3493>} for
  502. more information.
  503. @param ip: An IPv6 address literal.
  504. @type ip: C{str}
  505. @param port: A port number.
  506. @type port: C{int}
  507. @return: a 4-tuple of C{(host, port, flow, scope)}, suitable for use as an
  508. IPv6 address.
  509. @raise socket.gaierror: if either the IP or port is not numeric as it
  510. should be.
  511. """
  512. return socket.getaddrinfo(ip, port, 0, 0, 0, _NUMERIC_ONLY)[0][4]
  513. class _BaseTCPClient(object):
  514. """
  515. Code shared with other (non-POSIX) reactors for management of outgoing TCP
  516. connections (both TCPv4 and TCPv6).
  517. @note: In order to be functional, this class must be mixed into the same
  518. hierarchy as L{_BaseBaseClient}. It would subclass L{_BaseBaseClient}
  519. directly, but the class hierarchy here is divided in strange ways out
  520. of the need to share code along multiple axes; specifically, with the
  521. IOCP reactor and also with UNIX clients in other reactors.
  522. @ivar _addressType: The Twisted _IPAddress implementation for this client
  523. @type _addressType: L{IPv4Address} or L{IPv6Address}
  524. @ivar connector: The L{Connector} which is driving this L{_BaseTCPClient}'s
  525. connection attempt.
  526. @ivar addr: The address that this socket will be connecting to.
  527. @type addr: If IPv4, a 2-C{tuple} of C{(str host, int port)}. If IPv6, a
  528. 4-C{tuple} of (C{str host, int port, int ignored, int scope}).
  529. @ivar createInternetSocket: Subclasses must implement this as a method to
  530. create a python socket object of the appropriate address family and
  531. socket type.
  532. @type createInternetSocket: 0-argument callable returning
  533. C{socket._socketobject}.
  534. """
  535. _addressType = address.IPv4Address
  536. def __init__(self, host, port, bindAddress, connector, reactor=None):
  537. # BaseClient.__init__ is invoked later
  538. self.connector = connector
  539. self.addr = (host, port)
  540. whenDone = self.resolveAddress
  541. err = None
  542. skt = None
  543. if abstract.isIPAddress(host):
  544. self._requiresResolution = False
  545. elif abstract.isIPv6Address(host):
  546. self._requiresResolution = False
  547. self.addr = _resolveIPv6(host, port)
  548. self.addressFamily = socket.AF_INET6
  549. self._addressType = address.IPv6Address
  550. else:
  551. self._requiresResolution = True
  552. try:
  553. skt = self.createInternetSocket()
  554. except socket.error as se:
  555. err = error.ConnectBindError(se.args[0], se.args[1])
  556. whenDone = None
  557. if whenDone and bindAddress is not None:
  558. try:
  559. if abstract.isIPv6Address(bindAddress[0]):
  560. bindinfo = _resolveIPv6(*bindAddress)
  561. else:
  562. bindinfo = bindAddress
  563. skt.bind(bindinfo)
  564. except socket.error as se:
  565. err = error.ConnectBindError(se.args[0], se.args[1])
  566. whenDone = None
  567. self._finishInit(whenDone, skt, err, reactor)
  568. def getHost(self):
  569. """
  570. Returns an L{IPv4Address} or L{IPv6Address}.
  571. This indicates the address from which I am connecting.
  572. """
  573. return self._addressType('TCP', *self.socket.getsockname()[:2])
  574. def getPeer(self):
  575. """
  576. Returns an L{IPv4Address} or L{IPv6Address}.
  577. This indicates the address that I am connected to.
  578. """
  579. # an ipv6 realAddress has more than two elements, but the IPv6Address
  580. # constructor still only takes two.
  581. return self._addressType('TCP', *self.realAddress[:2])
  582. def __repr__(self):
  583. s = '<%s to %s at %x>' % (self.__class__, self.addr, id(self))
  584. return s
  585. class Client(_BaseTCPClient, BaseClient):
  586. """
  587. A transport for a TCP protocol; either TCPv4 or TCPv6.
  588. Do not create these directly; use L{IReactorTCP.connectTCP}.
  589. """
  590. class Server(_TLSServerMixin, Connection):
  591. """
  592. Serverside socket-stream connection class.
  593. This is a serverside network connection transport; a socket which came from
  594. an accept() on a server.
  595. @ivar _base: L{Connection}, which is the base class of this class which has
  596. all of the useful file descriptor methods. This is used by
  597. L{_TLSServerMixin} to call the right methods to directly manipulate the
  598. transport, as is necessary for writing TLS-encrypted bytes (whereas
  599. those methods on L{Server} will go through another layer of TLS if it
  600. has been enabled).
  601. """
  602. _base = Connection
  603. _addressType = address.IPv4Address
  604. def __init__(self, sock, protocol, client, server, sessionno, reactor):
  605. """
  606. Server(sock, protocol, client, server, sessionno)
  607. Initialize it with a socket, a protocol, a descriptor for my peer (a
  608. tuple of host, port describing the other end of the connection), an
  609. instance of Port, and a session number.
  610. """
  611. Connection.__init__(self, sock, protocol, reactor)
  612. if len(client) != 2:
  613. self._addressType = address.IPv6Address
  614. self.server = server
  615. self.client = client
  616. self.sessionno = sessionno
  617. self.hostname = client[0]
  618. logPrefix = self._getLogPrefix(self.protocol)
  619. self.logstr = "%s,%s,%s" % (logPrefix,
  620. sessionno,
  621. self.hostname)
  622. if self.server is not None:
  623. self.repstr = "<%s #%s on %s>" % (self.protocol.__class__.__name__,
  624. self.sessionno,
  625. self.server._realPortNumber)
  626. self.startReading()
  627. self.connected = 1
  628. def __repr__(self):
  629. """
  630. A string representation of this connection.
  631. """
  632. return self.repstr
  633. @classmethod
  634. def _fromConnectedSocket(cls, fileDescriptor, addressFamily, factory,
  635. reactor):
  636. """
  637. Create a new L{Server} based on an existing connected I{SOCK_STREAM}
  638. socket.
  639. Arguments are the same as to L{Server.__init__}, except where noted.
  640. @param fileDescriptor: An integer file descriptor associated with a
  641. connected socket. The socket must be in non-blocking mode. Any
  642. additional attributes desired, such as I{FD_CLOEXEC}, must also be
  643. set already.
  644. @param addressFamily: The address family (sometimes called I{domain})
  645. of the existing socket. For example, L{socket.AF_INET}.
  646. @return: A new instance of C{cls} wrapping the socket given by
  647. C{fileDescriptor}.
  648. """
  649. addressType = address.IPv4Address
  650. if addressFamily == socket.AF_INET6:
  651. addressType = address.IPv6Address
  652. skt = socket.fromfd(fileDescriptor, addressFamily, socket.SOCK_STREAM)
  653. addr = skt.getpeername()
  654. protocolAddr = addressType('TCP', addr[0], addr[1])
  655. localPort = skt.getsockname()[1]
  656. protocol = factory.buildProtocol(protocolAddr)
  657. if protocol is None:
  658. skt.close()
  659. return
  660. self = cls(skt, protocol, addr, None, addr[1], reactor)
  661. self.repstr = "<%s #%s on %s>" % (
  662. self.protocol.__class__.__name__, self.sessionno, localPort)
  663. protocol.makeConnection(self)
  664. return self
  665. def getHost(self):
  666. """
  667. Returns an L{IPv4Address} or L{IPv6Address}.
  668. This indicates the server's address.
  669. """
  670. host, port = self.socket.getsockname()[:2]
  671. return self._addressType('TCP', host, port)
  672. def getPeer(self):
  673. """
  674. Returns an L{IPv4Address} or L{IPv6Address}.
  675. This indicates the client's address.
  676. """
  677. return self._addressType('TCP', *self.client[:2])
  678. @implementer(interfaces.IListeningPort)
  679. class Port(base.BasePort, _SocketCloser):
  680. """
  681. A TCP server port, listening for connections.
  682. When a connection is accepted, this will call a factory's buildProtocol
  683. with the incoming address as an argument, according to the specification
  684. described in L{twisted.internet.interfaces.IProtocolFactory}.
  685. If you wish to change the sort of transport that will be used, the
  686. C{transport} attribute will be called with the signature expected for
  687. C{Server.__init__}, so it can be replaced.
  688. @ivar deferred: a deferred created when L{stopListening} is called, and
  689. that will fire when connection is lost. This is not to be used it
  690. directly: prefer the deferred returned by L{stopListening} instead.
  691. @type deferred: L{defer.Deferred}
  692. @ivar disconnecting: flag indicating that the L{stopListening} method has
  693. been called and that no connections should be accepted anymore.
  694. @type disconnecting: C{bool}
  695. @ivar connected: flag set once the listen has successfully been called on
  696. the socket.
  697. @type connected: C{bool}
  698. @ivar _type: A string describing the connections which will be created by
  699. this port. Normally this is C{"TCP"}, since this is a TCP port, but
  700. when the TLS implementation re-uses this class it overrides the value
  701. with C{"TLS"}. Only used for logging.
  702. @ivar _preexistingSocket: If not L{None}, a L{socket.socket} instance which
  703. was created and initialized outside of the reactor and will be used to
  704. listen for connections (instead of a new socket being created by this
  705. L{Port}).
  706. """
  707. socketType = socket.SOCK_STREAM
  708. transport = Server
  709. sessionno = 0
  710. interface = ''
  711. backlog = 50
  712. _type = 'TCP'
  713. # Actual port number being listened on, only set to a non-None
  714. # value when we are actually listening.
  715. _realPortNumber = None
  716. # An externally initialized socket that we will use, rather than creating
  717. # our own.
  718. _preexistingSocket = None
  719. addressFamily = socket.AF_INET
  720. _addressType = address.IPv4Address
  721. def __init__(self, port, factory, backlog=50, interface='', reactor=None):
  722. """Initialize with a numeric port to listen on.
  723. """
  724. base.BasePort.__init__(self, reactor=reactor)
  725. self.port = port
  726. self.factory = factory
  727. self.backlog = backlog
  728. if abstract.isIPv6Address(interface):
  729. self.addressFamily = socket.AF_INET6
  730. self._addressType = address.IPv6Address
  731. self.interface = interface
  732. @classmethod
  733. def _fromListeningDescriptor(cls, reactor, fd, addressFamily, factory):
  734. """
  735. Create a new L{Port} based on an existing listening I{SOCK_STREAM}
  736. socket.
  737. Arguments are the same as to L{Port.__init__}, except where noted.
  738. @param fd: An integer file descriptor associated with a listening
  739. socket. The socket must be in non-blocking mode. Any additional
  740. attributes desired, such as I{FD_CLOEXEC}, must also be set already.
  741. @param addressFamily: The address family (sometimes called I{domain}) of
  742. the existing socket. For example, L{socket.AF_INET}.
  743. @return: A new instance of C{cls} wrapping the socket given by C{fd}.
  744. """
  745. port = socket.fromfd(fd, addressFamily, cls.socketType)
  746. interface = port.getsockname()[0]
  747. self = cls(None, factory, None, interface, reactor)
  748. self._preexistingSocket = port
  749. return self
  750. def __repr__(self):
  751. if self._realPortNumber is not None:
  752. return "<%s of %s on %s>" % (self.__class__,
  753. self.factory.__class__, self._realPortNumber)
  754. else:
  755. return "<%s of %s (not listening)>" % (self.__class__, self.factory.__class__)
  756. def createInternetSocket(self):
  757. s = base.BasePort.createInternetSocket(self)
  758. if platformType == "posix" and sys.platform != "cygwin":
  759. s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  760. return s
  761. def startListening(self):
  762. """Create and bind my socket, and begin listening on it.
  763. This is called on unserialization, and must be called after creating a
  764. server to begin listening on the specified port.
  765. """
  766. if self._preexistingSocket is None:
  767. # Create a new socket and make it listen
  768. try:
  769. skt = self.createInternetSocket()
  770. if self.addressFamily == socket.AF_INET6:
  771. addr = _resolveIPv6(self.interface, self.port)
  772. else:
  773. addr = (self.interface, self.port)
  774. skt.bind(addr)
  775. except socket.error as le:
  776. raise CannotListenError(self.interface, self.port, le)
  777. skt.listen(self.backlog)
  778. else:
  779. # Re-use the externally specified socket
  780. skt = self._preexistingSocket
  781. self._preexistingSocket = None
  782. # Avoid shutting it down at the end.
  783. self._shouldShutdown = False
  784. # Make sure that if we listened on port 0, we update that to
  785. # reflect what the OS actually assigned us.
  786. self._realPortNumber = skt.getsockname()[1]
  787. log.msg("%s starting on %s" % (
  788. self._getLogPrefix(self.factory), self._realPortNumber))
  789. # The order of the next 5 lines is kind of bizarre. If no one
  790. # can explain it, perhaps we should re-arrange them.
  791. self.factory.doStart()
  792. self.connected = True
  793. self.socket = skt
  794. self.fileno = self.socket.fileno
  795. self.numberAccepts = 100
  796. self.startReading()
  797. def _buildAddr(self, address):
  798. host, port = address[:2]
  799. return self._addressType('TCP', host, port)
  800. def doRead(self):
  801. """Called when my socket is ready for reading.
  802. This accepts a connection and calls self.protocol() to handle the
  803. wire-level protocol.
  804. """
  805. try:
  806. if platformType == "posix":
  807. numAccepts = self.numberAccepts
  808. else:
  809. # win32 event loop breaks if we do more than one accept()
  810. # in an iteration of the event loop.
  811. numAccepts = 1
  812. for i in range(numAccepts):
  813. # we need this so we can deal with a factory's buildProtocol
  814. # calling our loseConnection
  815. if self.disconnecting:
  816. return
  817. try:
  818. skt, addr = self.socket.accept()
  819. except socket.error as e:
  820. if e.args[0] in (EWOULDBLOCK, EAGAIN):
  821. self.numberAccepts = i
  822. break
  823. elif e.args[0] == EPERM:
  824. # Netfilter on Linux may have rejected the
  825. # connection, but we get told to try to accept()
  826. # anyway.
  827. continue
  828. elif e.args[0] in (EMFILE, ENOBUFS, ENFILE, ENOMEM, ECONNABORTED):
  829. # Linux gives EMFILE when a process is not allowed to
  830. # allocate any more file descriptors. *BSD and Win32
  831. # give (WSA)ENOBUFS. Linux can also give ENFILE if the
  832. # system is out of inodes, or ENOMEM if there is
  833. # insufficient memory to allocate a new dentry.
  834. # ECONNABORTED is documented as possible on all
  835. # relevant platforms (Linux, Windows, macOS, and the
  836. # BSDs) but occurs only on the BSDs. It occurs when a
  837. # client sends a FIN or RST after the server sends a
  838. # SYN|ACK but before application code calls accept(2).
  839. # On Linux, calling accept(2) on such a listener
  840. # returns a connection that fails as though the it were
  841. # terminated after being fully established. This
  842. # appears to be an implementation choice (see
  843. # inet_accept in inet/ipv4/af_inet.c). On macOS X,
  844. # such a listener is not considered readable, so
  845. # accept(2) will never be called. Calling accept(2) on
  846. # such a listener, however, does not return at all.
  847. log.msg("Could not accept new connection (%s)" % (
  848. errorcode[e.args[0]],))
  849. break
  850. raise
  851. fdesc._setCloseOnExec(skt.fileno())
  852. protocol = self.factory.buildProtocol(self._buildAddr(addr))
  853. if protocol is None:
  854. skt.close()
  855. continue
  856. s = self.sessionno
  857. self.sessionno = s+1
  858. transport = self.transport(skt, protocol, addr, self, s, self.reactor)
  859. protocol.makeConnection(transport)
  860. else:
  861. self.numberAccepts = self.numberAccepts+20
  862. except:
  863. # Note that in TLS mode, this will possibly catch SSL.Errors
  864. # raised by self.socket.accept()
  865. #
  866. # There is no "except SSL.Error:" above because SSL may be
  867. # None if there is no SSL support. In any case, all the
  868. # "except SSL.Error:" suite would probably do is log.deferr()
  869. # and return, so handling it here works just as well.
  870. log.deferr()
  871. def loseConnection(self, connDone=failure.Failure(main.CONNECTION_DONE)):
  872. """
  873. Stop accepting connections on this port.
  874. This will shut down the socket and call self.connectionLost(). It
  875. returns a deferred which will fire successfully when the port is
  876. actually closed, or with a failure if an error occurs shutting down.
  877. """
  878. self.disconnecting = True
  879. self.stopReading()
  880. if self.connected:
  881. self.deferred = deferLater(
  882. self.reactor, 0, self.connectionLost, connDone)
  883. return self.deferred
  884. stopListening = loseConnection
  885. def _logConnectionLostMsg(self):
  886. """
  887. Log message for closing port
  888. """
  889. log.msg('(%s Port %s Closed)' % (self._type, self._realPortNumber))
  890. def connectionLost(self, reason):
  891. """
  892. Cleans up the socket.
  893. """
  894. self._logConnectionLostMsg()
  895. self._realPortNumber = None
  896. base.BasePort.connectionLost(self, reason)
  897. self.connected = False
  898. self._closeSocket(True)
  899. del self.socket
  900. del self.fileno
  901. try:
  902. self.factory.doStop()
  903. finally:
  904. self.disconnecting = False
  905. def logPrefix(self):
  906. """Returns the name of my class, to prefix log entries with.
  907. """
  908. return reflect.qual(self.factory.__class__)
  909. def getHost(self):
  910. """
  911. Return an L{IPv4Address} or L{IPv6Address} indicating the listening
  912. address of this port.
  913. """
  914. host, port = self.socket.getsockname()[:2]
  915. return self._addressType('TCP', host, port)
  916. class Connector(base.BaseConnector):
  917. """
  918. A L{Connector} provides of L{twisted.internet.interfaces.IConnector} for
  919. all POSIX-style reactors.
  920. @ivar _addressType: the type returned by L{Connector.getDestination}.
  921. Either L{IPv4Address} or L{IPv6Address}, depending on the type of
  922. address.
  923. @type _addressType: C{type}
  924. """
  925. _addressType = address.IPv4Address
  926. def __init__(self, host, port, factory, timeout, bindAddress, reactor=None):
  927. if isinstance(port, _portNameType):
  928. try:
  929. port = socket.getservbyname(port, 'tcp')
  930. except socket.error as e:
  931. raise error.ServiceNameUnknownError(string="%s (%r)" % (e, port))
  932. self.host, self.port = host, port
  933. if abstract.isIPv6Address(host):
  934. self._addressType = address.IPv6Address
  935. self.bindAddress = bindAddress
  936. base.BaseConnector.__init__(self, factory, timeout, reactor)
  937. def _makeTransport(self):
  938. """
  939. Create a L{Client} bound to this L{Connector}.
  940. @return: a new L{Client}
  941. @rtype: L{Client}
  942. """
  943. return Client(self.host, self.port, self.bindAddress, self, self.reactor)
  944. def getDestination(self):
  945. """
  946. @see: L{twisted.internet.interfaces.IConnector.getDestination}.
  947. """
  948. return self._addressType('TCP', self.host, self.port)