libervia-web: libervia/server/pages.py comparison

comparison 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

comparison

equal deleted inserted replaced

-:fc91b78b71db
+:595e7fef41f3
 log.info(_("{page} created").format(page=resource))
 else:
 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:
 given_csrf = self.getPostedData(request, "csrf_token")
 except KeyError:
 template_data = request.template_data
 scripts = template_data.setdefault("scripts", utils.OrderedSet())
 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};"))
 def registerURI(self, uri_tuple, get_uri_cb):
 redirect_url=urllib.parse.quote_plus(request.uri)
 if url is None
 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
+@param *args: arguments to add to the URL as path elements empty or None
-empty or None arguments will be ignored
+arguments will be ignored
+@param **kwargs: query parameters
 """
 url_args = [quote(a) for a in args if a]
 if self.name is not None and self.name in self.pages_redirects:
 #  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
-return os.path.join(self.url, *url_args)
+else:
+url = os.path.join(self.url, *url_args)
+else:
+url = 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
 @return(unicode): current URL
 if extra is None:
 extra = {}
 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:
 extra['rsm_after'] = params['after']
 elif 'before' in params:
 # RSM returns list in order (oldest first), but we want most recent first
 # so we start by the end
 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 request: current HTTP request
-@param pubsub_data(dict): pubsub metadata
+@param pubsub_data: pubsub metadata
 """
 template_data = request.template_data
 extra = {}
 try:
 rsm = pubsub_data["rsm"]
 # if we have a search query, we must keep it
 search = self.getPostedData(request, 'search', raise_on_missing=False)
 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
 # as we can't know if we are on the first page or not.
 matching value with available translations.
 """
 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()
 if not lang:
 continue
 # template_data are the variables passed to template
 if not hasattr(request, "template_data"):
 # 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,
+"profile": profile,
-"csrf_token": session_data.csrf_token,
+# it's important to not add CSRF token and session uuid if service profile
-"session_uuid": session_data.uuid,
+# 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": []
 }
 # XXX: here is the code which need to be executed once
 #      at the beginning of the request hanling

Mercurial > libervia-web

comparison libervia/server/pages.py @ 1483:595e7fef41f3