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

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

merge bookmark @

author	Goffi <goffi@goffi.org>
date	Fri, 12 Nov 2021 17:48:30 +0100
parents	fc91b78b71db 83cd4862b134
children	6643855770a5

comparison

equal deleted inserted replaced

-:fc91b78b71db
+:595e7fef41f3
 import sys
 import urllib.parse
 import urllib.request, urllib.error
 import time
 import copy
+from typing import Optional
 from pathlib import Path
 from twisted.application import service
 from twisted.internet import reactor, defer, inotify
 from twisted.web import server
 from twisted.web import static
 def getChildWithDefault(self, path, request):
 return super().getChildWithDefault(path, request)
 def getChildForRequest(self, request):
 return super().getChildForRequest(request)
 class LiberviaRootResource(ProtectedFile):
 """Specialized resource for Libervia root
 handle redirections declared in sat.conf
 self.uri_callbacks = {}
 self.pages_redirects = {}
 self.cached_urls = {}
 self.main_menu = None
-# map SàT application names => data
+# map Libervia application names => data
-self.sat_apps = {}
+self.libervia_apps = {}
 self.build_path = host.getBuildPath(site_name)
 self.build_path.mkdir(parents=True, exist_ok=True)
 self.dev_build_path = host.getBuildPath(site_name, dev=True)
 self.dev_build_path.mkdir(parents=True, exist_ok=True)
 self.putChild(
 log.info(_(
 "starting application {app_name}").format(app_name=app_name))
 await self.host.bridgeCall(
 "applicationStart", app_name, data_format.serialise(extra)
 )
-app_data = self.sat_apps[app_name] = data_format.deserialise(
+app_data = self.libervia_apps[app_name] = data_format.deserialise(
 await self.host.bridgeCall(
 "applicationExposedGet", app_name, "", ""))
 try:
 web_port = int(app_data['ports']['web'].split(':')[1])
 ## redirections
 self.redirections = {}
 self.inv_redirections = {}  # new URL to old URL map
-for old, new_data in url_redirections.items():
+for old, new_data_list in url_redirections.items():
-# new_data can be a dictionary or a unicode url
+# several redirections can be used for one path by using a list.
-if isinstance(new_data, dict):
+# The redirection will be done using first item of the list, and all items
-# new_data dict must contain either "url", "page" or "path" key
+# will be used for inverse redirection.
-# (exclusive)
+# e.g. if a => [b, c], a will redirect to c, and b and c will both be
-# if "path" is used, a file url is constructed with it
+# equivalent to a
-if len({"path", "url", "page"}.intersection(list(new_data.keys()))) != 1:
+if not isinstance(new_data_list, list):
-raise ValueError(
+new_data_list = [new_data_list]
-'You must have one and only one of "url", "page" or "path" key '
+for new_data in new_data_list:
-'in your url_redirections_dict data')
+# new_data can be a dictionary or a unicode url
-if "url" in new_data:
+if isinstance(new_data, dict):
-new = new_data["url"]
+# new_data dict must contain either "url", "page" or "path" key
-elif "page" in new_data:
+# (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["type"] = "page"
+new_data = {}
-new.setdefault("path_args", [])
+else:
-if not isinstance(new["path_args"], list):
+log.error(
-log.error(
+_("ignoring invalid redirection value: {new_data}").format(
-_('"path_args" in redirection of {old} must be a list. '
+new_data=new_data
-'Ignoring the redirection'.format(old=old)))
+)
+)
+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)
+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
+# 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
-new.setdefault("query_args", {})
+request_data = self._getRequestData(location)
-if not isinstance(new["query_args"], dict):
+self.inv_redirections[location] = old
-log.error(
-_(
+elif new_url.scheme in ("", "http", "https"):
-'"query_args" in redirection of {old} must be a '
+# direct redirection
-'dictionary. Ignoring the redirection'.format(old=old)))
+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))
+# we don't want to use redirection system, so we continue here
+continue
+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
-new["path_args"] = [quote(a) for a in new["path_args"]]
-# we keep an inversed dict of page redirection
+log.info("[{host_name}] Added redirection from /{old} to application "
-# (page/path_args => redirecting URL)
+"{app_name}".format(
-# so getURL can return the redirecting URL if the same arguments
+host_name=self.host_name,
-# are used # making the URL consistent
+old=old,
-args_hash = tuple(new["path_args"])
+app_name=app_name))
-self.pages_redirects.setdefault(new_data["page"], {})[
-args_hash
+# normal redirection system is not used here
-] = 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
-)
-)
-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)
-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"]
-)
-)
-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)
-# 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:
-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:
-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))
-# 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))
-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))
-# normal redirection system is not used here
-continue
 elif new_url.scheme == "proxy":
 # a reverse proxy
 host, port = new_url.hostname, new_url.port
 if host is None or port is None:
 raise ValueError(
 f"{new_url.netloc} with URL prefix {url_prefix}/"
 )
 # 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.setdefault(old, request_data)
-self.redirections[old] = request_data
+if not old:
-if not old:
+log.info(_("[{host_name}] Root URL redirected to {uri}")
-log.info(_("[{host_name}] Root URL redirected to {uri}")
+.format(host_name=self.host_name,
-.format(host_name=self.host_name,
+uri=request_data[1]))
-uri=request_data[1]))
 # the default root URL, if not redirected
 if not "" in self.redirections:
 self.redirections[""] = self._getRequestData(C.LIBERVIA_PAGE_START)
 "menu item as list must be in the form [page_name, absolue URL]"
 )
 log.error(msg)
 raise ValueError(msg)
 page_name, url = menu
-elif menu.startswith("sat-app:"):
+elif menu.startswith("libervia-app:"):
-app_name = menu[8:].strip().lower()
+app_name = menu[13:].strip().lower()
 app_data = await self._startApp(app_name)
 front_url = app_data['front_url']
 options = self.host.options
 url_redirections = options["url_redirections_dict"].setdefault(
 self.site_name, {})
 if isinstance(resource, web_resource.NoResource):
 # 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]
 request.postpath = path_elt[idx:]
 return self._redirect(request, request_data)
 @return (urlparse.SplitResult): SplitResult instance with only scheme and
 netloc filled
 """
 ext_data = self.base_url_ext_data
 url_path = request.URLPath()
