Funktionierender Prototyp des Serious Games zur Vermittlung von Wissen zu Software-Engineering-Arbeitsmodellen.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

component.py 15KB

1 year ago
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380
  1. ###############################################################################
  2. #
  3. # The MIT License (MIT)
  4. #
  5. # Copyright (c) typedef int GmbH
  6. #
  7. # Permission is hereby granted, free of charge, to any person obtaining a copy
  8. # of this software and associated documentation files (the "Software"), to deal
  9. # in the Software without restriction, including without limitation the rights
  10. # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  11. # copies of the Software, and to permit persons to whom the Software is
  12. # furnished to do so, subject to the following conditions:
  13. #
  14. # The above copyright notice and this permission notice shall be included in
  15. # all copies or substantial portions of the Software.
  16. #
  17. # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  18. # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  19. # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  20. # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  21. # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  22. # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  23. # THE SOFTWARE.
  24. #
  25. ###############################################################################
  26. from functools import wraps
  27. from typing import List
  28. from twisted.internet.interfaces import IStreamClientEndpoint
  29. from twisted.internet.endpoints import UNIXClientEndpoint
  30. from twisted.internet.endpoints import TCP4ClientEndpoint
  31. from twisted.python.failure import Failure
  32. from twisted.internet.error import ReactorNotRunning
  33. try:
  34. _TLS = True
  35. from twisted.internet.endpoints import SSL4ClientEndpoint
  36. from twisted.internet.ssl import optionsForClientTLS, CertificateOptions, Certificate
  37. from twisted.internet.interfaces import IOpenSSLClientConnectionCreator
  38. from OpenSSL import SSL
  39. except ImportError:
  40. _TLS = False
  41. # there's no optionsForClientTLS in older Twisteds or we might be
  42. # missing OpenSSL entirely.
  43. import txaio
  44. from autobahn.twisted.websocket import WampWebSocketClientFactory
  45. from autobahn.twisted.rawsocket import WampRawSocketClientFactory
  46. from autobahn.wamp import component
  47. from autobahn.twisted.wamp import Session
  48. from autobahn.wamp.serializer import create_transport_serializers, create_transport_serializer
  49. __all__ = ('Component', 'run')
  50. def _unique_list(seq):
  51. """
  52. Return a list with unique elements from sequence, preserving order.
  53. """
  54. seen = set()
  55. return [x for x in seq if x not in seen and not seen.add(x)]
  56. def _camel_case_from_snake_case(snake):
  57. parts = snake.split('_')
  58. return parts[0] + ''.join(s.capitalize() for s in parts[1:])
  59. def _create_transport_factory(reactor, transport, session_factory):
  60. """
  61. Create a WAMP-over-XXX transport factory.
  62. """
  63. if transport.type == 'websocket':
  64. serializers = create_transport_serializers(transport)
  65. factory = WampWebSocketClientFactory(
  66. session_factory,
  67. url=transport.url,
  68. serializers=serializers,
  69. proxy=transport.proxy, # either None or a dict with host, port
  70. )
  71. elif transport.type == 'rawsocket':
  72. serializer = create_transport_serializer(transport.serializers[0])
  73. factory = WampRawSocketClientFactory(session_factory, serializer=serializer)
  74. else:
  75. assert(False), 'should not arrive here'
  76. # set the options one at a time so we can give user better feedback
  77. for k, v in transport.options.items():
  78. try:
  79. factory.setProtocolOptions(**{k: v})
  80. except (TypeError, KeyError):
  81. # this allows us to document options as snake_case
  82. # until everything internally is upgraded from
  83. # camelCase
  84. try:
  85. factory.setProtocolOptions(
  86. **{_camel_case_from_snake_case(k): v}
  87. )
  88. except (TypeError, KeyError):
  89. raise ValueError(
  90. "Unknown {} transport option: {}={}".format(transport.type, k, v)
  91. )
  92. return factory
  93. def _create_transport_endpoint(reactor, endpoint_config):
  94. """
  95. Create a Twisted client endpoint for a WAMP-over-XXX transport.
  96. """
  97. if IStreamClientEndpoint.providedBy(endpoint_config):
  98. endpoint = IStreamClientEndpoint(endpoint_config)
  99. else:
  100. # create a connecting TCP socket
  101. if endpoint_config['type'] == 'tcp':
  102. version = endpoint_config.get('version', 4)
  103. if version not in [4, 6]:
  104. raise ValueError('invalid IP version {} in client endpoint configuration'.format(version))
  105. host = endpoint_config['host']
  106. if type(host) != str:
  107. raise ValueError('invalid type {} for host in client endpoint configuration'.format(type(host)))
  108. port = endpoint_config['port']
  109. if type(port) != int:
  110. raise ValueError('invalid type {} for port in client endpoint configuration'.format(type(port)))
  111. timeout = endpoint_config.get('timeout', 10) # in seconds
  112. if type(timeout) != int:
  113. raise ValueError('invalid type {} for timeout in client endpoint configuration'.format(type(timeout)))
  114. tls = endpoint_config.get('tls', None)
  115. # create a TLS enabled connecting TCP socket
  116. if tls:
  117. if not _TLS:
  118. raise RuntimeError('TLS configured in transport, but TLS support is not installed (eg OpenSSL?)')
  119. # FIXME: create TLS context from configuration
  120. if IOpenSSLClientConnectionCreator.providedBy(tls):
  121. # eg created from twisted.internet.ssl.optionsForClientTLS()
  122. context = IOpenSSLClientConnectionCreator(tls)
  123. elif isinstance(tls, dict):
  124. for k in tls.keys():
  125. if k not in ["hostname", "trust_root"]:
  126. raise ValueError("Invalid key '{}' in 'tls' config".format(k))
  127. hostname = tls.get('hostname', host)
  128. if type(hostname) != str:
  129. raise ValueError('invalid type {} for hostname in TLS client endpoint configuration'.format(hostname))
  130. trust_root = None
  131. cert_fname = tls.get("trust_root", None)
  132. if cert_fname is not None:
  133. trust_root = Certificate.loadPEM(open(cert_fname, 'r').read())
  134. context = optionsForClientTLS(hostname, trustRoot=trust_root)
  135. elif isinstance(tls, CertificateOptions):
  136. context = tls
  137. elif tls is True:
  138. context = optionsForClientTLS(host)
  139. else:
  140. raise RuntimeError('unknown type {} for "tls" configuration in transport'.format(type(tls)))
  141. if version == 4:
  142. endpoint = SSL4ClientEndpoint(reactor, host, port, context, timeout=timeout)
  143. elif version == 6:
  144. # there is no SSL6ClientEndpoint!
  145. raise RuntimeError('TLS on IPv6 not implemented')
  146. else:
  147. assert(False), 'should not arrive here'
  148. # create a non-TLS connecting TCP socket
  149. else:
  150. if host.endswith(".onion"):
  151. # hmm, can't log here?
  152. # self.log.info("{host} appears to be a Tor endpoint", host=host)
  153. try:
  154. import txtorcon
  155. endpoint = txtorcon.TorClientEndpoint(host, port)
  156. except ImportError:
  157. raise RuntimeError(
  158. "{} appears to be a Tor Onion service, but txtorcon is not installed".format(
  159. host,
  160. )
  161. )
  162. elif version == 4:
  163. endpoint = TCP4ClientEndpoint(reactor, host, port, timeout=timeout)
  164. elif version == 6:
  165. try:
  166. from twisted.internet.endpoints import TCP6ClientEndpoint
  167. except ImportError:
  168. raise RuntimeError('IPv6 is not supported (please upgrade Twisted)')
  169. endpoint = TCP6ClientEndpoint(reactor, host, port, timeout=timeout)
  170. else:
  171. assert(False), 'should not arrive here'
  172. # create a connecting Unix domain socket
  173. elif endpoint_config['type'] == 'unix':
  174. path = endpoint_config['path']
  175. timeout = int(endpoint_config.get('timeout', 10)) # in seconds
  176. endpoint = UNIXClientEndpoint(reactor, path, timeout=timeout)
  177. else:
  178. assert(False), 'should not arrive here'
  179. return endpoint
  180. class Component(component.Component):
  181. """
  182. A component establishes a transport and attached a session
  183. to a realm using the transport for communication.
  184. The transports a component tries to use can be configured,
  185. as well as the auto-reconnect strategy.
  186. """
  187. log = txaio.make_logger()
  188. session_factory = Session
  189. """
  190. The factory of the session we will instantiate.
  191. """
  192. def _is_ssl_error(self, e):
  193. """
  194. Internal helper.
  195. This is so we can just return False if we didn't import any
  196. TLS/SSL libraries. Otherwise, returns True if this is an
  197. OpenSSL.SSL.Error
  198. """
  199. if _TLS:
  200. return isinstance(e, SSL.Error)
  201. return False
  202. def _check_native_endpoint(self, endpoint):
  203. if IStreamClientEndpoint.providedBy(endpoint):
  204. pass
  205. elif isinstance(endpoint, dict):
  206. if 'tls' in endpoint:
  207. tls = endpoint['tls']
  208. if isinstance(tls, (dict, bool)):
  209. pass
  210. elif IOpenSSLClientConnectionCreator.providedBy(tls):
  211. pass
  212. elif isinstance(tls, CertificateOptions):
  213. pass
  214. else:
  215. raise ValueError(
  216. "'tls' configuration must be a dict, CertificateOptions or"
  217. " IOpenSSLClientConnectionCreator provider"
  218. )
  219. else:
  220. raise ValueError(
  221. "'endpoint' configuration must be a dict or IStreamClientEndpoint"
  222. " provider"
  223. )
  224. def _connect_transport(self, reactor, transport, session_factory, done):
  225. """
  226. Create and connect a WAMP-over-XXX transport.
  227. :param done: is a Deferred/Future from the parent which we
  228. should signal upon error if it is not done yet (XXX maybe an
  229. "on_error" callable instead?)
  230. """
  231. transport_factory = _create_transport_factory(reactor, transport, session_factory)
  232. if transport.proxy:
  233. transport_endpoint = _create_transport_endpoint(
  234. reactor,
  235. {
  236. "type": "tcp",
  237. "host": transport.proxy["host"],
  238. "port": transport.proxy["port"],
  239. }
  240. )
  241. else:
  242. transport_endpoint = _create_transport_endpoint(reactor, transport.endpoint)
  243. d = transport_endpoint.connect(transport_factory)
  244. def on_connect_success(proto):
  245. # if e.g. an SSL handshake fails, we will have
  246. # successfully connected (i.e. get here) but need to
  247. # 'listen' for the "connectionLost" from the underlying
  248. # protocol in case of handshake failure .. so we wrap
  249. # it. Also, we don't increment transport.success_count
  250. # here on purpose (because we might not succeed).
  251. orig = proto.connectionLost
  252. @wraps(orig)
  253. def lost(fail):
  254. rtn = orig(fail)
  255. if not txaio.is_called(done):
  256. txaio.reject(done, fail)
  257. return rtn
  258. proto.connectionLost = lost
  259. def on_connect_failure(err):
  260. transport.connect_failures += 1
  261. # failed to establish a connection in the first place
  262. txaio.reject(done, err)
  263. txaio.add_callbacks(d, on_connect_success, None)
  264. txaio.add_callbacks(d, None, on_connect_failure)
  265. return d
  266. def start(self, reactor=None):
  267. """
  268. This starts the Component, which means it will start connecting
  269. (and re-connecting) to its configured transports. A Component
  270. runs until it is "done", which means one of:
  271. - There was a "main" function defined, and it completed successfully;
  272. - Something called ``.leave()`` on our session, and we left successfully;
  273. - ``.stop()`` was called, and completed successfully;
  274. - none of our transports were able to connect successfully (failure);
  275. :returns: a Deferred that fires (with ``None``) when we are
  276. "done" or with a Failure if something went wrong.
  277. """
  278. if reactor is None:
  279. self.log.warn("Using default reactor")
  280. from twisted.internet import reactor
  281. return self._start(loop=reactor)
  282. def run(components: List[Component], log_level: str = 'info', stop_at_close: bool = True):
  283. """
  284. High-level API to run a series of components.
  285. This will only return once all the components have stopped
  286. (including, possibly, after all re-connections have failed if you
  287. have re-connections enabled). Under the hood, this calls
  288. :meth:`twisted.internet.reactor.run` -- if you wish to manage the
  289. reactor loop yourself, use the
  290. :meth:`autobahn.twisted.component.Component.start` method to start
  291. each component yourself.
  292. :param components: the Component(s) you wish to run
  293. :param log_level: a valid log-level (or None to avoid calling start_logging)
  294. :param stop_at_close: Flag to control whether to stop the reactor when done.
  295. """
  296. # only for Twisted > 12
  297. # ...so this isn't in all Twisted versions we test against -- need
  298. # to do "something else" if we can't import .. :/ (or drop some
  299. # support)
  300. from twisted.internet.task import react
  301. # actually, should we even let people "not start" the logging? I'm
  302. # not sure that's wise... (double-check: if they already called
  303. # txaio.start_logging() what happens if we call it again?)
  304. if log_level is not None:
  305. txaio.start_logging(level=log_level)
  306. log = txaio.make_logger()
  307. if stop_at_close:
  308. def done_callback(reactor, arg):
  309. if isinstance(arg, Failure):
  310. log.error('Something went wrong: {log_failure}', failure=arg)
  311. try:
  312. log.warn('Stopping reactor ..')
  313. reactor.stop()
  314. except ReactorNotRunning:
  315. pass
  316. else:
  317. done_callback = None
  318. react(component._run, (components, done_callback))