def from_request(cls, request, raise_invalid_params=True,
                     default_limit=None):
        direction = parse_string(request, "dir", default='f', allowed_values=['f', 'b'])

        from_tok = parse_string(request, "from")
        to_tok = parse_string(request, "to")

        try:
            if from_tok == "END":
                from_tok = None  # For backwards compat.
            elif from_tok:
                from_tok = StreamToken.from_string(from_tok)
        except Exception:
            raise SynapseError(400, "'from' paramater is invalid")

        try:
            if to_tok:
                to_tok = StreamToken.from_string(to_tok)
        except Exception:
            raise SynapseError(400, "'to' paramater is invalid")

        limit = parse_integer(request, "limit", default=default_limit)

        if limit and limit < 0:
            raise SynapseError(400, "Limit must be 0 or above")

        try:
            return PaginationConfig(from_tok, to_tok, direction, limit)
        except Exception:
            logger.exception("Failed to create pagination config")
            raise SynapseError(400, "Invalid request.")

    def _async_render_GET(self, request):
        server_name, media_id, _ = parse_media_id(request)
        width = parse_integer(request, "width")
        height = parse_integer(request, "height")
        method = parse_string(request, "method", "scale")
        m_type = parse_string(request, "type", "image/png")

        if server_name == self.server_name:
            if self.dynamic_thumbnails:
                yield self._select_or_generate_local_thumbnail(
                    request, media_id, width, height, method, m_type
                )
            else:
                yield self._respond_local_thumbnail(
                    request, media_id, width, height, method, m_type
                )
        else:
            if self.dynamic_thumbnails:
                yield self._select_or_generate_remote_thumbnail(
                    request, server_name, media_id,
                    width, height, method, m_type
                )
            else:
                yield self._respond_remote_thumbnail(
                    request, server_name, media_id,
                    width, height, method, m_type
                )

    def on_GET(self, request):
        server = parse_string(request, "server", default=None)

        try:
            yield self.auth.get_user_by_req(request, allow_guest=True)
        except AuthError as e:
            # We allow people to not be authed if they're just looking at our
            # room list, but require auth when we proxy the request.
            # In both cases we call the auth function, as that has the side
            # effect of logging who issued this request if an access token was
            # provided.
            if server:
                raise e
            else:
                pass

        limit = parse_integer(request, "limit", 0)
        since_token = parse_string(request, "since", None)

        handler = self.hs.get_room_list_handler()
        if server:
            data = yield handler.get_remote_public_room_list(
                server,
                limit=limit,
                since_token=since_token,
            )
        else:
            data = yield handler.get_local_public_room_list(
                limit=limit,
                since_token=since_token,
            )

        defer.returnValue((200, data))

    def on_GET(self, request):
        requester = yield self.auth.get_user_by_req(request, rights="delete_pusher")
        user = requester.user

        app_id = parse_string(request, "app_id", required=True)
        pushkey = parse_string(request, "pushkey", required=True)

        try:
            yield self.pusher_pool.remove_pusher(
                app_id=app_id,
                pushkey=pushkey,
                user_id=user.to_string(),
            )
        except StoreError as se:
            if se.code != 404:
                # This is fine: they're already unsubscribed
                raise

        self.notifier.on_new_replication_data()

        request.setResponseCode(200)
        request.setHeader(b"Content-Type", b"text/html; charset=utf-8")
        request.setHeader(b"Content-Length", b"%d" % (
            len(PushersRemoveRestServlet.SUCCESS_HTML),
        ))
        request.write(PushersRemoveRestServlet.SUCCESS_HTML)
        finish_request(request)
        defer.returnValue(None)

    def _async_render_POST(self, request):
        resp_bytes = parse_string(request, 'SAMLResponse', required=True)
        relay_state = parse_string(request, 'RelayState', required=True)

        try:
            saml2_auth = self._saml_client.parse_authn_request_response(
                resp_bytes, saml2.BINDING_HTTP_POST,
            )
        except Exception as e:
            logger.warning("Exception parsing SAML2 response", exc_info=1)
            raise CodeMessageException(
                400, "Unable to parse SAML2 response: %s" % (e,),
            )

        if saml2_auth.not_signed:
            raise CodeMessageException(400, "SAML2 response was not signed")

        if "uid" not in saml2_auth.ava:
            raise CodeMessageException(400, "uid not in SAML2 response")

        username = saml2_auth.ava["uid"][0]

        displayName = saml2_auth.ava.get("displayName", [None])[0]
        return self._sso_auth_handler.on_successful_auth(
            username, request, relay_state,
            user_display_name=displayName,
        )

    def on_GET(self, request, room_id, parent_id, relation_type=None, event_type=None):
        requester = yield self.auth.get_user_by_req(request, allow_guest=True)

        yield self.auth.check_in_room_or_world_readable(
            room_id, requester.user.to_string()
        )

        # This checks that a) the event exists and b) the user is allowed to
        # view it.
        yield self.event_handler.get_event(requester.user, room_id, parent_id)

        if relation_type not in (RelationTypes.ANNOTATION, None):
            raise SynapseError(400, "Relation type must be 'annotation'")

        limit = parse_integer(request, "limit", default=5)
        from_token = parse_string(request, "from")
        to_token = parse_string(request, "to")

        if from_token:
            from_token = AggregationPaginationToken.from_string(from_token)

        if to_token:
            to_token = AggregationPaginationToken.from_string(to_token)

        res = yield self.store.get_aggregation_groups_for_event(
            event_id=parent_id,
            event_type=event_type,
            limit=limit,
            from_token=from_token,
            to_token=to_token,
        )

        defer.returnValue((200, res.to_dict()))

    def _async_render_POST(self, request):
        """
        Args:
            request (twisted.web.http.Request):
        """
        version = parse_string(request, "v", required=True)
        username = parse_string(request, "u", required=True)
        userhmac = parse_string(request, "h", required=True, encoding=None)

        self._check_hash(username, userhmac)

        if username.startswith('@'):
            qualified_user_id = username
        else:
            qualified_user_id = UserID(username, self.hs.hostname).to_string()

        try:
            yield self.store.user_set_consent_version(qualified_user_id, version)
        except StoreError as e:
            if e.code != 404:
                raise
            raise NotFoundError("Unknown user")
        yield self.registration_handler.post_consent_actions(qualified_user_id)

        try:
            self._render_template(request, "success.html")
        except TemplateNotFound:
            raise NotFoundError("success.html not found")

    def on_PUT(self, request):
        spec = _rule_spec_from_path(request.postpath)
        try:
            priority_class = _priority_class_from_spec(spec)
        except InvalidRuleException as e:
            raise SynapseError(400, str(e))

        requester = yield self.auth.get_user_by_req(request)

        if '/' in spec['rule_id'] or '\\' in spec['rule_id']:
            raise SynapseError(400, "rule_id may not contain slashes")

        content = parse_json_value_from_request(request)

        user_id = requester.user.to_string()

        if 'attr' in spec:
            yield self.set_rule_attr(user_id, spec, content)
            self.notify_user(user_id)
            defer.returnValue((200, {}))

        if spec['rule_id'].startswith('.'):
            # Rule ids starting with '.' are reserved for server default rules.
            raise SynapseError(400, "cannot add new rule_ids that start with '.'")

        try:
            (conditions, actions) = _rule_tuple_from_request_object(
                spec['template'],
                spec['rule_id'],
                content,
            )
        except InvalidRuleException as e:
            raise SynapseError(400, str(e))

        before = parse_string(request, "before")
        if before:
            before = _namespaced_rule_id(spec, before)

        after = parse_string(request, "after")
        if after:
            after = _namespaced_rule_id(spec, after)

        try:
            yield self.store.add_push_rule(
                user_id=user_id,
                rule_id=_namespaced_rule_id_from_spec(spec),
                priority_class=priority_class,
                conditions=conditions,
                actions=actions,
                before=before,
                after=after
            )
            self.notify_user(user_id)
        except InconsistentRuleException as e:
            raise SynapseError(400, str(e))
        except RuleNotFoundException as e:
            raise SynapseError(400, str(e))

        defer.returnValue((200, {}))

    def on_GET(self, request):
        requester = yield self.auth.get_user_by_req(request)
        user_id = requester.user.to_string()

        from_token = parse_string(request, "from", required=False)
        limit = parse_integer(request, "limit", default=50)
        only = parse_string(request, "only", required=False)

        limit = min(limit, 500)

        push_actions = yield self.store.get_push_actions_for_user(
            user_id, from_token, limit, only_highlight=(only == "highlight")
        )

        receipts_by_room = yield self.store.get_receipts_for_user_with_orderings(
            user_id, 'm.read'
        )

        notif_event_ids = [pa["event_id"] for pa in push_actions]
        notif_events = yield self.store.get_events(notif_event_ids)

        returned_push_actions = []

        next_token = None

        for pa in push_actions:
            returned_pa = {
                "room_id": pa["room_id"],
                "profile_tag": pa["profile_tag"],
                "actions": pa["actions"],
                "ts": pa["received_ts"],
                "event": serialize_event(
                    notif_events[pa["event_id"]],
                    self.clock.time_msec(),
                    event_format=format_event_for_client_v2_without_room_id,
                ),
            }

            if pa["room_id"] not in receipts_by_room:
                returned_pa["read"] = False
            else:
                receipt = receipts_by_room[pa["room_id"]]

                returned_pa["read"] = (
                    receipt["topological_ordering"], receipt["stream_ordering"]
                ) >= (
                    pa["topological_ordering"], pa["stream_ordering"]
                )
            returned_push_actions.append(returned_pa)
            next_token = str(pa["stream_ordering"])

        defer.returnValue((200, {
            "notifications": returned_push_actions,
            "next_token": next_token,
        }))

    def on_GET(self, request):
        user, client = yield self.auth.get_user_by_req(request)

        timeout = parse_integer(request, "timeout", default=0)
        limit = parse_integer(request, "limit", required=True)
        gap = parse_boolean(request, "gap", default=True)
        sort = parse_string(request, "sort", default="timeline,asc", allowed_values=self.ALLOWED_SORT)
        since = parse_string(request, "since")
        set_presence = parse_string(request, "set_presence", default="online", allowed_values=self.ALLOWED_PRESENCE)
        backfill = parse_boolean(request, "backfill", default=False)
        filter_id = parse_string(request, "filter", default=None)

        logger.info(
            "/sync: user=%r, timeout=%r, limit=%r, gap=%r, sort=%r, since=%r,"
            " set_presence=%r, backfill=%r, filter_id=%r"
            % (user, timeout, limit, gap, sort, since, set_presence, backfill, filter_id)
        )

        # TODO(mjark): Load filter and apply overrides.
        try:
            filter = yield self.filtering.get_user_filter(user.localpart, filter_id)
        except:
            filter = Filter({})
        # filter = filter.apply_overrides(http_request)
        # if filter.matches(event):
        #   # stuff

        sync_config = SyncConfig(
            user=user, client_info=client, gap=gap, limit=limit, sort=sort, backfill=backfill, filter=filter
        )

        if since is not None:
            since_token = StreamToken.from_string(since)
        else:
            since_token = None

        sync_result = yield self.sync_handler.wait_for_sync_for_user(
            sync_config, since_token=since_token, timeout=timeout
        )

        time_now = self.clock.time_msec()

        response_content = {
            "public_user_data": self.encode_user_data(sync_result.public_user_data, filter, time_now),
            "private_user_data": self.encode_user_data(sync_result.private_user_data, filter, time_now),
            "rooms": self.encode_rooms(sync_result.rooms, filter, time_now, client.token_id),
            "next_batch": sync_result.next_batch.to_string(),
        }

        defer.returnValue((200, response_content))

    def on_POST(self, request):
        yield self.auth.get_user_by_req(request, allow_guest=True)

        server = parse_string(request, "server", default=None)
        content = parse_json_object_from_request(request)

        limit = int(content.get("limit", 100))
        since_token = content.get("since", None)
        search_filter = content.get("filter", None)

        handler = self.hs.get_room_list_handler()
        if server:
            data = yield handler.get_remote_public_room_list(
                server,
                limit=limit,
                since_token=since_token,
                search_filter=search_filter,
            )
        else:
            data = yield handler.get_local_public_room_list(
                limit=limit,
                since_token=since_token,
                search_filter=search_filter,
            )

        defer.returnValue((200, data))

    def on_PUT_or_POST(
        self, request, room_id, parent_id, relation_type, event_type, txn_id=None
    ):
        requester = yield self.auth.get_user_by_req(request, allow_guest=True)

        if event_type == EventTypes.Member:
            # Add relations to a membership is meaningless, so we just deny it
            # at the CS API rather than trying to handle it correctly.
            raise SynapseError(400, "Cannot send member events with relations")

        content = parse_json_object_from_request(request)

        aggregation_key = parse_string(request, "key", encoding="utf-8")

        content["m.relates_to"] = {
            "event_id": parent_id,
            "key": aggregation_key,
            "rel_type": relation_type,
        }

        event_dict = {
            "type": event_type,
            "content": content,
            "room_id": room_id,
            "sender": requester.user.to_string(),
        }

        event = yield self.event_creation_handler.create_and_send_nonmember_event(
            requester, event_dict=event_dict, txn_id=txn_id
        )

        defer.returnValue((200, {"event_id": event.event_id}))

    def on_GET(self, request, target_user_id):
        """Get request to search user table for specific users according to
        search term.
        This needs user to have a administrator access in Synapse.
        """
        target_user = UserID.from_string(target_user_id)
        requester = yield self.auth.get_user_by_req(request)
        is_admin = yield self.auth.is_server_admin(requester.user)

        if not is_admin:
            raise AuthError(403, "You are not a server admin")

        # To allow all users to get the users list
        # if not is_admin and target_user != auth_user:
        #     raise AuthError(403, "You are not a server admin")

        if not self.hs.is_mine(target_user):
            raise SynapseError(400, "Can only users a local user")

        term = parse_string(request, "term", required=True)
        logger.info("term: %s ", term)

        ret = yield self.handlers.admin_handler.search_users(
            term
        )
        defer.returnValue((200, ret))

 def on_GET(self, request):
     client_redirect_url = parse_string(request, "redirectUrl", required=True)
     http_client = self.hs.get_simple_http_client()
     uri = self.cas_server_url + "/proxyValidate"
     args = {
         "ticket": parse_string(request, "ticket", required=True),
         "service": self.cas_service_url
     }
     try:
         body = yield http_client.get_raw(uri, args)
     except PartialDownloadError as pde:
         # Twisted raises this error if the connection is closed,
         # even if that's being used old-http style to signal end-of-data
         body = pde.response
     result = yield self.handle_cas_response(request, body, client_redirect_url)
     defer.returnValue(result)

    def _async_render_GET(self, request):
        limit = parse_integer(request, "limit", 100)
        timeout = parse_integer(request, "timeout", 10 * 1000)

        request.setHeader(b"Content-Type", b"application/json")

        request_streams = {
            name: parse_integer(request, name)
            for names in STREAM_NAMES for name in names
        }
        request_streams["streams"] = parse_string(request, "streams")

        def replicate():
            return self.replicate(request_streams, limit)

        result = yield self.notifier.wait_for_replication(replicate, timeout)

        for stream_name, stream_content in result.items():
            logger.info(
                "Replicating %d rows of %s from %s -> %s",
                len(stream_content["rows"]),
                stream_name,
                request_streams.get(stream_name),
                stream_content["position"],
            )

        request.write(json.dumps(result, ensure_ascii=False))
        finish_request(request)

    def on_GET(self, request):
        requester = yield self.auth.get_user_by_req(request, allow_guest=True)

        from_token_string = parse_string(request, "from")

        # We want to enforce they do pass us one, but we ignore it and return
        # changes after the "to" as well as before.
        parse_string(request, "to")

        from_token = StreamToken.from_string(from_token_string)

        user_id = requester.user.to_string()

        results = yield self.device_handler.get_user_ids_changed(
            user_id, from_token,
        )

        defer.returnValue((200, results))

    def on_GET(self, request):
        ip = self.hs.get_ip_from_request(request)
        with self.ratelimiter.ratelimit(ip) as wait_deferred:
            yield wait_deferred

            username = parse_string(request, "username", required=True)

            yield self.registration_handler.check_username(username)

            defer.returnValue((200, {"available": True}))

    def on_POST(self, request):
        requester = yield self.auth.get_user_by_req(request)

        content = parse_json_object_from_request(request)

        batch = parse_string(request, "next_batch")
        results = yield self.handlers.search_handler.search(
            requester.user,
            content,
            batch,
        )

        defer.returnValue((200, results))

    def _async_render_GET(self, request):
        """
        Args:
            request (twisted.web.http.Request):
        """

        version = parse_string(request, "v", default=self._default_consent_version)
        username = parse_string(request, "u", required=False, default="")
        userhmac = None
        has_consented = False
        public_version = username == ""
        if not public_version:
            userhmac_bytes = parse_string(request, "h", required=True, encoding=None)

            self._check_hash(username, userhmac_bytes)

            if username.startswith('@'):
                qualified_user_id = username
            else:
                qualified_user_id = UserID(username, self.hs.hostname).to_string()

            u = yield self.store.get_user_by_id(qualified_user_id)
            if u is None:
                raise NotFoundError("Unknown user")

            has_consented = u["consent_version"] == version
            userhmac = userhmac_bytes.decode("ascii")

        try:
            self._render_template(
                request, "%s.html" % (version,),
                user=username,
                userhmac=userhmac,
                version=version,
                has_consented=has_consented,
                public_version=public_version,
            )
        except TemplateNotFound:
            raise NotFoundError("Unknown policy version")

    def _async_render_POST(self, request):
        requester = yield self.auth.get_user_by_req(request)
        # TODO: The checks here are a bit late. The content will have
        # already been uploaded to a tmp file at this point
        content_length = request.getHeader(b"Content-Length").decode('ascii')
        if content_length is None:
            raise SynapseError(
                msg="Request must specify a Content-Length", code=400
            )
        if int(content_length) > self.max_upload_size:
            raise SynapseError(
                msg="Upload request body is too large",
                code=413,
            )

        upload_name = parse_string(request, b"filename", encoding=None)
        if upload_name:
            try:
                upload_name = upload_name.decode('utf8')
            except UnicodeDecodeError:
                raise SynapseError(
                    msg="Invalid UTF-8 filename parameter: %r" % (upload_name),
                    code=400,
                )

        headers = request.requestHeaders

        if headers.hasHeader(b"Content-Type"):
            media_type = headers.getRawHeaders(b"Content-Type")[0].decode('ascii')
        else:
            raise SynapseError(
                msg="Upload request missing 'Content-Type'",
                code=400,
            )

        # if headers.hasHeader(b"Content-Disposition"):
        #     disposition = headers.getRawHeaders(b"Content-Disposition")[0]
        # TODO(markjh): parse content-dispostion

        content_uri = yield self.media_repo.create_content(
            media_type, upload_name, request.content,
            content_length, requester.user
        )

        logger.info("Uploaded content with URI %r", content_uri)

        respond_with_json(
            request, 200, {"content_uri": content_uri}, send_cors=True
        )