# Copyright 2018 New Vector Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import re
from twisted.internet.defer import Deferred
from twisted.web.resource import Resource
from synapse.api.errors import Codes, RedirectException, SynapseError
from synapse.config.server import parse_listener_def
from synapse.http.server import DirectServeHtmlResource, JsonResource, OptionsResource
from synapse.http.site import SynapseSite
from synapse.logging.context import make_deferred_yieldable
from synapse.util import Clock
from tests import unittest
from tests.server import (
ThreadedMemoryReactorClock,
make_request,
render,
setup_test_homeserver,
)
class JsonResourceTests(unittest.TestCase):
def setUp(self):
self.reactor = ThreadedMemoryReactorClock()
self.hs_clock = Clock(self.reactor)
self.homeserver = setup_test_homeserver(
self.addCleanup, http_client=None, clock=self.hs_clock, reactor=self.reactor
)
def test_handler_for_request(self):
"""
JsonResource.handler_for_request gives correctly decoded URL args to
the callback, while Twisted will give the raw bytes of URL query
arguments.
"""
got_kwargs = {}
def _callback(request, **kwargs):
got_kwargs.update(kwargs)
return 200, kwargs
res = JsonResource(self.homeserver)
res.register_paths(
"GET",
[re.compile("^/_matrix/foo/(?P<room_id>[^/]*)$")],
_callback,
"test_servlet",
)
request, channel = make_request(
self.reactor, b"GET", b"/_matrix/foo/%E2%98%83?a=%E2%98%83"
)
render(request, res, self.reactor)
self.assertEqual(request.args, {b"a": ["\N{SNOWMAN}".encode("utf8")]})
self.assertEqual(got_kwargs, {"room_id": "\N{SNOWMAN}"})
def test_callback_direct_exception(self):
"""
If the web callback raises an uncaught exception, it will be translated
into a 500.
"""
def _callback(request, **kwargs):
raise Exception("boo")
res = JsonResource(self.homeserver)
res.register_paths(
"GET", [re.compile("^/_matrix/foo$")], _callback, "test_servlet"
)
request, channel = make_request(self.reactor, b"GET", b"/_matrix/foo")
render(request, res, self.reactor)
self.assertEqual(channel.result["code"], b"500")
def test_callback_indirect_exception(self):
"""
If the web callback raises an uncaught exception in a Deferred, it will
be translated into a 500.
"""
def _throw(*args):
raise Exception("boo")
def _callback(request, **kwargs):
d = Deferred()
d.addCallback(_throw)
self.reactor.callLater(1, d.callback, True)
return make_deferred_yieldable(d)
res = JsonResource(self.homeserver)
res.register_paths(
"GET", [re.compile("^/_matrix/foo$")], _callback, "test_servlet"
)
request, channel = make_request(self.reactor, b"GET", b"/_matrix/foo")
render(request, res, self.reactor)
self.assertEqual(channel.result["code"], b"500")
def test_callback_synapseerror(self):
"""
If the web callback raises a SynapseError, it returns the appropriate
status code and message set in it.
"""
def _callback(request, **kwargs):
raise SynapseError(403, "Forbidden!!one!", Codes.FORBIDDEN)
res = JsonResource(self.homeserver)
res.register_paths(
"GET", [re.compile("^/_matrix/foo$")], _callback, "test_servlet"
)
request, channel = make_request(self.reactor, b"GET", b"/_matrix/foo")
render(request, res, self.reactor)
self.assertEqual(channel.result["code"], b"403")
self.assertEqual(channel.json_body["error"], "Forbidden!!one!")
self.assertEqual(channel.json_body["errcode"], "M_FORBIDDEN")
def test_no_handler(self):
"""
If there is no handler to process the request, Synapse will return 400.
"""
def _callback(request, **kwargs):
"""
Not ever actually called!
"""
self.fail("shouldn't ever get here")
res = JsonResource(self.homeserver)
res.register_paths(
"GET", [re.compile("^/_matrix/foo$")], _callback, "test_servlet"
)
request, channel = make_request(self.reactor, b"GET", b"/_matrix/foobar")
render(request, res, self.reactor)
self.assertEqual(channel.result["code"], b"400")
self.assertEqual(channel.json_body["error"], "Unrecognized request")
self.assertEqual(channel.json_body["errcode"], "M_UNRECOGNIZED")
class OptionsResourceTests(unittest.TestCase):
def setUp(self):
self.reactor = ThreadedMemoryReactorClock()
class DummyResource(Resource):
isLeaf = True
def render(self, request):
return request.path
# Setup a resource with some children.
self.resource = OptionsResource()
self.resource.putChild(b"res", DummyResource())
def _make_request(self, method, path):
"""Create a request from the method/path and return a channel with the response."""
request, channel = make_request(self.reactor, method, path, shorthand=False)
request.prepath = [] # This doesn't get set properly by make_request.
# Create a site and query for the resource.
site = SynapseSite(
"test",
"site_tag",
parse_listener_def({"type": "http", "port": 0}),
self.resource,
"1.0",
)
request.site = site
resource = site.getResourceFor(request)
# Finally, render the resource and return the channel.
render(request, resource, self.reactor)
return channel
def test_unknown_options_request(self):
"""An OPTIONS requests to an unknown URL still returns 204 No Content."""
channel = self._make_request(b"OPTIONS", b"/foo/")
self.assertEqual(channel.result["code"], b"204")
self.assertNotIn("body", channel.result)
# Ensure the correct CORS headers have been added
self.assertTrue(
channel.headers.hasHeader(b"Access-Control-Allow-Origin"),
"has CORS Origin header",
)
self.assertTrue(
channel.headers.hasHeader(b"Access-Control-Allow-Methods"),
"has CORS Methods header",
)
self.assertTrue(
channel.headers.hasHeader(b"Access-Control-Allow-Headers"),
"has CORS Headers header",
)
def test_known_options_request(self):
"""An OPTIONS requests to an known URL still returns 204 No Content."""
channel = self._make_request(b"OPTIONS", b"/res/")
self.assertEqual(channel.result["code"], b"204")
self.assertNotIn("body", channel.result)
# Ensure the correct CORS headers have been added
self.assertTrue(
channel.headers.hasHeader(b"Access-Control-Allow-Origin"),
"has CORS Origin header",
)
self.assertTrue(
channel.headers.hasHeader(b"Access-Control-Allow-Methods"),
"has CORS Methods header",
)
self.assertTrue(
channel.headers.hasHeader(b"Access-Control-Allow-Headers"),
"has CORS Headers header",
)
def test_unknown_request(self):
"""A non-OPTIONS request to an unknown URL should 404."""
channel = self._make_request(b"GET", b"/foo/")
self.assertEqual(channel.result["code"], b"404")
def test_known_request(self):
"""A non-OPTIONS request to an known URL should query the proper resource."""
channel = self._make_request(b"GET", b"/res/")
self.assertEqual(channel.result["code"], b"200")
self.assertEqual(channel.result["body"], b"/res/")
class WrapHtmlRequestHandlerTests(unittest.TestCase):
class TestResource(DirectServeHtmlResource):
callback = None
async def _async_render_GET(self, request):
await self.callback(request)
def setUp(self):
self.reactor = ThreadedMemoryReactorClock()
def test_good_response(self):
def callback(request):
request.write(b"response")
request.finish()
res = WrapHtmlRequestHandlerTests.TestResource()
res.callback = callback
request, channel = make_request(self.reactor, b"GET", b"/path")
render(request, res, self.reactor)
self.assertEqual(channel.result["code"], b"200")
body = channel.result["body"]
self.assertEqual(body, b"response")
def test_redirect_exception(self):
"""
If the callback raises a RedirectException, it is turned into a 30x
with the right location.
"""
def callback(request, **kwargs):
raise RedirectException(b"/look/an/eagle", 301)
res = WrapHtmlRequestHandlerTests.TestResource()
res.callback = callback
request, channel = make_request(self.reactor, b"GET", b"/path")
render(request, res, self.reactor)
self.assertEqual(channel.result["code"], b"301")
headers = channel.result["headers"]
location_headers = [v for k, v in headers if k == b"Location"]
self.assertEqual(location_headers, [b"/look/an/eagle"])
def test_redirect_exception_with_cookie(self):
"""
If the callback raises a RedirectException which sets a cookie, that is
returned too
"""
def callback(request, **kwargs):
e = RedirectException(b"/no/over/there", 304)
e.cookies.append(b"session=yespls")
raise e
res = WrapHtmlRequestHandlerTests.TestResource()
res.callback = callback
request, channel = make_request(self.reactor, b"GET", b"/path")
render(request, res, self.reactor)
self.assertEqual(channel.result["code"], b"304")
headers = channel.result["headers"]
location_headers = [v for k, v in headers if k == b"Location"]
self.assertEqual(location_headers, [b"/no/over/there"])
cookies_headers = [v for k, v in headers if k == b"Set-Cookie"]
self.assertEqual(cookies_headers, [b"session=yespls"])