twisted/internet/endpoints.py

   1 # -*- test-case-name: twisted.internet.test.test_endpoints -*-
   2 # Copyright (c) Twisted Matrix Laboratories.
   3 # See LICENSE for details.
   4 """
   5 Implementations of L{IStreamServerEndpoint} and L{IStreamClientEndpoint} that
   6 wrap the L{IReactorTCP}, L{IReactorSSL}, and L{IReactorUNIX} interfaces.
   7
   8 This also implements an extensible mini-language for describing endpoints,
   9 parsed by the L{clientFromString} and L{serverFromString} functions.
  10
  11 @since: 10.1
  12 """
  13
  14 import os, socket
  15
  16 from zope.interface import implements, directlyProvides
  17 import warnings
  18
  19 from twisted.internet import interfaces, defer, error, fdesc
  20 from twisted.internet.protocol import ClientFactory, Protocol
  21 from twisted.plugin import IPlugin, getPlugins
  22 from twisted.internet.interfaces import IStreamServerEndpointStringParser
  23 from twisted.internet.interfaces import IStreamClientEndpointStringParser
  24 from twisted.python.filepath import FilePath
  25 from twisted.python.systemd import ListenFDs
  26
  27
  28 __all__ = ["clientFromString", "serverFromString",
  29            "TCP4ServerEndpoint", "TCP4ClientEndpoint",
  30            "UNIXServerEndpoint", "UNIXClientEndpoint",
  31            "SSL4ServerEndpoint", "SSL4ClientEndpoint",
  32            "AdoptedStreamServerEndpoint"]
  33
  34
  35 class _WrappingProtocol(Protocol):
  36     """
  37     Wrap another protocol in order to notify my user when a connection has
  38     been made.
  39
  40     @ivar _connectedDeferred: The L{Deferred} that will callback
  41         with the C{wrappedProtocol} when it is connected.
  42
  43     @ivar _wrappedProtocol: An L{IProtocol} provider that will be
  44         connected.
  45     """
  46
  47     def __init__(self, connectedDeferred, wrappedProtocol):
  48         """
  49         @param connectedDeferred: The L{Deferred} that will callback
  50             with the C{wrappedProtocol} when it is connected.
  51
  52         @param wrappedProtocol: An L{IProtocol} provider that will be
  53             connected.
  54         """
  55         self._connectedDeferred = connectedDeferred
  56         self._wrappedProtocol = wrappedProtocol
  57
  58         for iface in [interfaces.IHalfCloseableProtocol,
  59                       interfaces.IFileDescriptorReceiver]:
  60             if iface.providedBy(self._wrappedProtocol):
  61                 directlyProvides(self, iface)
  62
  63
  64     def logPrefix(self):
  65         """
  66         Transparently pass through the wrapped protocol's log prefix.
  67         """
  68         if interfaces.ILoggingContext.providedBy(self._wrappedProtocol):
  69             return self._wrappedProtocol.logPrefix()
  70         return self._wrappedProtocol.__class__.__name__
  71
  72
  73     def connectionMade(self):
  74         """
  75         Connect the C{self._wrappedProtocol} to our C{self.transport} and
  76         callback C{self._connectedDeferred} with the C{self._wrappedProtocol}
  77         """
  78         self._wrappedProtocol.makeConnection(self.transport)
  79         self._connectedDeferred.callback(self._wrappedProtocol)
  80
  81
  82     def dataReceived(self, data):
  83         """
  84         Proxy C{dataReceived} calls to our C{self._wrappedProtocol}
  85         """
  86         return self._wrappedProtocol.dataReceived(data)
  87
  88
  89     def fileDescriptorReceived(self, descriptor):
  90         """
  91         Proxy C{fileDescriptorReceived} calls to our C{self._wrappedProtocol}
  92         """
  93         return self._wrappedProtocol.fileDescriptorReceived(descriptor)
  94
  95
  96     def connectionLost(self, reason):
  97         """
  98         Proxy C{connectionLost} calls to our C{self._wrappedProtocol}
  99         """
 100         return self._wrappedProtocol.connectionLost(reason)
 101
 102
 103     def readConnectionLost(self):
 104         """
 105         Proxy L{IHalfCloseableProtocol.readConnectionLost} to our
 106         C{self._wrappedProtocol}
 107         """
 108         self._wrappedProtocol.readConnectionLost()
 109
 110
 111     def writeConnectionLost(self):
 112         """
 113         Proxy L{IHalfCloseableProtocol.writeConnectionLost} to our
 114         C{self._wrappedProtocol}
 115         """
 116         self._wrappedProtocol.writeConnectionLost()
 117
 118
 119
 120 class _WrappingFactory(ClientFactory):
 121     """
 122     Wrap a factory in order to wrap the protocols it builds.
 123
 124     @ivar _wrappedFactory: A provider of I{IProtocolFactory} whose buildProtocol
 125         method will be called and whose resulting protocol will be wrapped.
 126
 127     @ivar _onConnection: An L{Deferred} that fires when the protocol is
 128         connected
 129
 130     @ivar _connector: A L{connector <twisted.internet.interfaces.IConnector>}
 131         that is managing the current or previous connection attempt.
 132     """
 133     protocol = _WrappingProtocol
 134
 135     def __init__(self, wrappedFactory):
 136         """
 137         @param wrappedFactory: A provider of I{IProtocolFactory} whose
 138             buildProtocol method will be called and whose resulting protocol
 139             will be wrapped.
 140         """
 141         self._wrappedFactory = wrappedFactory
 142         self._onConnection = defer.Deferred(canceller=self._canceller)
 143
 144
 145     def startedConnecting(self, connector):
 146         """
 147         A connection attempt was started.  Remember the connector which started
 148         said attempt, for use later.
 149         """
 150         self._connector = connector
 151
 152
 153     def _canceller(self, deferred):
 154         """
 155         The outgoing connection attempt was cancelled.  Fail that L{Deferred}
 156         with a L{error.ConnectingCancelledError}.
 157
 158         @param deferred: The L{Deferred <defer.Deferred>} that was cancelled;
 159             should be the same as C{self._onConnection}.
 160         @type deferred: L{Deferred <defer.Deferred>}
 161
 162         @note: This relies on startedConnecting having been called, so it may
 163             seem as though there's a race condition where C{_connector} may not
 164             have been set.  However, using public APIs, this condition is
 165             impossible to catch, because a connection API
 166             (C{connectTCP}/C{SSL}/C{UNIX}) is always invoked before a
 167             L{_WrappingFactory}'s L{Deferred <defer.Deferred>} is returned to
 168             C{connect()}'s caller.
 169
 170         @return: C{None}
 171         """
 172         deferred.errback(
 173             error.ConnectingCancelledError(
 174                 self._connector.getDestination()))
 175         self._connector.stopConnecting()
 176
 177
 178     def doStart(self):
 179         """
 180         Start notifications are passed straight through to the wrapped factory.
 181         """
 182         self._wrappedFactory.doStart()
 183
 184
 185     def doStop(self):
 186         """
 187         Stop notifications are passed straight through to the wrapped factory.
 188         """
 189         self._wrappedFactory.doStop()
 190
 191
 192     def buildProtocol(self, addr):
 193         """
 194         Proxy C{buildProtocol} to our C{self._wrappedFactory} or errback
 195         the C{self._onConnection} L{Deferred}.
 196
 197         @return: An instance of L{_WrappingProtocol} or C{None}
 198         """
 199         try:
 200             proto = self._wrappedFactory.buildProtocol(addr)
 201         except:
 202             self._onConnection.errback()
 203         else:
 204             return self.protocol(self._onConnection, proto)
 205
 206
 207     def clientConnectionFailed(self, connector, reason):
 208         """
 209         Errback the C{self._onConnection} L{Deferred} when the
 210         client connection fails.
 211         """
 212         if not self._onConnection.called:
 213             self._onConnection.errback(reason)
 214
 215
 216
 217 class TCP4ServerEndpoint(object):
 218     """
 219     TCP server endpoint with an IPv4 configuration
 220
 221     @ivar _reactor: An L{IReactorTCP} provider.
 222
 223     @type _port: int
 224     @ivar _port: The port number on which to listen for incoming connections.
 225
 226     @type _backlog: int
 227     @ivar _backlog: size of the listen queue
 228
 229     @type _interface: str
 230     @ivar _interface: the hostname to bind to, defaults to '' (all)
 231     """
 232     implements(interfaces.IStreamServerEndpoint)
 233
 234     def __init__(self, reactor, port, backlog=50, interface=''):
 235         """
 236         @param reactor: An L{IReactorTCP} provider.
 237         @param port: The port number used listening
 238         @param backlog: size of the listen queue
 239         @param interface: the hostname to bind to, defaults to '' (all)
 240         """
 241         self._reactor = reactor
 242         self._port = port
 243         self._listenArgs = dict(backlog=50, interface='')
 244         self._backlog = backlog
 245         self._interface = interface
 246
 247
 248     def listen(self, protocolFactory):
 249         """
 250         Implement L{IStreamServerEndpoint.listen} to listen on a TCP socket
 251         """
 252         return defer.execute(self._reactor.listenTCP,
 253                              self._port,
 254                              protocolFactory,
 255                              backlog=self._backlog,
 256                              interface=self._interface)
 257
 258
 259
 260 class TCP4ClientEndpoint(object):
 261     """
 262     TCP client endpoint with an IPv4 configuration.
 263
 264     @ivar _reactor: An L{IReactorTCP} provider.
 265
 266     @type _host: str
 267     @ivar _host: The hostname to connect to as a C{str}
 268
 269     @type _port: int
 270     @ivar _port: The port to connect to as C{int}
 271
 272     @type _timeout: int
 273     @ivar _timeout: number of seconds to wait before assuming the
 274         connection has failed.
 275
 276     @type _bindAddress: tuple
 277     @type _bindAddress: a (host, port) tuple of local address to bind
 278         to, or None.
 279     """
 280     implements(interfaces.IStreamClientEndpoint)
 281
 282     def __init__(self, reactor, host, port, timeout=30, bindAddress=None):
 283         """
 284         @param reactor: An L{IReactorTCP} provider
 285         @param host: A hostname, used when connecting
 286         @param port: The port number, used when connecting
 287         @param timeout: number of seconds to wait before assuming the
 288             connection has failed.
 289         @param bindAddress: a (host, port tuple of local address to bind to,
 290             or None.
 291         """
 292         self._reactor = reactor
 293         self._host = host
 294         self._port = port
 295         self._timeout = timeout
 296         self._bindAddress = bindAddress
 297
 298
 299     def connect(self, protocolFactory):
 300         """
 301         Implement L{IStreamClientEndpoint.connect} to connect via TCP.
 302         """
 303         try:
 304             wf = _WrappingFactory(protocolFactory)
 305             self._reactor.connectTCP(
 306                 self._host, self._port, wf,
 307                 timeout=self._timeout, bindAddress=self._bindAddress)
 308             return wf._onConnection
 309         except:
 310             return defer.fail()
 311
 312
 313
 314 class SSL4ServerEndpoint(object):
 315     """
 316     SSL secured TCP server endpoint with an IPv4 configuration.
 317
 318     @ivar _reactor: An L{IReactorSSL} provider.
 319
 320     @type _host: str
 321     @ivar _host: The hostname to connect to as a C{str}
 322
 323     @type _port: int
 324     @ivar _port: The port to connect to as C{int}
 325
 326     @type _sslContextFactory: L{OpenSSLCertificateOptions}
 327     @var _sslContextFactory: SSL Configuration information as an
 328         L{OpenSSLCertificateOptions}
 329
 330     @type _backlog: int
 331     @ivar _backlog: size of the listen queue
 332
 333     @type _interface: str
 334     @ivar _interface: the hostname to bind to, defaults to '' (all)
 335     """
 336     implements(interfaces.IStreamServerEndpoint)
 337
 338     def __init__(self, reactor, port, sslContextFactory,
 339                  backlog=50, interface=''):
 340         """
 341         @param reactor: An L{IReactorSSL} provider.
 342         @param port: The port number used listening
 343         @param sslContextFactory: An instance of
 344             L{twisted.internet._sslverify.OpenSSLCertificateOptions}.
 345         @param timeout: number of seconds to wait before assuming the
 346             connection has failed.
 347         @param bindAddress: a (host, port tuple of local address to bind to,
 348             or None.
 349         """
 350         self._reactor = reactor
 351         self._port = port
 352         self._sslContextFactory = sslContextFactory
 353         self._backlog = backlog
 354         self._interface = interface
 355
 356
 357     def listen(self, protocolFactory):
 358         """
 359         Implement L{IStreamServerEndpoint.listen} to listen for SSL on a
 360         TCP socket.
 361         """
 362         return defer.execute(self._reactor.listenSSL, self._port,
 363                              protocolFactory,
 364                              contextFactory=self._sslContextFactory,
 365                              backlog=self._backlog,
 366                              interface=self._interface)
 367
 368
 369
 370 class SSL4ClientEndpoint(object):
 371     """
 372     SSL secured TCP client endpoint with an IPv4 configuration
 373
 374     @ivar _reactor: An L{IReactorSSL} provider.
 375
 376     @type _host: str
 377     @ivar _host: The hostname to connect to as a C{str}
 378
 379     @type _port: int
 380     @ivar _port: The port to connect to as C{int}
 381
 382     @type _sslContextFactory: L{OpenSSLCertificateOptions}
 383     @var _sslContextFactory: SSL Configuration information as an
 384         L{OpenSSLCertificateOptions}
 385
 386     @type _timeout: int
 387     @ivar _timeout: number of seconds to wait before assuming the
 388         connection has failed.
 389
 390     @type _bindAddress: tuple
 391     @ivar _bindAddress: a (host, port) tuple of local address to bind
 392         to, or None.
 393     """
 394     implements(interfaces.IStreamClientEndpoint)
 395
 396     def __init__(self, reactor, host, port, sslContextFactory,
 397                  timeout=30, bindAddress=None):
 398         """
 399         @param reactor: An L{IReactorSSL} provider.
 400         @param host: A hostname, used when connecting
 401         @param port: The port number, used when connecting
 402         @param sslContextFactory: SSL Configuration information as An instance
 403             of L{OpenSSLCertificateOptions}.
 404         @param timeout: number of seconds to wait before assuming the
 405             connection has failed.
 406         @param bindAddress: a (host, port tuple of local address to bind to,
 407             or None.
 408         """
 409         self._reactor = reactor
 410         self._host = host
 411         self._port = port
 412         self._sslContextFactory = sslContextFactory
 413         self._timeout = timeout
 414         self._bindAddress = bindAddress
 415
 416
 417     def connect(self, protocolFactory):
 418         """
 419         Implement L{IStreamClientEndpoint.connect} to connect with SSL over
 420         TCP.
 421         """
 422         try:
 423             wf = _WrappingFactory(protocolFactory)
 424             self._reactor.connectSSL(
 425                 self._host, self._port, wf, self._sslContextFactory,
 426                 timeout=self._timeout, bindAddress=self._bindAddress)
 427             return wf._onConnection
 428         except:
 429             return defer.fail()
 430
 431
 432
 433 class UNIXServerEndpoint(object):
 434     """
 435     UnixSocket server endpoint.
 436
 437     @type path: str
 438     @ivar path: a path to a unix socket on the filesystem.
 439
 440     @type _listenArgs: dict
 441     @ivar _listenArgs: A C{dict} of keyword args that will be passed
 442         to L{IReactorUNIX.listenUNIX}
 443
 444     @var _reactor: An L{IReactorTCP} provider.
 445     """
 446     implements(interfaces.IStreamServerEndpoint)
 447
 448     def __init__(self, reactor, address, backlog=50, mode=0666, wantPID=0):
 449         """
 450         @param reactor: An L{IReactorUNIX} provider.
 451         @param address: The path to the Unix socket file, used when listening
 452         @param listenArgs: An optional dict of keyword args that will be
 453             passed to L{IReactorUNIX.listenUNIX}
 454         @param backlog: number of connections to allow in backlog.
 455         @param mode: mode to set on the unix socket.  This parameter is
 456             deprecated.  Permissions should be set on the directory which
 457             contains the UNIX socket.
 458         @param wantPID: if True, create a pidfile for the socket.
 459         """
 460         self._reactor = reactor
 461         self._address = address
 462         self._backlog = backlog
 463         self._mode = mode
 464         self._wantPID = wantPID
 465
 466
 467     def listen(self, protocolFactory):
 468         """
 469         Implement L{IStreamServerEndpoint.listen} to listen on a UNIX socket.
 470         """
 471         return defer.execute(self._reactor.listenUNIX, self._address,
 472                              protocolFactory,
 473                              backlog=self._backlog,
 474                              mode=self._mode,
 475                              wantPID=self._wantPID)
 476
 477
 478
 479 class UNIXClientEndpoint(object):
 480     """
 481     UnixSocket client endpoint.
 482
 483     @type _path: str
 484     @ivar _path: a path to a unix socket on the filesystem.
 485
 486     @type _timeout: int
 487     @ivar _timeout: number of seconds to wait before assuming the connection
 488         has failed.
 489
 490     @type _checkPID: bool
 491     @ivar _checkPID: if True, check for a pid file to verify that a server
 492         is listening.
 493
 494     @var _reactor: An L{IReactorUNIX} provider.
 495     """
 496     implements(interfaces.IStreamClientEndpoint)
 497
 498     def __init__(self, reactor, path, timeout=30, checkPID=0):
 499         """
 500         @param reactor: An L{IReactorUNIX} provider.
 501         @param path: The path to the Unix socket file, used when connecting
 502         @param timeout: number of seconds to wait before assuming the
 503             connection has failed.
 504         @param checkPID: if True, check for a pid file to verify that a server
 505             is listening.
 506         """
 507         self._reactor = reactor
 508         self._path = path
 509         self._timeout = timeout
 510         self._checkPID = checkPID
 511
 512
 513     def connect(self, protocolFactory):
 514         """
 515         Implement L{IStreamClientEndpoint.connect} to connect via a
 516         UNIX Socket
 517         """
 518         try:
 519             wf = _WrappingFactory(protocolFactory)
 520             self._reactor.connectUNIX(
 521                 self._path, wf,
 522                 timeout=self._timeout,
 523                 checkPID=self._checkPID)
 524             return wf._onConnection
 525         except:
 526             return defer.fail()
 527
 528
 529
 530 class AdoptedStreamServerEndpoint(object):
 531     """
 532     An endpoint for listening on a file descriptor initialized outside of
 533     Twisted.
 534
 535     @ivar _used: A C{bool} indicating whether this endpoint has been used to
 536         listen with a factory yet.  C{True} if so.
 537     """
 538     _close = os.close
 539     _setNonBlocking = staticmethod(fdesc.setNonBlocking)
 540
 541     def __init__(self, reactor, fileno, addressFamily):
 542         """
 543         @param reactor: An L{IReactorSocket} provider.
 544
 545         @param fileno: An integer file descriptor corresponding to a listening
 546             I{SOCK_STREAM} socket.
 547
 548         @param addressFamily: The address family of the socket given by
 549             C{fileno}.
 550         """
 551         self.reactor = reactor
 552         self.fileno = fileno
 553         self.addressFamily = addressFamily
 554         self._used = False
 555
 556
 557     def listen(self, factory):
 558         """
 559         Implement L{IStreamServerEndpoint.listen} to start listening on, and
 560         then close, C{self._fileno}.
 561         """
 562         if self._used:
 563             return defer.fail(error.AlreadyListened())
 564         self._used = True
 565
 566         try:
 567             self._setNonBlocking(self.fileno)
 568             port = self.reactor.adoptStreamPort(
 569                 self.fileno, self.addressFamily, factory)
 570             self._close(self.fileno)
 571         except:
 572             return defer.fail()
 573         return defer.succeed(port)
 574
 575
 576
 577 def _parseTCP(factory, port, interface="", backlog=50):
 578     """
 579     Internal parser function for L{_parseServer} to convert the string
 580     arguments for a TCP(IPv4) stream endpoint into the structured arguments.
 581
 582     @param factory: the protocol factory being parsed, or C{None}.  (This was a
 583         leftover argument from when this code was in C{strports}, and is now
 584         mostly None and unused.)
 585
 586     @type factory: L{IProtocolFactory} or C{NoneType}
 587
 588     @param port: the integer port number to bind
 589     @type port: C{str}
 590
 591     @param interface: the interface IP to listen on
 592     @param backlog: the length of the listen queue
 593     @type backlog: C{str}
 594
 595     @return: a 2-tuple of (args, kwargs), describing  the parameters to
 596         L{IReactorTCP.listenTCP} (or, modulo argument 2, the factory, arguments
 597         to L{TCP4ServerEndpoint}.
 598     """
 599     return (int(port), factory), {'interface': interface,
 600                                   'backlog': int(backlog)}
 601
 602
 603
 604 def _parseUNIX(factory, address, mode='666', backlog=50, lockfile=True):
 605     """
 606     Internal parser function for L{_parseServer} to convert the string
 607     arguments for a UNIX (AF_UNIX/SOCK_STREAM) stream endpoint into the
 608     structured arguments.
 609
 610     @param factory: the protocol factory being parsed, or C{None}.  (This was a
 611         leftover argument from when this code was in C{strports}, and is now
 612         mostly None and unused.)
 613
 614     @type factory: L{IProtocolFactory} or C{NoneType}
 615
 616     @param address: the pathname of the unix socket
 617     @type address: C{str}
 618
 619     @param backlog: the length of the listen queue
 620     @type backlog: C{str}
 621
 622     @param lockfile: A string '0' or '1', mapping to True and False
 623         respectively.  See the C{wantPID} argument to C{listenUNIX}
 624
 625     @return: a 2-tuple of (args, kwargs), describing  the parameters to
 626         L{IReactorTCP.listenUNIX} (or, modulo argument 2, the factory,
 627         arguments to L{UNIXServerEndpoint}.
 628     """
 629     return (
 630         (address, factory),
 631         {'mode': int(mode, 8), 'backlog': int(backlog),
 632          'wantPID': bool(int(lockfile))})
 633
 634
 635
 636 def _parseSSL(factory, port, privateKey="server.pem", certKey=None,
 637               sslmethod=None, interface='', backlog=50):
 638     """
 639     Internal parser function for L{_parseServer} to convert the string
 640     arguments for an SSL (over TCP/IPv4) stream endpoint into the structured
 641     arguments.
 642
 643     @param factory: the protocol factory being parsed, or C{None}.  (This was a
 644         leftover argument from when this code was in C{strports}, and is now
 645         mostly None and unused.)
 646
 647     @type factory: L{IProtocolFactory} or C{NoneType}
 648
 649     @param port: the integer port number to bind
 650     @type port: C{str}
 651
 652     @param interface: the interface IP to listen on
 653     @param backlog: the length of the listen queue
 654     @type backlog: C{str}
 655
 656     @param privateKey: The file name of a PEM format private key file.
 657     @type privateKey: C{str}
 658
 659     @param certKey: The file name of a PEM format certificate file.
 660     @type certKey: C{str}
 661
 662     @param sslmethod: The string name of an SSL method, based on the name of a
 663         constant in C{OpenSSL.SSL}.  Must be one of: "SSLv23_METHOD",
 664         "SSLv2_METHOD", "SSLv3_METHOD", "TLSv1_METHOD".
 665     @type sslmethod: C{str}
 666
 667     @return: a 2-tuple of (args, kwargs), describing  the parameters to
 668         L{IReactorSSL.listenSSL} (or, modulo argument 2, the factory, arguments
 669         to L{SSL4ServerEndpoint}.
 670     """
 671     from twisted.internet import ssl
 672     if certKey is None:
 673         certKey = privateKey
 674     kw = {}
 675     if sslmethod is not None:
 676         kw['sslmethod'] = getattr(ssl.SSL, sslmethod)
 677     cf = ssl.DefaultOpenSSLContextFactory(privateKey, certKey, **kw)
 678     return ((int(port), factory, cf),
 679             {'interface': interface, 'backlog': int(backlog)})
 680
 681
 682 class _SystemdParser(object):
 683     """
 684     Stream server endpoint string parser for the I{systemd} endpoint type.
 685
 686     @ivar prefix: See L{IStreamClientEndpointStringParser.prefix}.
 687
 688     @ivar _sddaemon: A L{ListenFDs} instance used to translate an index into an
 689         actual file descriptor.
 690     """
 691     implements(IPlugin, IStreamServerEndpointStringParser)
 692
 693     _sddaemon = ListenFDs.fromEnvironment()
 694
 695     prefix = "systemd"
 696
 697     def _parseServer(self, reactor, domain, index):
 698         """
 699         Internal parser function for L{_parseServer} to convert the string
 700         arguments for a systemd server endpoint into structured arguments for
 701         L{AdoptedStreamServerEndpoint}.
 702
 703         @param reactor: An L{IReactorSocket} provider.
 704
 705         @param domain: The domain (or address family) of the socket inherited
 706             from systemd.  This is a string like C{"INET"} or C{"UNIX"}, ie the
 707             name of an address family from the L{socket} module, without the
 708             C{"AF_"} prefix.
 709         @type domain: C{str}
 710
 711         @param index: An offset into the list of file descriptors inherited from
 712             systemd.
 713         @type index: C{str}
 714
 715         @return: A two-tuple of parsed positional arguments and parsed keyword
 716             arguments (a tuple and a dictionary).  These can be used to
 717             construct a L{AdoptedStreamServerEndpoint}.
 718         """
 719         index = int(index)
 720         fileno = self._sddaemon.inheritedDescriptors()[index]
 721         addressFamily = getattr(socket, 'AF_' + domain)
 722         return AdoptedStreamServerEndpoint(reactor, fileno, addressFamily)
 723
 724
 725     def parseStreamServer(self, reactor, *args, **kwargs):
 726         # Delegate to another function with a sane signature.  This function has
 727         # an insane signature to trick zope.interface into believing the
 728         # interface is correctly implemented.
 729         return self._parseServer(reactor, *args, **kwargs)
 730
 731
 732
 733 _serverParsers = {"tcp": _parseTCP,
 734                   "unix": _parseUNIX,
 735                   "ssl": _parseSSL,
 736                   }
 737
 738 _OP, _STRING = range(2)
 739
 740 def _tokenize(description):
 741     """
 742     Tokenize a strports string and yield each token.
 743
 744     @param description: a string as described by L{serverFromString} or
 745         L{clientFromString}.
 746
 747     @return: an iterable of 2-tuples of (L{_OP} or L{_STRING}, string).  Tuples
 748         starting with L{_OP} will contain a second element of either ':' (i.e.
 749         'next parameter') or '=' (i.e. 'assign parameter value').  For example,
 750         the string 'hello:greet\=ing=world' would result in a generator
 751         yielding these values::
 752
 753             _STRING, 'hello'
 754             _OP, ':'
 755             _STRING, 'greet=ing'
 756             _OP, '='
 757             _STRING, 'world'
 758     """
 759     current = ''
 760     ops = ':='
 761     nextOps = {':': ':=', '=': ':'}
 762     description = iter(description)
 763     for n in description:
 764         if n in ops:
 765             yield _STRING, current
 766             yield _OP, n
 767             current = ''
 768             ops = nextOps[n]
 769         elif n == '\\':
 770             current += description.next()
 771         else:
 772             current += n
 773     yield _STRING, current
 774
 775
 776
 777 def _parse(description):
 778     """
 779     Convert a description string into a list of positional and keyword
 780     parameters, using logic vaguely like what Python does.
 781
 782     @param description: a string as described by L{serverFromString} or
 783         L{clientFromString}.
 784
 785     @return: a 2-tuple of C{(args, kwargs)}, where 'args' is a list of all
 786         ':'-separated C{str}s not containing an '=' and 'kwargs' is a map of
 787         all C{str}s which do contain an '='.  For example, the result of
 788         C{_parse('a:b:d=1:c')} would be C{(['a', 'b', 'c'], {'d': '1'})}.
 789     """
 790     args, kw = [], {}
 791     def add(sofar):
 792         if len(sofar) == 1:
 793             args.append(sofar[0])
 794         else:
 795             kw[sofar[0]] = sofar[1]
 796     sofar = ()
 797     for (type, value) in _tokenize(description):
 798         if type is _STRING:
 799             sofar += (value,)
 800         elif value == ':':
 801             add(sofar)
 802             sofar = ()
 803     add(sofar)
 804     return args, kw
 805
 806
 807 # Mappings from description "names" to endpoint constructors.
 808 _endpointServerFactories = {
 809     'TCP': TCP4ServerEndpoint,
 810     'SSL': SSL4ServerEndpoint,
 811     'UNIX': UNIXServerEndpoint,
 812     }
 813
 814 _endpointClientFactories = {
 815     'TCP': TCP4ClientEndpoint,
 816     'SSL': SSL4ClientEndpoint,
 817     'UNIX': UNIXClientEndpoint,
 818     }
 819
 820
 821 _NO_DEFAULT = object()
 822
 823 def _parseServer(description, factory, default=None):
 824     """
 825     Parse a stports description into a 2-tuple of arguments and keyword values.
 826
 827     @param description: A description in the format explained by
 828         L{serverFromString}.
 829     @type description: C{str}
 830
 831     @param factory: A 'factory' argument; this is left-over from
 832         twisted.application.strports, it's not really used.
 833     @type factory: L{IProtocolFactory} or L{None}
 834
 835     @param default: Deprecated argument, specifying the default parser mode to
 836         use for unqualified description strings (those which do not have a ':'
 837         and prefix).
 838     @type default: C{str} or C{NoneType}
 839
 840     @return: a 3-tuple of (plugin or name, arguments, keyword arguments)
 841     """
 842     args, kw = _parse(description)
 843     if not args or (len(args) == 1 and not kw):
 844         deprecationMessage = (
 845             "Unqualified strport description passed to 'service'."
 846             "Use qualified endpoint descriptions; for example, 'tcp:%s'."
 847             % (description,))
 848         if default is None:
 849             default = 'tcp'
 850             warnings.warn(
 851                 deprecationMessage, category=DeprecationWarning, stacklevel=4)
 852         elif default is _NO_DEFAULT:
 853             raise ValueError(deprecationMessage)
 854         # If the default has been otherwise specified, the user has already
 855         # been warned.
 856         args[0:0] = [default]
 857     endpointType = args[0]
 858     parser = _serverParsers.get(endpointType)
 859     if parser is None:
 860         for plugin in getPlugins(IStreamServerEndpointStringParser):
 861             if plugin.prefix == endpointType:
 862                 return (plugin, args[1:], kw)
 863         raise ValueError("Unknown endpoint type: '%s'" % (endpointType,))
 864     return (endpointType.upper(),) + parser(factory, *args[1:], **kw)
 865
 866
 867
 868 def _serverFromStringLegacy(reactor, description, default):
 869     """
 870     Underlying implementation of L{serverFromString} which avoids exposing the
 871     deprecated 'default' argument to anything but L{strports.service}.
 872     """
 873     nameOrPlugin, args, kw = _parseServer(description, None, default)
 874     if type(nameOrPlugin) is not str:
 875         plugin = nameOrPlugin
 876         return plugin.parseStreamServer(reactor, *args, **kw)
 877     else:
 878         name = nameOrPlugin
 879     # Chop out the factory.
 880     args = args[:1] + args[2:]
 881     return _endpointServerFactories[name](reactor, *args, **kw)
 882
 883
 884
 885 def serverFromString(reactor, description):
 886     """
 887     Construct a stream server endpoint from an endpoint description string.
 888
 889     The format for server endpoint descriptions is a simple string.  It is a
 890     prefix naming the type of endpoint, then a colon, then the arguments for
 891     that endpoint.
 892
 893     For example, you can call it like this to create an endpoint that will
 894     listen on TCP port 80::
 895
 896         serverFromString(reactor, "tcp:80")
 897
 898     Additional arguments may be specified as keywords, separated with colons.
 899     For example, you can specify the interface for a TCP server endpoint to
 900     bind to like this::
 901
 902         serverFromString(reactor, "tcp:80:interface=127.0.0.1")
 903
 904     SSL server endpoints may be specified with the 'ssl' prefix, and the
 905     private key and certificate files may be specified by the C{privateKey} and
 906     C{certKey} arguments::
 907
 908         serverFromString(reactor, "ssl:443:privateKey=key.pem:certKey=crt.pem")
 909
 910     If a private key file name (C{privateKey}) isn't provided, a "server.pem"
 911     file is assumed to exist which contains the private key. If the certificate
 912     file name (C{certKey}) isn't provided, the private key file is assumed to
 913     contain the certificate as well.
 914
 915     You may escape colons in arguments with a backslash, which you will need to
 916     use if you want to specify a full pathname argument on Windows::
 917
 918         serverFromString(reactor,
 919             "ssl:443:privateKey=C\\:/key.pem:certKey=C\\:/cert.pem")
 920
 921     finally, the 'unix' prefix may be used to specify a filesystem UNIX socket,
 922     optionally with a 'mode' argument to specify the mode of the socket file
 923     created by C{listen}::
 924
 925         serverFromString(reactor, "unix:/var/run/finger")
 926         serverFromString(reactor, "unix:/var/run/finger:mode=660")
 927
 928     This function is also extensible; new endpoint types may be registered as
 929     L{IStreamServerEndpointStringParser} plugins.  See that interface for more
 930     information.
 931
 932     @param reactor: The server endpoint will be constructed with this reactor.
 933
 934     @param description: The strports description to parse.
 935
 936     @return: A new endpoint which can be used to listen with the parameters
 937         given by by C{description}.
 938
 939     @rtype: L{IStreamServerEndpoint<twisted.internet.interfaces.IStreamServerEndpoint>}
 940
 941     @raise ValueError: when the 'description' string cannot be parsed.
 942
 943     @since: 10.2
 944     """
 945     return _serverFromStringLegacy(reactor, description, _NO_DEFAULT)
 946
 947
 948
 949 def quoteStringArgument(argument):
 950     """
 951     Quote an argument to L{serverFromString} and L{clientFromString}.  Since
 952     arguments are separated with colons and colons are escaped with
 953     backslashes, some care is necessary if, for example, you have a pathname,
 954     you may be tempted to interpolate into a string like this::
 955
 956         serverFromString("ssl:443:privateKey=%s" % (myPathName,))
 957
 958     This may appear to work, but will have portability issues (Windows
 959     pathnames, for example).  Usually you should just construct the appropriate
 960     endpoint type rather than interpolating strings, which in this case would
 961     be L{SSL4ServerEndpoint}.  There are some use-cases where you may need to
 962     generate such a string, though; for example, a tool to manipulate a
 963     configuration file which has strports descriptions in it.  To be correct in
 964     those cases, do this instead::
 965
 966         serverFromString("ssl:443:privateKey=%s" %
 967                          (quoteStringArgument(myPathName),))
 968
 969     @param argument: The part of the endpoint description string you want to
 970         pass through.
 971
 972     @type argument: C{str}
 973
 974     @return: The quoted argument.
 975
 976     @rtype: C{str}
 977     """
 978     return argument.replace('\\', '\\\\').replace(':', '\\:')
 979
 980
 981
 982 def _parseClientTCP(*args, **kwargs):
 983     """
 984     Perform any argument value coercion necessary for TCP client parameters.
 985
 986     Valid positional arguments to this function are host and port.
 987
 988     Valid keyword arguments to this function are all L{IReactorTCP.connectTCP}
 989     arguments.
 990
 991     @return: The coerced values as a C{dict}.
 992     """
 993
 994     if len(args) == 2:
 995         kwargs['port'] = int(args[1])
 996         kwargs['host'] = args[0]
 997     elif len(args) == 1:
 998         if 'host' in kwargs:
 999             kwargs['port'] = int(args[0])
