diff libervia/server/server.py @ 1457:792a2e902ee9

server: fix inverse URL redirection for root path + allow multiple inverse redirections: - inverse redirection is now working for root path - a list can now be used in `url_redirections_dict`: the first item only will be used for redirection, but all items will be used for inverse redirection. e.g.: if in `url_redirections_dict` we have `"/": ["/u/some_user/blog", "/blog/view/some_user_jid@example.org"]`, root will redirect to "/u/some_user/blog", but both "/u/some_user/blog" and "/blog/view/some_user_jid@example.org" will have an inverse redirection to the root path fix 395
author Goffi <goffi@goffi.org>
date Wed, 29 Sep 2021 17:39:06 +0200
parents 396d5606477f
children db13f5c768a0
line wrap: on
line diff
--- a/libervia/server/server.py	Wed Sep 29 17:38:29 2021 +0200
+++ b/libervia/server/server.py	Wed Sep 29 17:39:06 2021 +0200
@@ -337,180 +337,196 @@
         self.redirections = {}
         self.inv_redirections = {}  # new URL to old URL map
 
-        for old, new_data in url_redirections.items():
-            # new_data can be a dictionary or a unicode url
-            if isinstance(new_data, dict):
-                # new_data dict must contain either "url", "page" or "path" key
-                # (exclusive)
-                # if "path" is used, a file url is constructed with it
-                if len({"path", "url", "page"}.intersection(list(new_data.keys()))) != 1:
-                    raise ValueError(
-                        'You must have one and only one of "url", "page" or "path" key '
-                        'in your url_redirections_dict data')
-                if "url" in new_data:
-                    new = new_data["url"]
-                elif "page" in new_data:
-                    new = new_data
-                    new["type"] = "page"
-                    new.setdefault("path_args", [])
-                    if not isinstance(new["path_args"], list):
-                        log.error(
-                            _('"path_args" in redirection of {old} must be a list. '
-                              'Ignoring the redirection'.format(old=old)))
-                        continue
-                    new.setdefault("query_args", {})
-                    if not isinstance(new["query_args"], dict):
-                        log.error(
-                            _(
-                                '"query_args" in redirection of {old} must be a '
-                                'dictionary. Ignoring the redirection'.format(old=old)))
-                        continue
-                    new["path_args"] = [quote(a) for a in new["path_args"]]
-                    # we keep an inversed dict of page redirection
-                    # (page/path_args => redirecting URL)
-                    # so getURL can return the redirecting URL if the same arguments
-                    # are used # making the URL consistent
-                    args_hash = tuple(new["path_args"])
-                    self.pages_redirects.setdefault(new_data["page"], {})[
-                        args_hash
-                    ] = old
+        for old, new_data_list in url_redirections.items():
+            # several redirections can be used for one path by using a list.
+            # The redirection will be done using first item of the list, and all items
+            # will be used for inverse redirection.
+            # e.g. if a => [b, c], a will redirect to c, and b and c will both be
+            # equivalent to a
+            if not isinstance(new_data_list, list):
+                new_data_list = [new_data_list]
+            for new_data in new_data_list:
+                # new_data can be a dictionary or a unicode url
+                if isinstance(new_data, dict):
+                    # new_data dict must contain either "url", "page" or "path" key
+                    # (exclusive)
+                    # if "path" is used, a file url is constructed with it
+                    if ((
+                        len(
+                            {"path", "url", "page"}.intersection(list(new_data.keys()))
+                        ) != 1
+                    )):
+                        raise ValueError(
+                            'You must have one and only one of "url", "page" or "path" '
+                            'key in your url_redirections_dict data'
+                        )
+                    if "url" in new_data:
+                        new = new_data["url"]
+                    elif "page" in new_data:
+                        new = new_data
+                        new["type"] = "page"
+                        new.setdefault("path_args", [])
+                        if not isinstance(new["path_args"], list):
+                            log.error(
+                                _('"path_args" in redirection of {old} must be a list. '
+                                  'Ignoring the redirection'.format(old=old)))
+                            continue
+                        new.setdefault("query_args", {})
+                        if not isinstance(new["query_args"], dict):
+                            log.error(
+                                _(
+                                    '"query_args" in redirection of {old} must be a '
+                                    'dictionary. Ignoring the redirection'
+                                ).format(old=old)
+                            )
+                            continue
+                        new["path_args"] = [quote(a) for a in new["path_args"]]
+                        # we keep an inversed dict of page redirection
+                        # (page/path_args => redirecting URL)
+                        # so getURL can return the redirecting URL if the same arguments
+                        # are used # making the URL consistent
+                        args_hash = tuple(new["path_args"])
+                        self.pages_redirects.setdefault(new_data["page"], {}).setdefault(
+                            args_hash,
+                            old
+                        )
 
