Mercurial > libervia-web
diff libervia/server/pages.py @ 1483:595e7fef41f3
merge bookmark @
author | Goffi <goffi@goffi.org> |
---|---|
date | Fri, 12 Nov 2021 17:48:30 +0100 |
parents | 095e94ca6728 |
children | 774a81a6e8b5 |
line wrap: on
line diff
--- a/libervia/server/pages.py Tue Sep 28 18:18:37 2021 +0200 +++ b/libervia/server/pages.py Fri Nov 12 17:48:30 2021 +0100 @@ -546,9 +546,14 @@ log.info(_("{page} reloaded").format(page=resource)) def checkCSRF(self, request): - csrf_token = self.host.getSessionData( + session = self.host.getSessionData( request, session_iface.ISATSession - ).csrf_token + ) + if session.profile is None: + # CSRF doesn't make sense when no user is logged + log.debug("disabling CSRF check because service profile is used") + return + csrf_token = session.csrf_token given_csrf = request.getHeader("X-Csrf-Token") if given_csrf is None: try: @@ -577,6 +582,10 @@ for name, value in kwargs.items(): if value is None: value = "null" + elif isinstance(value, str): + # FIXME: workaround for subtype used by python-dbus (dbus.String) + # to be removed when we get rid of python-dbus + value = repr(str(value)) else: value = repr(value) scripts.add(Script(content=f"var {name}={value};")) @@ -665,11 +674,12 @@ else url.encode("utf-8"), ) - def getURL(self, *args): + def getURL(self, *args: str, **kwargs: str) -> str: """retrieve URL of the page set arguments - *args(list[unicode]): argument to add to the URL as path elements - empty or None arguments will be ignored + @param *args: arguments to add to the URL as path elements empty or None + arguments will be ignored + @param **kwargs: query parameters """ url_args = [quote(a) for a in args if a] @@ -677,15 +687,29 @@ # we check for redirection redirect_data = self.pages_redirects[self.name] args_hash = tuple(args) - for limit in range(len(args) + 1): + for limit in range(len(args), -1, -1): current_hash = args_hash[:limit] if current_hash in redirect_data: url_base = redirect_data[current_hash] remaining = args[limit:] remaining_url = "/".join(remaining) - return os.path.join("/", url_base, remaining_url) + url = urllib.parse.urljoin(url_base, remaining_url) + break + else: + url = os.path.join(self.url, *url_args) + else: + url = os.path.join(self.url, *url_args) - return os.path.join(self.url, *url_args) + if kwargs: + encoded = urllib.parse.urlencode( + {k: v for k, v in kwargs.items()} + ) + url += f"?{encoded}" + + return self.host.checkRedirection( + self.vhost_root, + url + ) def getCurrentURL(self, request): """retrieve URL used to access this page @@ -961,7 +985,7 @@ else: assert not {"rsm_max", "rsm_after", "rsm_before", C.KEY_ORDER_BY}.intersection(list(extra.keys())) - extra["rsm_max"] = str(page_max) + extra["rsm_max"] = params.get("page_max", str(page_max)) if order_by is not None: extra[C.KEY_ORDER_BY] = order_by if 'after' in params: @@ -974,13 +998,13 @@ extra['rsm_before'] = "" return extra - def setPagination(self, request, pubsub_data): + def setPagination(self, request: server.Request, pubsub_data: dict) -> None: """Add to template_data if suitable "previous_page_url" and "next_page_url" will be added using respectively "before" and "after" URL parameters - @param request(server.Request): current HTTP request - @param pubsub_data(dict): pubsub metadata + @param request: current HTTP request + @param pubsub_data: pubsub metadata """ template_data = request.template_data extra = {} @@ -996,6 +1020,11 @@ if search is not None: extra['search'] = search.strip() + # same for page_max + page_max = self.getPostedData(request, 'page_max', raise_on_missing=False) + if page_max is not None: + extra['page_max'] = page_max + if rsm.get("index", 1) > 0: # We only show previous button if it's not the first page already. # If we have no index, we default to display the button anyway @@ -1709,7 +1738,7 @@ accept_language = request.getHeader("accept-language") if not accept_language: return - accepted = {a.strip() for a in accept_language.split(',')} + accepted = [a.strip() for a in accept_language.split(',')] available = [str(l) for l in self.host.renderer.translations] for lang in accepted: lang = lang.split(';')[0].strip().lower() @@ -1785,10 +1814,13 @@ # if template_data doesn't exist, it's the beginning of the request workflow # so we fill essential data session_data = self.host.getSessionData(request, session_iface.ISATSession) + profile = session_data.profile request.template_data = { - "profile": session_data.profile, - "csrf_token": session_data.csrf_token, - "session_uuid": session_data.uuid, + "profile": profile, + # it's important to not add CSRF token and session uuid if service profile + # is used because the page may be cached, and the token then leaked + "csrf_token": "" if profile is None else session_data.csrf_token, + "session_uuid": "public" if profile is None else session_data.uuid, "breadcrumbs": [] }