1000         else:
1001             kwargs['host'] = args[0]
1002
1003     try:
1004         kwargs['port'] = int(kwargs['port'])
1005     except KeyError:
1006         pass
1007
1008     try:
1009         kwargs['timeout'] = int(kwargs['timeout'])
1010     except KeyError:
1011         pass
1012     return kwargs
1013
1014
1015
1016 def _loadCAsFromDir(directoryPath):
1017     """
1018     Load certificate-authority certificate objects in a given directory.
1019
1020     @param directoryPath: a L{FilePath} pointing at a directory to load .pem
1021         files from.
1022
1023     @return: a C{list} of L{OpenSSL.crypto.X509} objects.
1024     """
1025     from twisted.internet import ssl
1026
1027     caCerts = {}
1028     for child in directoryPath.children():
1029         if not child.basename().split('.')[-1].lower() == 'pem':
1030             continue
1031         try:
1032             data = child.getContent()
1033         except IOError:
1034             # Permission denied, corrupt disk, we don't care.
1035             continue
1036         try:
1037             theCert = ssl.Certificate.loadPEM(data)
1038         except ssl.SSL.Error:
1039             # Duplicate certificate, invalid certificate, etc.  We don't care.
1040             pass
1041         else:
1042             caCerts[theCert.digest()] = theCert.original
1043     return caCerts.values()
1044
1045
1046
1047 def _parseClientSSL(*args, **kwargs):
1048     """
1049     Perform any argument value coercion necessary for SSL client parameters.
1050
1051     Valid keyword arguments to this function are all L{IReactorSSL.connectSSL}
1052     arguments except for C{contextFactory}.  Instead, C{certKey} (the path name
1053     of the certificate file) C{privateKey} (the path name of the private key
1054     associated with the certificate) are accepted and used to construct a
1055     context factory.
1056
1057     Valid positional arguments to this function are host and port.
1058
1059     @param caCertsDir: The one parameter which is not part of
1060         L{IReactorSSL.connectSSL}'s signature, this is a path name used to
1061         construct a list of certificate authority certificates.  The directory
1062         will be scanned for files ending in C{.pem}, all of which will be
1063         considered valid certificate authorities for this connection.
1064
1065     @type caCertsDir: C{str}
1066
1067     @return: The coerced values as a C{dict}.
1068     """
1069     from twisted.internet import ssl
1070     kwargs = _parseClientTCP(*args, **kwargs)
1071     certKey = kwargs.pop('certKey', None)
1072     privateKey = kwargs.pop('privateKey', None)
1073     caCertsDir = kwargs.pop('caCertsDir', None)
1074     if certKey is not None:
1075         certx509 = ssl.Certificate.loadPEM(
1076             FilePath(certKey).getContent()).original
1077     else:
1078         certx509 = None
1079     if privateKey is not None:
1080         privateKey = ssl.PrivateCertificate.loadPEM(
1081             FilePath(privateKey).getContent()).privateKey.original
1082     else:
1083         privateKey = None
1084     if caCertsDir is not None:
1085         verify = True
1086         caCerts = _loadCAsFromDir(FilePath(caCertsDir))
1087     else:
1088         verify = False
1089         caCerts = None
1090     kwargs['sslContextFactory'] = ssl.CertificateOptions(
1091         method=ssl.SSL.SSLv23_METHOD,
1092         certificate=certx509,
1093         privateKey=privateKey,
1094         verify=verify,
1095         caCerts=caCerts
1096     )
1097     return kwargs
1098
1099
1100
1101 def _parseClientUNIX(*args, **kwargs):
1102     """
1103     Perform any argument value coercion necessary for UNIX client parameters.
1104
1105     Valid keyword arguments to this function are all L{IReactorUNIX.connectUNIX}
1106     keyword arguments except for C{checkPID}.  Instead, C{lockfile} is accepted
1107     and has the same meaning.  Also C{path} is used instead of C{address}.
1108
1109     Valid positional arguments to this function are C{path}.
1110
1111     @return: The coerced values as a C{dict}.
1112     """
1113     if len(args) == 1:
1114         kwargs['path'] = args[0]
1115
1116     try:
1117         kwargs['checkPID'] = bool(int(kwargs.pop('lockfile')))
1118     except KeyError:
1119         pass
1120     try:
1121         kwargs['timeout'] = int(kwargs['timeout'])
1122     except KeyError:
1123         pass
1124     return kwargs
1125
1126 _clientParsers = {
1127     'TCP': _parseClientTCP,
1128     'SSL': _parseClientSSL,
1129     'UNIX': _parseClientUNIX,
1130     }
1131
1132
1133
1134 def clientFromString(reactor, description):
1135     """
1136     Construct a client endpoint from a description string.
1137
1138     Client description strings are much like server description strings,
1139     although they take all of their arguments as keywords, aside from host and
1140     port.
1141
1142     You can create a TCP client endpoint with the 'host' and 'port' arguments,
1143     like so::
1144
1145         clientFromString(reactor, "tcp:host=www.example.com:port=80")
1146
1147     or, without specifying host and port keywords::
1148
1149         clientFromString(reactor, "tcp:www.example.com:80")
1150
1151     Or you can specify only one or the other, as in the following 2 examples::
1152
1153         clientFromString(reactor, "tcp:host=www.example.com:80")
1154         clientFromString(reactor, "tcp:www.example.com:port=80")
1155
1156     or an SSL client endpoint with those arguments, plus the arguments used by
1157     the server SSL, for a client certificate::
1158
1159         clientFromString(reactor, "ssl:web.example.com:443:"
1160                                   "privateKey=foo.pem:certKey=foo.pem")
1161
1162     to specify your certificate trust roots, you can identify a directory with
1163     PEM files in it with the C{caCertsDir} argument::
1164
1165         clientFromString(reactor, "ssl:host=web.example.com:port=443:"
1166                                   "caCertsDir=/etc/ssl/certs")
1167
1168     You can create a UNIX client endpoint with the 'path' argument and optional
1169     'lockfile' and 'timeout' arguments::
1170
1171         clientFromString(reactor, "unix:path=/var/foo/bar:lockfile=1:timeout=9")
1172
1173     or, with the path as a positional argument with or without optional
1174     arguments as in the following 2 examples::
1175
1176         clientFromString(reactor, "unix:/var/foo/bar")
1177         clientFromString(reactor, "unix:/var/foo/bar:lockfile=1:timeout=9")
1178
1179     This function is also extensible; new endpoint types may be registered as
1180     L{IStreamClientEndpointStringParser} plugins.  See that interface for more
1181     information.
1182
1183     @param reactor: The client endpoint will be constructed with this reactor.
1184
1185     @param description: The strports description to parse.
1186
1187     @return: A new endpoint which can be used to connect with the parameters
1188         given by by C{description}.
1189     @rtype: L{IStreamClientEndpoint<twisted.internet.interfaces.IStreamClientEndpoint>}
1190
1191     @since: 10.2
1192     """
1193     args, kwargs = _parse(description)
1194     aname = args.pop(0)
1195     name = aname.upper()
1196     for plugin in getPlugins(IStreamClientEndpointStringParser):
1197         if plugin.prefix.upper() == name:
1198             return plugin.parseStreamClient(*args, **kwargs)
1199     if name not in _clientParsers:
1200         raise ValueError("Unknown endpoint type: %r" % (aname,))
1201     kwargs = _clientParsers[name](*args, **kwargs)
1202     return _endpointClientFactories[name](reactor, **kwargs)