server.py 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642
  1. import json
  2. import logging
  3. from collections import deque
  4. from io import SEEK_END, BytesIO
  5. from typing import Callable, Dict, Iterable, MutableMapping, Optional, Tuple, Union
  6. import attr
  7. from typing_extensions import Deque
  8. from zope.interface import implementer
  9. from twisted.internet import address, threads, udp
  10. from twisted.internet._resolver import SimpleResolverComplexifier
  11. from twisted.internet.defer import Deferred, fail, succeed
  12. from twisted.internet.error import DNSLookupError
  13. from twisted.internet.interfaces import (
  14. IHostnameResolver,
  15. IProtocol,
  16. IPullProducer,
  17. IPushProducer,
  18. IReactorPluggableNameResolver,
  19. IResolverSimple,
  20. ITransport,
  21. )
  22. from twisted.python.failure import Failure
  23. from twisted.test.proto_helpers import AccumulatingProtocol, MemoryReactorClock
  24. from twisted.web.http_headers import Headers
  25. from twisted.web.resource import IResource
  26. from twisted.web.server import Site
  27. from synapse.http.site import SynapseRequest
  28. from synapse.util import Clock
  29. from tests.utils import setup_test_homeserver as _sth
  30. logger = logging.getLogger(__name__)
  31. class TimedOutException(Exception):
  32. """
  33. A web query timed out.
  34. """
  35. @attr.s
  36. class FakeChannel:
  37. """
  38. A fake Twisted Web Channel (the part that interfaces with the
  39. wire).
  40. """
  41. site = attr.ib(type=Union[Site, "FakeSite"])
  42. _reactor = attr.ib()
  43. result = attr.ib(type=dict, default=attr.Factory(dict))
  44. _ip = attr.ib(type=str, default="127.0.0.1")
  45. _producer = None # type: Optional[Union[IPullProducer, IPushProducer]]
  46. @property
  47. def json_body(self):
  48. return json.loads(self.text_body)
  49. @property
  50. def text_body(self) -> str:
  51. """The body of the result, utf-8-decoded.
  52. Raises an exception if the request has not yet completed.
  53. """
  54. if not self.is_finished:
  55. raise Exception("Request not yet completed")
  56. return self.result["body"].decode("utf8")
  57. def is_finished(self) -> bool:
  58. """check if the response has been completely received"""
  59. return self.result.get("done", False)
  60. @property
  61. def code(self):
  62. if not self.result:
  63. raise Exception("No result yet.")
  64. return int(self.result["code"])
  65. @property
  66. def headers(self) -> Headers:
  67. if not self.result:
  68. raise Exception("No result yet.")
  69. h = Headers()
  70. for i in self.result["headers"]:
  71. h.addRawHeader(*i)
  72. return h
  73. def writeHeaders(self, version, code, reason, headers):
  74. self.result["version"] = version
  75. self.result["code"] = code
  76. self.result["reason"] = reason
  77. self.result["headers"] = headers
  78. def write(self, content):
  79. assert isinstance(content, bytes), "Should be bytes! " + repr(content)
  80. if "body" not in self.result:
  81. self.result["body"] = b""
  82. self.result["body"] += content
  83. def registerProducer(self, producer, streaming):
  84. self._producer = producer
  85. self.producerStreaming = streaming
  86. def _produce():
  87. if self._producer:
  88. self._producer.resumeProducing()
  89. self._reactor.callLater(0.1, _produce)
  90. if not streaming:
  91. self._reactor.callLater(0.0, _produce)
  92. def unregisterProducer(self):
  93. if self._producer is None:
  94. return
  95. self._producer = None
  96. def requestDone(self, _self):
  97. self.result["done"] = True
  98. def getPeer(self):
  99. # We give an address so that getClientIP returns a non null entry,
  100. # causing us to record the MAU
  101. return address.IPv4Address("TCP", self._ip, 3423)
  102. def getHost(self):
  103. # this is called by Request.__init__ to configure Request.host.
  104. return address.IPv4Address("TCP", "127.0.0.1", 8888)
  105. def isSecure(self):
  106. return False
  107. @property
  108. def transport(self):
  109. return self
  110. def await_result(self, timeout_ms: int = 1000) -> None:
  111. """
  112. Wait until the request is finished.
  113. """
  114. end_time = self._reactor.seconds() + timeout_ms / 1000.0
  115. self._reactor.run()
  116. while not self.is_finished():
  117. # If there's a producer, tell it to resume producing so we get content
  118. if self._producer:
  119. self._producer.resumeProducing()
  120. if self._reactor.seconds() > end_time:
  121. raise TimedOutException("Timed out waiting for request to finish.")
  122. self._reactor.advance(0.1)
  123. def extract_cookies(self, cookies: MutableMapping[str, str]) -> None:
  124. """Process the contents of any Set-Cookie headers in the response
  125. Any cookines found are added to the given dict
  126. """
  127. headers = self.headers.getRawHeaders("Set-Cookie")
  128. if not headers:
  129. return
  130. for h in headers:
  131. parts = h.split(";")
  132. k, v = parts[0].split("=", maxsplit=1)
  133. cookies[k] = v
  134. class FakeSite:
  135. """
  136. A fake Twisted Web Site, with mocks of the extra things that
  137. Synapse adds.
  138. """
  139. server_version_string = b"1"
  140. site_tag = "test"
  141. access_logger = logging.getLogger("synapse.access.http.fake")
  142. def __init__(self, resource: IResource):
  143. """
  144. Args:
  145. resource: the resource to be used for rendering all requests
  146. """
  147. self._resource = resource
  148. def getResourceFor(self, request):
  149. return self._resource
  150. def make_request(
  151. reactor,
  152. site: Union[Site, FakeSite],
  153. method,
  154. path,
  155. content=b"",
  156. access_token=None,
  157. request=SynapseRequest,
  158. shorthand=True,
  159. federation_auth_origin=None,
  160. content_is_form=False,
  161. await_result: bool = True,
  162. custom_headers: Optional[
  163. Iterable[Tuple[Union[bytes, str], Union[bytes, str]]]
  164. ] = None,
  165. client_ip: str = "127.0.0.1",
  166. ) -> FakeChannel:
  167. """
  168. Make a web request using the given method, path and content, and render it
  169. Returns the fake Channel object which records the response to the request.
  170. Args:
  171. site: The twisted Site to use to render the request
  172. method (bytes/unicode): The HTTP request method ("verb").
  173. path (bytes/unicode): The HTTP path, suitably URL encoded (e.g.
  174. escaped UTF-8 & spaces and such).
  175. content (bytes or dict): The body of the request. JSON-encoded, if
  176. a dict.
  177. shorthand: Whether to try and be helpful and prefix the given URL
  178. with the usual REST API path, if it doesn't contain it.
  179. federation_auth_origin (bytes|None): if set to not-None, we will add a fake
  180. Authorization header pretenting to be the given server name.
  181. content_is_form: Whether the content is URL encoded form data. Adds the
  182. 'Content-Type': 'application/x-www-form-urlencoded' header.
  183. custom_headers: (name, value) pairs to add as request headers
  184. await_result: whether to wait for the request to complete rendering. If true,
  185. will pump the reactor until the the renderer tells the channel the request
  186. is finished.
  187. client_ip: The IP to use as the requesting IP. Useful for testing
  188. ratelimiting.
  189. Returns:
  190. channel
  191. """
  192. if not isinstance(method, bytes):
  193. method = method.encode("ascii")
  194. if not isinstance(path, bytes):
  195. path = path.encode("ascii")
  196. # Decorate it to be the full path, if we're using shorthand
  197. if (
  198. shorthand
  199. and not path.startswith(b"/_matrix")
  200. and not path.startswith(b"/_synapse")
  201. ):
  202. if path.startswith(b"/"):
  203. path = path[1:]
  204. path = b"/_matrix/client/r0/" + path
  205. if not path.startswith(b"/"):
  206. path = b"/" + path
  207. if isinstance(content, dict):
  208. content = json.dumps(content).encode("utf8")
  209. if isinstance(content, str):
  210. content = content.encode("utf8")
  211. channel = FakeChannel(site, reactor, ip=client_ip)
  212. req = request(channel)
  213. req.content = BytesIO(content)
  214. # Twisted expects to be at the end of the content when parsing the request.
  215. req.content.seek(SEEK_END)
  216. if access_token:
  217. req.requestHeaders.addRawHeader(
  218. b"Authorization", b"Bearer " + access_token.encode("ascii")
  219. )
  220. if federation_auth_origin is not None:
  221. req.requestHeaders.addRawHeader(
  222. b"Authorization",
  223. b"X-Matrix origin=%s,key=,sig=" % (federation_auth_origin,),
  224. )
  225. if content:
  226. if content_is_form:
  227. req.requestHeaders.addRawHeader(
  228. b"Content-Type", b"application/x-www-form-urlencoded"
  229. )
  230. else:
  231. # Assume the body is JSON
  232. req.requestHeaders.addRawHeader(b"Content-Type", b"application/json")
  233. if custom_headers:
  234. for k, v in custom_headers:
  235. req.requestHeaders.addRawHeader(k, v)
  236. req.parseCookies()
  237. req.requestReceived(method, path, b"1.1")
  238. if await_result:
  239. channel.await_result()
  240. return channel
  241. @implementer(IReactorPluggableNameResolver)
  242. class ThreadedMemoryReactorClock(MemoryReactorClock):
  243. """
  244. A MemoryReactorClock that supports callFromThread.
  245. """
  246. def __init__(self):
  247. self.threadpool = ThreadPool(self)
  248. self._tcp_callbacks = {}
  249. self._udp = []
  250. lookups = self.lookups = {} # type: Dict[str, str]
  251. self._thread_callbacks = deque() # type: Deque[Callable[[], None]]
  252. @implementer(IResolverSimple)
  253. class FakeResolver:
  254. def getHostByName(self, name, timeout=None):
  255. if name not in lookups:
  256. return fail(DNSLookupError("OH NO: unknown %s" % (name,)))
  257. return succeed(lookups[name])
  258. self.nameResolver = SimpleResolverComplexifier(FakeResolver())
  259. super().__init__()
  260. def installNameResolver(self, resolver: IHostnameResolver) -> IHostnameResolver:
  261. raise NotImplementedError()
  262. def listenUDP(self, port, protocol, interface="", maxPacketSize=8196):
  263. p = udp.Port(port, protocol, interface, maxPacketSize, self)
  264. p.startListening()
  265. self._udp.append(p)
  266. return p
  267. def callFromThread(self, callback, *args, **kwargs):
  268. """
  269. Make the callback fire in the next reactor iteration.
  270. """
  271. cb = lambda: callback(*args, **kwargs)
  272. # it's not safe to call callLater() here, so we append the callback to a
  273. # separate queue.
  274. self._thread_callbacks.append(cb)
  275. def getThreadPool(self):
  276. return self.threadpool
  277. def add_tcp_client_callback(self, host, port, callback):
  278. """Add a callback that will be invoked when we receive a connection
  279. attempt to the given IP/port using `connectTCP`.
  280. Note that the callback gets run before we return the connection to the
  281. client, which means callbacks cannot block while waiting for writes.
  282. """
  283. self._tcp_callbacks[(host, port)] = callback
  284. def connectTCP(self, host, port, factory, timeout=30, bindAddress=None):
  285. """Fake L{IReactorTCP.connectTCP}."""
  286. conn = super().connectTCP(
  287. host, port, factory, timeout=timeout, bindAddress=None
  288. )
  289. callback = self._tcp_callbacks.get((host, port))
  290. if callback:
  291. callback()
  292. return conn
  293. def advance(self, amount):
  294. # first advance our reactor's time, and run any "callLater" callbacks that
  295. # makes ready
  296. super().advance(amount)
  297. # now run any "callFromThread" callbacks
  298. while True:
  299. try:
  300. callback = self._thread_callbacks.popleft()
  301. except IndexError:
  302. break
  303. callback()
  304. # check for more "callLater" callbacks added by the thread callback
  305. # This isn't required in a regular reactor, but it ends up meaning that
  306. # our database queries can complete in a single call to `advance` [1] which
  307. # simplifies tests.
  308. #
  309. # [1]: we replace the threadpool backing the db connection pool with a
  310. # mock ThreadPool which doesn't really use threads; but we still use
  311. # reactor.callFromThread to feed results back from the db functions to the
  312. # main thread.
  313. super().advance(0)
  314. class ThreadPool:
  315. """
  316. Threadless thread pool.
  317. """
  318. def __init__(self, reactor):
  319. self._reactor = reactor
  320. def start(self):
  321. pass
  322. def stop(self):
  323. pass
  324. def callInThreadWithCallback(self, onResult, function, *args, **kwargs):
  325. def _(res):
  326. if isinstance(res, Failure):
  327. onResult(False, res)
  328. else:
  329. onResult(True, res)
  330. d = Deferred()
  331. d.addCallback(lambda x: function(*args, **kwargs))
  332. d.addBoth(_)
  333. self._reactor.callLater(0, d.callback, True)
  334. return d
  335. def setup_test_homeserver(cleanup_func, *args, **kwargs):
  336. """
  337. Set up a synchronous test server, driven by the reactor used by
  338. the homeserver.
  339. """
  340. server = _sth(cleanup_func, *args, **kwargs)
  341. # Make the thread pool synchronous.
  342. clock = server.get_clock()
  343. for database in server.get_datastores().databases:
  344. pool = database._db_pool
  345. def runWithConnection(func, *args, **kwargs):
  346. return threads.deferToThreadPool(
  347. pool._reactor,
  348. pool.threadpool,
  349. pool._runWithConnection,
  350. func,
  351. *args,
  352. **kwargs,
  353. )
  354. def runInteraction(interaction, *args, **kwargs):
  355. return threads.deferToThreadPool(
  356. pool._reactor,
  357. pool.threadpool,
  358. pool._runInteraction,
  359. interaction,
  360. *args,
  361. **kwargs,
  362. )
  363. pool.runWithConnection = runWithConnection
  364. pool.runInteraction = runInteraction
  365. pool.threadpool = ThreadPool(clock._reactor)
  366. pool.running = True
  367. # We've just changed the Databases to run DB transactions on the same
  368. # thread, so we need to disable the dedicated thread behaviour.
  369. server.get_datastores().main.USE_DEDICATED_DB_THREADS_FOR_EVENT_FETCHING = False
  370. return server
  371. def get_clock():
  372. clock = ThreadedMemoryReactorClock()
  373. hs_clock = Clock(clock)
  374. return clock, hs_clock
  375. @implementer(ITransport)
  376. @attr.s(cmp=False)
  377. class FakeTransport:
  378. """
  379. A twisted.internet.interfaces.ITransport implementation which sends all its data
  380. straight into an IProtocol object: it exists to connect two IProtocols together.
  381. To use it, instantiate it with the receiving IProtocol, and then pass it to the
  382. sending IProtocol's makeConnection method:
  383. server = HTTPChannel()
  384. client.makeConnection(FakeTransport(server, self.reactor))
  385. If you want bidirectional communication, you'll need two instances.
  386. """
  387. other = attr.ib()
  388. """The Protocol object which will receive any data written to this transport.
  389. :type: twisted.internet.interfaces.IProtocol
  390. """
  391. _reactor = attr.ib()
  392. """Test reactor
  393. :type: twisted.internet.interfaces.IReactorTime
  394. """
  395. _protocol = attr.ib(default=None)
  396. """The Protocol which is producing data for this transport. Optional, but if set
  397. will get called back for connectionLost() notifications etc.
  398. """
  399. disconnecting = False
  400. disconnected = False
  401. connected = True
  402. buffer = attr.ib(default=b"")
  403. producer = attr.ib(default=None)
  404. autoflush = attr.ib(default=True)
  405. def getPeer(self):
  406. return None
  407. def getHost(self):
  408. return None
  409. def loseConnection(self, reason=None):
  410. if not self.disconnecting:
  411. logger.info("FakeTransport: loseConnection(%s)", reason)
  412. self.disconnecting = True
  413. if self._protocol:
  414. self._protocol.connectionLost(reason)
  415. # if we still have data to write, delay until that is done
  416. if self.buffer:
  417. logger.info(
  418. "FakeTransport: Delaying disconnect until buffer is flushed"
  419. )
  420. else:
  421. self.connected = False
  422. self.disconnected = True
  423. def abortConnection(self):
  424. logger.info("FakeTransport: abortConnection()")
  425. if not self.disconnecting:
  426. self.disconnecting = True
  427. if self._protocol:
  428. self._protocol.connectionLost(None)
  429. self.disconnected = True
  430. def pauseProducing(self):
  431. if not self.producer:
  432. return
  433. self.producer.pauseProducing()
  434. def resumeProducing(self):
  435. if not self.producer:
  436. return
  437. self.producer.resumeProducing()
  438. def unregisterProducer(self):
  439. if not self.producer:
  440. return
  441. self.producer = None
  442. def registerProducer(self, producer, streaming):
  443. self.producer = producer
  444. self.producerStreaming = streaming
  445. def _produce():
  446. d = self.producer.resumeProducing()
  447. d.addCallback(lambda x: self._reactor.callLater(0.1, _produce))
  448. if not streaming:
  449. self._reactor.callLater(0.0, _produce)
  450. def write(self, byt):
  451. if self.disconnecting:
  452. raise Exception("Writing to disconnecting FakeTransport")
  453. self.buffer = self.buffer + byt
  454. # always actually do the write asynchronously. Some protocols (notably the
  455. # TLSMemoryBIOProtocol) get very confused if a read comes back while they are
  456. # still doing a write. Doing a callLater here breaks the cycle.
  457. if self.autoflush:
  458. self._reactor.callLater(0.0, self.flush)
  459. def writeSequence(self, seq):
  460. for x in seq:
  461. self.write(x)
  462. def flush(self, maxbytes=None):
  463. if not self.buffer:
  464. # nothing to do. Don't write empty buffers: it upsets the
  465. # TLSMemoryBIOProtocol
  466. return
  467. if self.disconnected:
  468. return
  469. if maxbytes is not None:
  470. to_write = self.buffer[:maxbytes]
  471. else:
  472. to_write = self.buffer
  473. logger.info("%s->%s: %s", self._protocol, self.other, to_write)
  474. try:
  475. self.other.dataReceived(to_write)
  476. except Exception as e:
  477. logger.exception("Exception writing to protocol: %s", e)
  478. return
  479. self.buffer = self.buffer[len(to_write) :]
  480. if self.buffer and self.autoflush:
  481. self._reactor.callLater(0.0, self.flush)
  482. if not self.buffer and self.disconnecting:
  483. logger.info("FakeTransport: Buffer now empty, completing disconnect")
  484. self.disconnected = True
  485. def connect_client(
  486. reactor: ThreadedMemoryReactorClock, client_id: int
  487. ) -> Tuple[IProtocol, AccumulatingProtocol]:
  488. """
  489. Connect a client to a fake TCP transport.
  490. Args:
  491. reactor
  492. factory: The connecting factory to build.
  493. """
  494. factory = reactor.tcpClients.pop(client_id)[2]
  495. client = factory.buildProtocol(None)
  496. server = AccumulatingProtocol()
  497. server.makeConnection(FakeTransport(client, reactor))
  498. client.makeConnection(FakeTransport(server, reactor))
  499. return client, server