-if not ext_data.scheme or not ext_data.netloc:
-#  ext_data is not specified, we check headers
+try:
-if request.requestHeaders.hasHeader("x-forwarded-host"):
+forwarded = request.requestHeaders.getRawHeaders(
-# we are behing a proxy
+"forwarded"
-# we fill proxy_scheme and proxy_netloc value
+)[0]
-proxy_host = request.requestHeaders.getRawHeaders("x-forwarded-host")[0]
+except TypeError:
-try:
+# we try deprecated headers
-proxy_server = request.requestHeaders.getRawHeaders(
+try:
-"x-forwarded-server"
+proxy_netloc = request.requestHeaders.getRawHeaders(
-)[0]
+"x-forwarded-host"
-except TypeError:
+)[0]
-# no x-forwarded-server found, we use proxy_host
+except TypeError:
-proxy_netloc = proxy_host
+proxy_netloc = None
-else:
+try:
-# if the proxy host has a port, we use it with server name
+proxy_scheme = request.requestHeaders.getRawHeaders(
-proxy_port = urllib.parse.urlsplit("//{}".format(proxy_host)).port
+"x-forwarded-proto"
-proxy_netloc = (
+)[0]
-"{}:{}".format(proxy_server, proxy_port)
+except TypeError:
-if proxy_port is not None
+proxy_scheme = None
-else proxy_server
-)
-try:
-proxy_scheme = request.requestHeaders.getRawHeaders(
-"x-forwarded-proto"
-)[0]
-except TypeError:
-proxy_scheme = None
-else:
-proxy_scheme, proxy_netloc = None, None
 else:
-proxy_scheme, proxy_netloc = None, None
+fwd_data = {
+k.strip(): v.strip()
+for k,v in (d.split("=") for d in forwarded.split(";"))
+}
+proxy_netloc = fwd_data.get("host")
+proxy_scheme = fwd_data.get("proto")
 return urllib.parse.SplitResult(
-ext_data.scheme or proxy_scheme or url_path.scheme.decode("utf-8"),
+ext_data.scheme or proxy_scheme or url_path.scheme.decode(),
-ext_data.netloc or proxy_netloc or url_path.netloc.decode("utf-8"),
+ext_data.netloc or proxy_netloc or url_path.netloc.decode(),
 ext_data.path or "/",
 "",
 "",
 )
-def getExtBaseURL(self, request, path="", query="", fragment="", scheme=None):
+def getExtBaseURL(
+self,
+request: server.Request,
+path: str = "",
+query: str = "",
+fragment: str = "",
+scheme: Optional[str] = None,
+) -> str:
 """Get external URL according to given elements
 external URL is the URL seen by external user
-@param path(unicode): same as for urlsplit.urlsplit
+@param path: same as for urlsplit.urlsplit
 path will be prefixed to follow found external URL if suitable
-@param params(unicode): same as for urlsplit.urlsplit
+@param params: same as for urlsplit.urlsplit
-@param query(unicode): same as for urlsplit.urlsplit
+@param query: same as for urlsplit.urlsplit
-@param fragment(unicode): same as for urlsplit.urlsplit
+@param fragment: same as for urlsplit.urlsplit
-@param scheme(unicode, None): if not None, will override scheme from base URL
+@param scheme: if not None, will override scheme from base URL
-@return (unicode): external URL
+@return: external URL
 """
 split_result = self.getExtBaseURLData(request)
 return urllib.parse.urlunsplit(
 (
 split_result.scheme if scheme is None else scheme,
 query,
 fragment,
 )
 )
-def checkRedirection(self, vhost_root, url):
+def checkRedirection(self, vhost_root: LiberviaRootResource, url_path: str) -> str:
 """check is a part of the URL prefix is redirected then replace it
-@param vhost_root(web_resource.Resource): root of this virtual host
+@param vhost_root: root of this virtual host
-@param url(unicode): url to check
+@param url_path: path of the url to check
-@return (unicode): possibly redirected URL which should link to the same location
+@return: possibly redirected URL which should link to the same location
 """
 inv_redirections = vhost_root.inv_redirections
-url_parts = url.strip("/").split("/")
+url_parts = url_path.strip("/").split("/")
-for idx in range(len(url), 0, -1):
+for idx in range(len(url_parts), -1, -1):
 test_url = "/" + "/".join(url_parts[:idx])
 if test_url in inv_redirections:
 rem_url = url_parts[idx:]
 return os.path.join(
 "/", "/".join([inv_redirections[test_url]] + rem_url)
 )
-return url
+return url_path
 ## Sessions ##
 def purgeSession(self, request):
 """helper method to purge a session during request handling"""

Mercurial > libervia-web

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