1
0

server.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519
  1. import json
  2. import logging
  3. from io import BytesIO
  4. from six import text_type
  5. import attr
  6. from zope.interface import implementer
  7. from twisted.internet import address, threads, udp
  8. from twisted.internet._resolver import SimpleResolverComplexifier
  9. from twisted.internet.defer import Deferred, fail, succeed
  10. from twisted.internet.error import DNSLookupError
  11. from twisted.internet.interfaces import (
  12. IReactorPluggableNameResolver,
  13. IReactorTCP,
  14. IResolverSimple,
  15. )
  16. from twisted.python.failure import Failure
  17. from twisted.test.proto_helpers import AccumulatingProtocol, MemoryReactorClock
  18. from twisted.web.http import unquote
  19. from twisted.web.http_headers import Headers
  20. from twisted.web.server import Site
  21. from synapse.http.site import SynapseRequest
  22. from synapse.util import Clock
  23. from tests.utils import setup_test_homeserver as _sth
  24. logger = logging.getLogger(__name__)
  25. class TimedOutException(Exception):
  26. """
  27. A web query timed out.
  28. """
  29. @attr.s
  30. class FakeChannel(object):
  31. """
  32. A fake Twisted Web Channel (the part that interfaces with the
  33. wire).
  34. """
  35. site = attr.ib(type=Site)
  36. _reactor = attr.ib()
  37. result = attr.ib(default=attr.Factory(dict))
  38. _producer = None
  39. @property
  40. def json_body(self):
  41. if not self.result:
  42. raise Exception("No result yet.")
  43. return json.loads(self.result["body"].decode("utf8"))
  44. @property
  45. def code(self):
  46. if not self.result:
  47. raise Exception("No result yet.")
  48. return int(self.result["code"])
  49. @property
  50. def headers(self):
  51. if not self.result:
  52. raise Exception("No result yet.")
  53. h = Headers()
  54. for i in self.result["headers"]:
  55. h.addRawHeader(*i)
  56. return h
  57. def writeHeaders(self, version, code, reason, headers):
  58. self.result["version"] = version
  59. self.result["code"] = code
  60. self.result["reason"] = reason
  61. self.result["headers"] = headers
  62. def write(self, content):
  63. assert isinstance(content, bytes), "Should be bytes! " + repr(content)
  64. if "body" not in self.result:
  65. self.result["body"] = b""
  66. self.result["body"] += content
  67. def registerProducer(self, producer, streaming):
  68. self._producer = producer
  69. self.producerStreaming = streaming
  70. def _produce():
  71. if self._producer:
  72. self._producer.resumeProducing()
  73. self._reactor.callLater(0.1, _produce)
  74. if not streaming:
  75. self._reactor.callLater(0.0, _produce)
  76. def unregisterProducer(self):
  77. if self._producer is None:
  78. return
  79. self._producer = None
  80. def requestDone(self, _self):
  81. self.result["done"] = True
  82. def getPeer(self):
  83. # We give an address so that getClientIP returns a non null entry,
  84. # causing us to record the MAU
  85. return address.IPv4Address("TCP", "127.0.0.1", 3423)
  86. def getHost(self):
  87. return None
  88. @property
  89. def transport(self):
  90. return self
  91. class FakeSite:
  92. """
  93. A fake Twisted Web Site, with mocks of the extra things that
  94. Synapse adds.
  95. """
  96. server_version_string = b"1"
  97. site_tag = "test"
  98. access_logger = logging.getLogger("synapse.access.http.fake")
  99. def make_request(
  100. reactor,
  101. method,
  102. path,
  103. content=b"",
  104. access_token=None,
  105. request=SynapseRequest,
  106. shorthand=True,
  107. federation_auth_origin=None,
  108. ):
  109. """
  110. Make a web request using the given method and path, feed it the
  111. content, and return the Request and the Channel underneath.
  112. Args:
  113. method (bytes/unicode): The HTTP request method ("verb").
  114. path (bytes/unicode): The HTTP path, suitably URL encoded (e.g.
  115. escaped UTF-8 & spaces and such).
  116. content (bytes or dict): The body of the request. JSON-encoded, if
  117. a dict.
  118. shorthand: Whether to try and be helpful and prefix the given URL
  119. with the usual REST API path, if it doesn't contain it.
  120. federation_auth_origin (bytes|None): if set to not-None, we will add a fake
  121. Authorization header pretenting to be the given server name.
  122. Returns:
  123. Tuple[synapse.http.site.SynapseRequest, channel]
  124. """
  125. if not isinstance(method, bytes):
  126. method = method.encode("ascii")
  127. if not isinstance(path, bytes):
  128. path = path.encode("ascii")
  129. # Decorate it to be the full path, if we're using shorthand
  130. if (
  131. shorthand
  132. and not path.startswith(b"/_matrix")
  133. and not path.startswith(b"/_synapse")
  134. ):
  135. path = b"/_matrix/client/r0/" + path
  136. path = path.replace(b"//", b"/")
  137. if not path.startswith(b"/"):
  138. path = b"/" + path
  139. if isinstance(content, text_type):
  140. content = content.encode("utf8")
  141. site = FakeSite()
  142. channel = FakeChannel(site, reactor)
  143. req = request(channel)
  144. req.process = lambda: b""
  145. req.content = BytesIO(content)
  146. req.postpath = list(map(unquote, path[1:].split(b"/")))
  147. if access_token:
  148. req.requestHeaders.addRawHeader(
  149. b"Authorization", b"Bearer " + access_token.encode("ascii")
  150. )
  151. if federation_auth_origin is not None:
  152. req.requestHeaders.addRawHeader(
  153. b"Authorization",
  154. b"X-Matrix origin=%s,key=,sig=" % (federation_auth_origin,),
  155. )
  156. if content:
  157. req.requestHeaders.addRawHeader(b"Content-Type", b"application/json")
  158. req.requestReceived(method, path, b"1.1")
  159. return req, channel
  160. def wait_until_result(clock, request, timeout=100):
  161. """
  162. Wait until the request is finished.
  163. """
  164. clock.run()
  165. x = 0
  166. while not request.finished:
  167. # If there's a producer, tell it to resume producing so we get content
  168. if request._channel._producer:
  169. request._channel._producer.resumeProducing()
  170. x += 1
  171. if x > timeout:
  172. raise TimedOutException("Timed out waiting for request to finish.")
  173. clock.advance(0.1)
  174. def render(request, resource, clock):
  175. request.render(resource)
  176. wait_until_result(clock, request)
  177. @implementer(IReactorPluggableNameResolver)
  178. class ThreadedMemoryReactorClock(MemoryReactorClock):
  179. """
  180. A MemoryReactorClock that supports callFromThread.
  181. """
  182. def __init__(self):
  183. self.threadpool = ThreadPool(self)
  184. self._udp = []
  185. lookups = self.lookups = {}
  186. @implementer(IResolverSimple)
  187. class FakeResolver(object):
  188. def getHostByName(self, name, timeout=None):
  189. if name not in lookups:
  190. return fail(DNSLookupError("OH NO: unknown %s" % (name,)))
  191. return succeed(lookups[name])
  192. self.nameResolver = SimpleResolverComplexifier(FakeResolver())
  193. super(ThreadedMemoryReactorClock, self).__init__()
  194. def listenUDP(self, port, protocol, interface="", maxPacketSize=8196):
  195. p = udp.Port(port, protocol, interface, maxPacketSize, self)
  196. p.startListening()
  197. self._udp.append(p)
  198. return p
  199. def callFromThread(self, callback, *args, **kwargs):
  200. """
  201. Make the callback fire in the next reactor iteration.
  202. """
  203. d = Deferred()
  204. d.addCallback(lambda x: callback(*args, **kwargs))
  205. self.callLater(0, d.callback, True)
  206. return d
  207. def getThreadPool(self):
  208. return self.threadpool
  209. class ThreadPool:
  210. """
  211. Threadless thread pool.
  212. """
  213. def __init__(self, reactor):
  214. self._reactor = reactor
  215. def start(self):
  216. pass
  217. def stop(self):
  218. pass
  219. def callInThreadWithCallback(self, onResult, function, *args, **kwargs):
  220. def _(res):
  221. if isinstance(res, Failure):
  222. onResult(False, res)
  223. else:
  224. onResult(True, res)
  225. d = Deferred()
  226. d.addCallback(lambda x: function(*args, **kwargs))
  227. d.addBoth(_)
  228. self._reactor.callLater(0, d.callback, True)
  229. return d
  230. def setup_test_homeserver(cleanup_func, *args, **kwargs):
  231. """
  232. Set up a synchronous test server, driven by the reactor used by
  233. the homeserver.
  234. """
  235. server = _sth(cleanup_func, *args, **kwargs)
  236. database = server.config.database.get_single_database()
  237. # Make the thread pool synchronous.
  238. clock = server.get_clock()
  239. for database in server.get_datastores().databases:
  240. pool = database._db_pool
  241. def runWithConnection(func, *args, **kwargs):
  242. return threads.deferToThreadPool(
  243. pool._reactor,
  244. pool.threadpool,
  245. pool._runWithConnection,
  246. func,
  247. *args,
  248. **kwargs
  249. )
  250. def runInteraction(interaction, *args, **kwargs):
  251. return threads.deferToThreadPool(
  252. pool._reactor,
  253. pool.threadpool,
  254. pool._runInteraction,
  255. interaction,
  256. *args,
  257. **kwargs
  258. )
  259. pool.runWithConnection = runWithConnection
  260. pool.runInteraction = runInteraction
  261. pool.threadpool = ThreadPool(clock._reactor)
  262. pool.running = True
  263. return server
  264. def get_clock():
  265. clock = ThreadedMemoryReactorClock()
  266. hs_clock = Clock(clock)
  267. return clock, hs_clock
  268. @attr.s(cmp=False)
  269. class FakeTransport(object):
  270. """
  271. A twisted.internet.interfaces.ITransport implementation which sends all its data
  272. straight into an IProtocol object: it exists to connect two IProtocols together.
  273. To use it, instantiate it with the receiving IProtocol, and then pass it to the
  274. sending IProtocol's makeConnection method:
  275. server = HTTPChannel()
  276. client.makeConnection(FakeTransport(server, self.reactor))
  277. If you want bidirectional communication, you'll need two instances.
  278. """
  279. other = attr.ib()
  280. """The Protocol object which will receive any data written to this transport.
  281. :type: twisted.internet.interfaces.IProtocol
  282. """
  283. _reactor = attr.ib()
  284. """Test reactor
  285. :type: twisted.internet.interfaces.IReactorTime
  286. """
  287. _protocol = attr.ib(default=None)
  288. """The Protocol which is producing data for this transport. Optional, but if set
  289. will get called back for connectionLost() notifications etc.
  290. """
  291. disconnecting = False
  292. disconnected = False
  293. connected = True
  294. buffer = attr.ib(default=b"")
  295. producer = attr.ib(default=None)
  296. autoflush = attr.ib(default=True)
  297. def getPeer(self):
  298. return None
  299. def getHost(self):
  300. return None
  301. def loseConnection(self, reason=None):
  302. if not self.disconnecting:
  303. logger.info("FakeTransport: loseConnection(%s)", reason)
  304. self.disconnecting = True
  305. if self._protocol:
  306. self._protocol.connectionLost(reason)
  307. # if we still have data to write, delay until that is done
  308. if self.buffer:
  309. logger.info(
  310. "FakeTransport: Delaying disconnect until buffer is flushed"
  311. )
  312. else:
  313. self.connected = False
  314. self.disconnected = True
  315. def abortConnection(self):
  316. logger.info("FakeTransport: abortConnection()")
  317. if not self.disconnecting:
  318. self.disconnecting = True
  319. if self._protocol:
  320. self._protocol.connectionLost(None)
  321. self.disconnected = True
  322. def pauseProducing(self):
  323. if not self.producer:
  324. return
  325. self.producer.pauseProducing()
  326. def resumeProducing(self):
  327. if not self.producer:
  328. return
  329. self.producer.resumeProducing()
  330. def unregisterProducer(self):
  331. if not self.producer:
  332. return
  333. self.producer = None
  334. def registerProducer(self, producer, streaming):
  335. self.producer = producer
  336. self.producerStreaming = streaming
  337. def _produce():
  338. d = self.producer.resumeProducing()
  339. d.addCallback(lambda x: self._reactor.callLater(0.1, _produce))
  340. if not streaming:
  341. self._reactor.callLater(0.0, _produce)
  342. def write(self, byt):
  343. if self.disconnecting:
  344. raise Exception("Writing to disconnecting FakeTransport")
  345. self.buffer = self.buffer + byt
  346. # always actually do the write asynchronously. Some protocols (notably the
  347. # TLSMemoryBIOProtocol) get very confused if a read comes back while they are
  348. # still doing a write. Doing a callLater here breaks the cycle.
  349. if self.autoflush:
  350. self._reactor.callLater(0.0, self.flush)
  351. def writeSequence(self, seq):
  352. for x in seq:
  353. self.write(x)
  354. def flush(self, maxbytes=None):
  355. if not self.buffer:
  356. # nothing to do. Don't write empty buffers: it upsets the
  357. # TLSMemoryBIOProtocol
  358. return
  359. if self.disconnected:
  360. return
  361. if getattr(self.other, "transport") is None:
  362. # the other has no transport yet; reschedule
  363. if self.autoflush:
  364. self._reactor.callLater(0.0, self.flush)
  365. return
  366. if maxbytes is not None:
  367. to_write = self.buffer[:maxbytes]
  368. else:
  369. to_write = self.buffer
  370. logger.info("%s->%s: %s", self._protocol, self.other, to_write)
  371. try:
  372. self.other.dataReceived(to_write)
  373. except Exception as e:
  374. logger.warning("Exception writing to protocol: %s", e)
  375. return
  376. self.buffer = self.buffer[len(to_write) :]
  377. if self.buffer and self.autoflush:
  378. self._reactor.callLater(0.0, self.flush)
  379. if not self.buffer and self.disconnecting:
  380. logger.info("FakeTransport: Buffer now empty, completing disconnect")
  381. self.disconnected = True
  382. def connect_client(reactor: IReactorTCP, client_id: int) -> AccumulatingProtocol:
  383. """
  384. Connect a client to a fake TCP transport.
  385. Args:
  386. reactor
  387. factory: The connecting factory to build.
  388. """
  389. factory = reactor.tcpClients[client_id][2]
  390. client = factory.buildProtocol(None)
  391. server = AccumulatingProtocol()
  392. server.makeConnection(FakeTransport(client, reactor))
  393. client.makeConnection(FakeTransport(server, reactor))
  394. reactor.tcpClients.pop(client_id)
  395. return client, server