-                    # we need lists in query_args because it will be used
-                    # as it in request.path_args
-                    for k, v in new["query_args"].items():
-                        if isinstance(v, str):
-                            new["query_args"][k] = [v]
-                elif "path" in new_data:
-                    new = "file:{}".format(urllib.parse.quote(new_data["path"]))
-            elif isinstance(new_data, str):
-                new = new_data
-                new_data = {}
-            else:
-                log.error(
-                    _("ignoring invalid redirection value: {new_data}").format(
-                        new_data=new_data
+                        # we need lists in query_args because it will be used
+                        # as it in request.path_args
+                        for k, v in new["query_args"].items():
+                            if isinstance(v, str):
+                                new["query_args"][k] = [v]
+                    elif "path" in new_data:
+                        new = "file:{}".format(urllib.parse.quote(new_data["path"]))
+                elif isinstance(new_data, str):
+                    new = new_data
+                    new_data = {}
+                else:
+                    log.error(
+                        _("ignoring invalid redirection value: {new_data}").format(
+                            new_data=new_data
+                        )
                     )
-                )
-                continue
+                    continue
 
-            # some normalization
-            if not old.strip():
-                # root URL special case
-                old = ""
-            elif not old.startswith("/"):
-                log.error(_("redirected url must start with '/', got {value}. Ignoring")
-                          .format(value=old))
-                continue
-            else:
-                old = self._normalizeURL(old)
+                # some normalization
+                if not old.strip():
+                    # root URL special case
+                    old = ""
+                elif not old.startswith("/"):
+                    log.error(
+                        _("redirected url must start with '/', got {value}. Ignoring")
+                        .format(value=old)
+                    )
+                    continue
+                else:
+                    old = self._normalizeURL(old)
 
-            if isinstance(new, dict):
-                # dict are handled differently, they contain data
-                # which ared use dynamically when the request is done
-                self.redirections[old] = new
-                if not old:
-                    if new["type"] == "page":
-                        log.info(
-                            _("Root URL redirected to page {name}").format(
-                                name=new["page"]
+                if isinstance(new, dict):
+                    # dict are handled differently, they contain data
+                    # which ared use dynamically when the request is done
+                    self.redirections.setdefault(old, new)
+                    if not old:
+                        if new["type"] == "page":
+                            log.info(
+                                _("Root URL redirected to page {name}").format(
+                                    name=new["page"]
+                                )
                             )
-                        )
-                else:
-                    if new["type"] == "page":
-                        page = self.getPageByName(new["page"])
-                        url = page.getURL(*new.get("path_args", []))
-                        self.inv_redirections[url] = old
-                continue
+                    else:
+                        if new["type"] == "page":
+                            page = self.getPageByName(new["page"])
+                            url = page.getURL(*new.get("path_args", []))
+                            self.inv_redirections[url] = old
+                    continue
 
-            # at this point we have a redirection URL in new, we can parse it
-            new_url = urllib.parse.urlsplit(new)
+                # at this point we have a redirection URL in new, we can parse it
+                new_url = urllib.parse.urlsplit(new)
 
-            # we handle the known URL schemes
-            if new_url.scheme == "xmpp":
-                location = self.getPagePathFromURI(new)
-                if location is None:
-                    log.warning(
-                        _("ignoring redirection, no page found to handle this URI: "
-                          "{uri}").format(uri=new))
-                    continue
-                request_data = self._getRequestData(location)
-                if old:
+                # we handle the known URL schemes
+                if new_url.scheme == "xmpp":
+                    location = self.getPagePathFromURI(new)
+                    if location is None:
+                        log.warning(
+                            _("ignoring redirection, no page found to handle this URI: "
+                              "{uri}").format(uri=new))
+                        continue
+                    request_data = self._getRequestData(location)
                     self.inv_redirections[location] = old
 
-            elif new_url.scheme in ("", "http", "https"):
-                # direct redirection
-                if new_url.netloc:
-                    raise NotImplementedError(
-                        "netloc ({netloc}) is not implemented yet for "
-                        "url_redirections_dict, it is not possible to redirect to an "
-                        "external website".format(netloc=new_url.netloc))
-                location = urllib.parse.urlunsplit(
-                    ("", "", new_url.path, new_url.query, new_url.fragment)
-                )
-                request_data = self._getRequestData(location)
-                if old:
+                elif new_url.scheme in ("", "http", "https"):
+                    # direct redirection
+                    if new_url.netloc:
+                        raise NotImplementedError(
+                            "netloc ({netloc}) is not implemented yet for "
+                            "url_redirections_dict, it is not possible to redirect to an "
+                            "external website".format(netloc=new_url.netloc))
+                    location = urllib.parse.urlunsplit(
+                        ("", "", new_url.path, new_url.query, new_url.fragment)
+                    )
+                    request_data = self._getRequestData(location)
                     self.inv_redirections[location] = old
 
-            elif new_url.scheme == "file":
-                # file or directory
-                if new_url.netloc:
-                    raise NotImplementedError(
-                        "netloc ({netloc}) is not implemented for url redirection to "
-                        "file system, it is not possible to redirect to an external "
-                        "host".format(
-                            netloc=new_url.netloc))
-                path = urllib.parse.unquote(new_url.path)
-                if not os.path.isabs(path):
-                    raise ValueError(
-                        "file redirection must have an absolute path: e.g. "
-                        "file:/path/to/my/file")
-                # for file redirection, we directly put child here
-                resource_class = (
-                    ProtectedFile if new_data.get("protected", True) else static.File
-                )
-                res = resource_class(path, defaultType="application/octet-stream")
-                self.addResourceToPath(old, res)
-                log.info("[{host_name}] Added redirection from /{old} to file system "
-                         "path {path}".format(host_name=self.host_name,
-                                               old=old,
-                                               path=path))
+                elif new_url.scheme == "file":
+                    # file or directory
+                    if new_url.netloc:
+                        raise NotImplementedError(
+                            "netloc ({netloc}) is not implemented for url redirection to "
+                            "file system, it is not possible to redirect to an external "
+                            "host".format(
+                                netloc=new_url.netloc))
+                    path = urllib.parse.unquote(new_url.path)
+                    if not os.path.isabs(path):
+                        raise ValueError(
+                            "file redirection must have an absolute path: e.g. "
+                            "file:/path/to/my/file")
+                    # for file redirection, we directly put child here
+                    resource_class = (
+                        ProtectedFile if new_data.get("protected", True) else static.File
+                    )
+                    res = resource_class(path, defaultType="application/octet-stream")
+                    self.addResourceToPath(old, res)
+                    log.info("[{host_name}] Added redirection from /{old} to file system "
+                             "path {path}".format(host_name=self.host_name,
+                                                   old=old,
+                                                   path=path))
 
-                # we don't want to use redirection system, so we continue here
-                continue
-
-            elif new_url.scheme == "sat-app":
-                # a SàT application
-
-                app_name = urllib.parse.unquote(new_url.path).lower().strip()
-                extra = {"url_prefix": f"/{old}"}
-                try:
-                    await self._startApp(app_name, extra)
-                except Exception as e:
-                    log.warning(_(
-                        "Can't launch {app_name!r} for path /{old}: {e}").format(
-                        app_name=app_name, old=old, e=e))
+                    # we don't want to use redirection system, so we continue here
                     continue
 
-                log.info("[{host_name}] Added redirection from /{old} to application "
-                         "{app_name}".format(
-                             host_name=self.host_name,
-                             old=old,
-                             app_name=app_name))
+                elif new_url.scheme == "libervia-app":
+                    # a Libervia application
+
+                    app_name = urllib.parse.unquote(new_url.path).lower().strip()
+                    extra = {"url_prefix": f"/{old}"}
+                    try:
+                        await self._startApp(app_name, extra)
+                    except Exception as e:
+                        log.warning(_(
+                            "Can't launch {app_name!r} for path /{old}: {e}").format(
+                            app_name=app_name, old=old, e=e))
+                        continue
 
-                # normal redirection system is not used here
-                continue
-            else:
-                raise NotImplementedError(
-                    "{scheme}: scheme is not managed for url_redirections_dict".format(
-                        scheme=new_url.scheme
+                    log.info("[{host_name}] Added redirection from /{old} to application "
+                             "{app_name}".format(
+                                 host_name=self.host_name,
+                                 old=old,
+                                 app_name=app_name))
+
+                    # normal redirection system is not used here
+                    continue
+                else:
+                    raise NotImplementedError(
+                        "{scheme}: scheme is not managed for url_redirections_dict".format(
+                            scheme=new_url.scheme
+                        )
                     )
-                )
 
-            self.redirections[old] = request_data
-            if not old:
-                log.info(_("[{host_name}] Root URL redirected to {uri}")
-                    .format(host_name=self.host_name,
-                            uri=request_data[1]))
+                self.redirections.setdefault(old, request_data)
+                if not old:
+                    log.info(_("[{host_name}] Root URL redirected to {uri}")
+                        .format(host_name=self.host_name,
+                                uri=request_data[1]))
 
         # the default root URL, if not redirected
         if not "" in self.redirections:
@@ -724,7 +740,7 @@
             # if nothing was found, we try our luck with redirections
             # XXX: we want redirections to happen only if everything else failed
             path_elt = request.prepath + request.postpath
-            for idx in range(len(path_elt), 0, -1):
+            for idx in range(len(path_elt), -1, -1):
                 test_url = b"/".join(path_elt[:idx]).decode('utf-8').lower()
                 if test_url in self.redirections:
                     request_data = self.redirections[test_url]