comparison mercurial/hgweb/hgweb_mod.py @ 43077:687b865b95ad

formatting: byteify all mercurial/ and hgext/ string literals

Done with

    python3.7 contrib/byteify-strings.py -i $(hg files 'set:mercurial/**.py - mercurial/thirdparty/** + hgext/**.py - hgext/fsmonitor/pywatchman/** - mercurial/__init__.py')
    black -l 80 -t py33 -S $(hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**" - hgext/fsmonitor/pywatchman/**')

# skip-blame mass-reformatting only

Differential Revision: https://phab.mercurial-scm.org/D6972

43076:2372284d9457

)


def getstyle(req, configfn, templatepath):
    styles = (
        req.qsparams.get('style', None),
        configfn('web', 'style'),
        'paper',
    )
    return styles, templater.stylemap(styles, templatepath)


def makebreadcrumb(url, prefix=''):
    '''Return a 'URL breadcrumb' list

    A 'URL breadcrumb' is a list of URL-name pairs,
    corresponding to each of the path items on a URL.
    This can be used to create path navigation entries.
    '''
    if url.endswith('/'):
        url = url[:-1]
    if prefix:
        url = '/' + prefix + url
    relpath = url
    if relpath.startswith('/'):
        relpath = relpath[1:]

    breadcrumb = []
    urlel = url
    pathitems = [''] + relpath.split('/')
    for pathel in reversed(pathitems):
        if not pathel or not urlel:
            break
        breadcrumb.append({'url': urlel, 'name': pathel})
        urlel = os.path.dirname(urlel)
    return templateutil.mappinglist(reversed(breadcrumb))


class requestcontext(object):

        self.repo = repo
        self.reponame = app.reponame
        self.req = req
        self.res = res

        self.maxchanges = self.configint('web', 'maxchanges')
        self.stripecount = self.configint('web', 'stripes')
        self.maxshortchanges = self.configint('web', 'maxshortchanges')
        self.maxfiles = self.configint('web', 'maxfiles')
        self.allowpull = self.configbool('web', 'allow-pull')

        # we use untrusted=False to prevent a repo owner from using
        # web.templates in .hg/hgrc to get access to any file readable
        # by the user running the CGI script
        self.templatepath = self.config('web', 'templates', untrusted=False)

        # This object is more expensive to build than simple config values.
        # It is shared across requests. The app will replace the object
        # if it is updated. Since this is a reference and nothing should
        # modify the underlying object, it should be constant for the lifetime

        return webutil.archivelist(self.repo.ui, nodeid)

    def templater(self, req):
        # determine scheme, port and server name
        # this is needed to create absolute urls
        logourl = self.config('web', 'logourl')
        logoimg = self.config('web', 'logoimg')
        staticurl = (
            self.config('web', 'staticurl')
            or req.apppath.rstrip('/') + '/static/'
        )
        if not staticurl.endswith('/'):
            staticurl += '/'

        # figure out which style to use

        vars = {}
        styles, (style, mapfile) = getstyle(req, self.config, self.templatepath)
        if style == styles[0]:
            vars['style'] = style

        sessionvars = webutil.sessionvars(vars, '?')

        if not self.reponame:
            self.reponame = (
                self.config('web', 'name', '')
                or req.reponame
                or req.apppath
                or self.repo.root
            )

        filters = {}
        templatefilter = registrar.templatefilter(filters)

        @templatefilter('websub', intype=bytes)
        def websubfilter(text):
            return templatefilters.websub(text, self.websubtable)

        # create the templater
        # TODO: export all keywords: defaults = templatekw.keywords.copy()
        defaults = {
            'url': req.apppath + '/',
            'logourl': logourl,
            'logoimg': logoimg,
            'staticurl': staticurl,
            'urlbase': req.advertisedbaseurl,
            'repo': self.reponame,
            'encoding': encoding.encoding,
            'sessionvars': sessionvars,
            'pathdef': makebreadcrumb(req.apppath),
            'style': style,
            'nonce': self.nonce,
        }
        templatekeyword = registrar.templatekeyword(defaults)

        @templatekeyword('motd', requires=())
        def motd(context, mapping):
            yield self.config('web', 'motd')

        tres = formatter.templateresources(self.repo.ui, self.repo)
        tmpl = templater.templater.frommapfile(
            mapfile, filters=filters, defaults=defaults, resources=tres
        )

            r = hg.repository(u, repo)
        else:
            # we trust caller to give us a private copy
            r = repo

        r.ui.setconfig('ui', 'report_untrusted', 'off', 'hgweb')
        r.baseui.setconfig('ui', 'report_untrusted', 'off', 'hgweb')
        r.ui.setconfig('ui', 'nontty', 'true', 'hgweb')
        r.baseui.setconfig('ui', 'nontty', 'true', 'hgweb')
        # resolve file patterns relative to repo root
        r.ui.setconfig('ui', 'forcecwd', r.root, 'hgweb')
        r.baseui.setconfig('ui', 'forcecwd', r.root, 'hgweb')
        # it's unlikely that we can replace signal handlers in WSGI server,
        # and mod_wsgi issues a big warning. a plain hgweb process (with no
        # threading) could replace signal handlers, but we don't bother
        # conditionally enabling it.
        r.ui.setconfig('ui', 'signal-safe-lock', 'false', 'hgweb')
        r.baseui.setconfig('ui', 'signal-safe-lock', 'false', 'hgweb')
        # displaying bundling progress bar while serving feel wrong and may
        # break some wsgi implementation.
        r.ui.setconfig('progress', 'disable', 'true', 'hgweb')
        r.baseui.setconfig('progress', 'disable', 'true', 'hgweb')
        self._repos = [hg.cachedlocalrepo(self._webifyrepo(r))]
        self._lastrepo = self._repos[0]
        hook.redirect(True)
        self.reponame = name

        """Start a server from CGI environment.

        Modern servers should be using WSGI and should avoid this
        method, if possible.
        """
        if not encoding.environ.get('GATEWAY_INTERFACE', '').startswith(
            "CGI/1."
        ):
            raise RuntimeError(
                "This function is only intended to be "
                "called while running as a CGI script."
            )
        wsgicgi.launch(self)

    def __call__(self, env, respond):
        """Run the WSGI application.

        This is typically only called by Mercurial. External consumers
        should be using instances of this class as the WSGI application.
        """
        with self._obtainrepo() as repo:
            profile = repo.ui.configbool('profiling', 'enabled')
            with profiling.profile(repo.ui, enabled=profile):
                for r in self._runwsgi(req, res, repo):
                    yield r

    def _runwsgi(self, req, res, repo):
        rctx = requestcontext(self, repo, req, res)

        # This state is global across all threads.
        encoding.encoding = rctx.config('web', 'encoding')
        rctx.repo.ui.environ = req.rawenv

        if rctx.csp:
            # hgwebdir may have added CSP header. Since we generate our own,
            # replace it.
            res.headers['Content-Security-Policy'] = rctx.csp

        # /api/* is reserved for various API implementations. Dispatch
        # accordingly. But URL paths can conflict with subrepos and virtual
        # repos in hgwebdir. So until we have a workaround for this, only
        # expose the URLs if the feature is enabled.
        apienabled = rctx.repo.ui.configbool('experimental', 'web.apiserver')
        if apienabled and req.dispatchparts and req.dispatchparts[0] == b'api':
            wireprotoserver.handlewsgiapirequest(
                rctx, req, res, self.check_perm
            )
            return res.sendresponse()

        # If PATH_INFO is present (signaled by ``req.dispatchpath`` having
        # a value), we use it. Otherwise fall back to the query string.
        if req.dispatchpath is not None:
            query = req.dispatchpath
        else:
            query = req.querystring.partition('&')[0].partition(';')[0]

        # translate user-visible url structure to internal structure

        args = query.split('/', 2)
        if 'cmd' not in req.qsparams and args and args[0]:
            cmd = args.pop(0)
            style = cmd.rfind('-')
            if style != -1:
                req.qsparams['style'] = cmd[:style]
                cmd = cmd[style + 1 :]

            # avoid accepting e.g. style parameter as command
            if util.safehasattr(webcommands, cmd):
                req.qsparams['cmd'] = cmd

            if cmd == 'static':
                req.qsparams['file'] = '/'.join(args)
            else:
                if args and args[0]:
                    node = args.pop(0).replace('%2F', '/')
                    req.qsparams['node'] = node
                if args:
                    if 'file' in req.qsparams:
                        del req.qsparams['file']
                    for a in args:
                        req.qsparams.add('file', a)

            ua = req.headers.get('User-Agent', '')
            if cmd == 'rev' and 'mercurial' in ua:
                req.qsparams['style'] = 'raw'

            if cmd == 'archive':
                fn = req.qsparams['node']
                for type_, spec in webutil.archivespecs.iteritems():
                    ext = spec[2]
                    if fn.endswith(ext):
                        req.qsparams['node'] = fn[: -len(ext)]
                        req.qsparams['type'] = type_
        else:
            cmd = req.qsparams.get('cmd', '')

        # process the web interface request

        try:
            rctx.tmpl = rctx.templater(req)
            ctype = rctx.tmpl.render(
                'mimetype', {'encoding': encoding.encoding}
            )

            # check read permissions non-static content
            if cmd != 'static':
                self.check_perm(rctx, req, None)

            if cmd == '':
                req.qsparams['cmd'] = rctx.tmpl.render('default', {})
                cmd = req.qsparams['cmd']

            # Don't enable caching if using a CSP nonce because then it wouldn't
            # be a nonce.
            if rctx.configbool('web', 'cache') and not rctx.nonce:
                tag = 'W/"%d"' % self.mtime
                if req.headers.get('If-None-Match') == tag:
                    res.status = '304 Not Modified'
                    # Content-Type may be defined globally. It isn't valid on a
                    # 304, so discard it.
                    try:
                        del res.headers[b'Content-Type']
                    except KeyError:
                        pass
                    # Response body not allowed on 304.
                    res.setbodybytes('')
                    return res.sendresponse()

                res.headers['ETag'] = tag

            if cmd not in webcommands.__all__:
                msg = 'no such method: %s' % cmd
                raise ErrorResponse(HTTP_BAD_REQUEST, msg)
            else:
                # Set some globals appropriate for web handlers. Commands can
                # override easily enough.
                res.status = '200 Script output follows'
                res.headers['Content-Type'] = ctype
                return getattr(webcommands, cmd)(rctx)

        except (error.LookupError, error.RepoLookupError) as err:
            msg = pycompat.bytestr(err)
            if util.safehasattr(err, 'name') and not isinstance(
                err, error.ManifestLookupError
            ):
                msg = 'revision not found: %s' % err.name

            res.status = '404 Not Found'
            res.headers['Content-Type'] = ctype
            return rctx.sendtemplate('error', error=msg)
        except (error.RepoError, error.StorageError) as e:
            res.status = '500 Internal Server Error'
            res.headers['Content-Type'] = ctype
            return rctx.sendtemplate('error', error=pycompat.bytestr(e))
        except error.Abort as e:
            res.status = '403 Forbidden'
            res.headers['Content-Type'] = ctype
            return rctx.sendtemplate('error', error=pycompat.bytestr(e))
        except ErrorResponse as e:
            for k, v in e.headers:
                res.headers[k] = v
            res.status = statusmessage(e.code, pycompat.bytestr(e))
            res.headers['Content-Type'] = ctype
            return rctx.sendtemplate('error', error=pycompat.bytestr(e))

    def check_perm(self, rctx, req, op):
        for permhook in permhooks:
            permhook(rctx, req, op)

    See the repoview module for details.

    The option has been around undocumented since Mercurial 2.5, but no
    user ever asked about it. So we better keep it undocumented for now."""
    # experimental config: web.view
    viewconfig = repo.ui.config('web', 'view', untrusted=True)
    if viewconfig == 'all':
        return repo.unfiltered()
    elif viewconfig in repoview.filtertable:
        return repo.filtered(viewconfig)
    else:
        return repo.filtered('served')

43077:687b865b95ad

)


def getstyle(req, configfn, templatepath):
    styles = (
        req.qsparams.get(b'style', None),
        configfn(b'web', b'style'),
        b'paper',
    )
    return styles, templater.stylemap(styles, templatepath)


def makebreadcrumb(url, prefix=b''):
    '''Return a 'URL breadcrumb' list

    A 'URL breadcrumb' is a list of URL-name pairs,
    corresponding to each of the path items on a URL.
    This can be used to create path navigation entries.
    '''
    if url.endswith(b'/'):
        url = url[:-1]
    if prefix:
        url = b'/' + prefix + url
    relpath = url
    if relpath.startswith(b'/'):
        relpath = relpath[1:]

    breadcrumb = []
    urlel = url
    pathitems = [b''] + relpath.split(b'/')
    for pathel in reversed(pathitems):
        if not pathel or not urlel:
            break
        breadcrumb.append({b'url': urlel, b'name': pathel})
        urlel = os.path.dirname(urlel)
    return templateutil.mappinglist(reversed(breadcrumb))


class requestcontext(object):

        self.repo = repo
        self.reponame = app.reponame
        self.req = req
        self.res = res

        self.maxchanges = self.configint(b'web', b'maxchanges')
        self.stripecount = self.configint(b'web', b'stripes')
        self.maxshortchanges = self.configint(b'web', b'maxshortchanges')
        self.maxfiles = self.configint(b'web', b'maxfiles')
        self.allowpull = self.configbool(b'web', b'allow-pull')

        # we use untrusted=False to prevent a repo owner from using
        # web.templates in .hg/hgrc to get access to any file readable
        # by the user running the CGI script
        self.templatepath = self.config(b'web', b'templates', untrusted=False)

        # This object is more expensive to build than simple config values.
        # It is shared across requests. The app will replace the object
        # if it is updated. Since this is a reference and nothing should
        # modify the underlying object, it should be constant for the lifetime

        return webutil.archivelist(self.repo.ui, nodeid)

    def templater(self, req):
        # determine scheme, port and server name
        # this is needed to create absolute urls
        logourl = self.config(b'web', b'logourl')
        logoimg = self.config(b'web', b'logoimg')
        staticurl = (
            self.config(b'web', b'staticurl')
            or req.apppath.rstrip(b'/') + b'/static/'
        )
        if not staticurl.endswith(b'/'):
            staticurl += b'/'

        # figure out which style to use

        vars = {}
        styles, (style, mapfile) = getstyle(req, self.config, self.templatepath)
        if style == styles[0]:
            vars[b'style'] = style

        sessionvars = webutil.sessionvars(vars, b'?')

        if not self.reponame:
            self.reponame = (
                self.config(b'web', b'name', b'')
                or req.reponame
                or req.apppath
                or self.repo.root
            )

        filters = {}
        templatefilter = registrar.templatefilter(filters)

        @templatefilter(b'websub', intype=bytes)
        def websubfilter(text):
            return templatefilters.websub(text, self.websubtable)

        # create the templater
        # TODO: export all keywords: defaults = templatekw.keywords.copy()
        defaults = {
            b'url': req.apppath + b'/',
            b'logourl': logourl,
            b'logoimg': logoimg,
            b'staticurl': staticurl,
            b'urlbase': req.advertisedbaseurl,
            b'repo': self.reponame,
            b'encoding': encoding.encoding,
            b'sessionvars': sessionvars,
            b'pathdef': makebreadcrumb(req.apppath),
            b'style': style,
            b'nonce': self.nonce,
        }
        templatekeyword = registrar.templatekeyword(defaults)

        @templatekeyword(b'motd', requires=())
        def motd(context, mapping):
            yield self.config(b'web', b'motd')

        tres = formatter.templateresources(self.repo.ui, self.repo)
        tmpl = templater.templater.frommapfile(
            mapfile, filters=filters, defaults=defaults, resources=tres
        )

            r = hg.repository(u, repo)
        else:
            # we trust caller to give us a private copy
            r = repo

        r.ui.setconfig(b'ui', b'report_untrusted', b'off', b'hgweb')
        r.baseui.setconfig(b'ui', b'report_untrusted', b'off', b'hgweb')
        r.ui.setconfig(b'ui', b'nontty', b'true', b'hgweb')
        r.baseui.setconfig(b'ui', b'nontty', b'true', b'hgweb')
        # resolve file patterns relative to repo root
        r.ui.setconfig(b'ui', b'forcecwd', r.root, b'hgweb')
        r.baseui.setconfig(b'ui', b'forcecwd', r.root, b'hgweb')
        # it's unlikely that we can replace signal handlers in WSGI server,
        # and mod_wsgi issues a big warning. a plain hgweb process (with no
        # threading) could replace signal handlers, but we don't bother
        # conditionally enabling it.
        r.ui.setconfig(b'ui', b'signal-safe-lock', b'false', b'hgweb')
        r.baseui.setconfig(b'ui', b'signal-safe-lock', b'false', b'hgweb')
        # displaying bundling progress bar while serving feel wrong and may
        # break some wsgi implementation.
        r.ui.setconfig(b'progress', b'disable', b'true', b'hgweb')
        r.baseui.setconfig(b'progress', b'disable', b'true', b'hgweb')
        self._repos = [hg.cachedlocalrepo(self._webifyrepo(r))]
        self._lastrepo = self._repos[0]
        hook.redirect(True)
        self.reponame = name

        """Start a server from CGI environment.

        Modern servers should be using WSGI and should avoid this
        method, if possible.
        """
        if not encoding.environ.get(b'GATEWAY_INTERFACE', b'').startswith(
            b"CGI/1."
        ):
            raise RuntimeError(
                b"This function is only intended to be "
                b"called while running as a CGI script."
            )
        wsgicgi.launch(self)

    def __call__(self, env, respond):
        """Run the WSGI application.

        This is typically only called by Mercurial. External consumers
        should be using instances of this class as the WSGI application.
        """
        with self._obtainrepo() as repo:
            profile = repo.ui.configbool(b'profiling', b'enabled')
            with profiling.profile(repo.ui, enabled=profile):
                for r in self._runwsgi(req, res, repo):
                    yield r

    def _runwsgi(self, req, res, repo):
        rctx = requestcontext(self, repo, req, res)

        # This state is global across all threads.
        encoding.encoding = rctx.config(b'web', b'encoding')
        rctx.repo.ui.environ = req.rawenv

        if rctx.csp:
            # hgwebdir may have added CSP header. Since we generate our own,
            # replace it.
            res.headers[b'Content-Security-Policy'] = rctx.csp

        # /api/* is reserved for various API implementations. Dispatch
        # accordingly. But URL paths can conflict with subrepos and virtual
        # repos in hgwebdir. So until we have a workaround for this, only
        # expose the URLs if the feature is enabled.
        apienabled = rctx.repo.ui.configbool(b'experimental', b'web.apiserver')
        if apienabled and req.dispatchparts and req.dispatchparts[0] == b'api':
            wireprotoserver.handlewsgiapirequest(
                rctx, req, res, self.check_perm
            )
            return res.sendresponse()

        # If PATH_INFO is present (signaled by ``req.dispatchpath`` having
        # a value), we use it. Otherwise fall back to the query string.
        if req.dispatchpath is not None:
            query = req.dispatchpath
        else:
            query = req.querystring.partition(b'&')[0].partition(b';')[0]

        # translate user-visible url structure to internal structure

        args = query.split(b'/', 2)
        if b'cmd' not in req.qsparams and args and args[0]:
            cmd = args.pop(0)
            style = cmd.rfind(b'-')
            if style != -1:
                req.qsparams[b'style'] = cmd[:style]
                cmd = cmd[style + 1 :]

            # avoid accepting e.g. style parameter as command
            if util.safehasattr(webcommands, cmd):
                req.qsparams[b'cmd'] = cmd

            if cmd == b'static':
                req.qsparams[b'file'] = b'/'.join(args)
            else:
                if args and args[0]:
                    node = args.pop(0).replace(b'%2F', b'/')
                    req.qsparams[b'node'] = node
                if args:
                    if b'file' in req.qsparams:
                        del req.qsparams[b'file']
                    for a in args:
                        req.qsparams.add(b'file', a)

            ua = req.headers.get(b'User-Agent', b'')
            if cmd == b'rev' and b'mercurial' in ua:
                req.qsparams[b'style'] = b'raw'

            if cmd == b'archive':
                fn = req.qsparams[b'node']
                for type_, spec in webutil.archivespecs.iteritems():
                    ext = spec[2]
                    if fn.endswith(ext):
                        req.qsparams[b'node'] = fn[: -len(ext)]
                        req.qsparams[b'type'] = type_
        else:
            cmd = req.qsparams.get(b'cmd', b'')

        # process the web interface request

        try:
            rctx.tmpl = rctx.templater(req)
            ctype = rctx.tmpl.render(
                b'mimetype', {b'encoding': encoding.encoding}
            )

            # check read permissions non-static content
            if cmd != b'static':
                self.check_perm(rctx, req, None)

            if cmd == b'':
                req.qsparams[b'cmd'] = rctx.tmpl.render(b'default', {})
                cmd = req.qsparams[b'cmd']

            # Don't enable caching if using a CSP nonce because then it wouldn't
            # be a nonce.
            if rctx.configbool(b'web', b'cache') and not rctx.nonce:
                tag = b'W/"%d"' % self.mtime
                if req.headers.get(b'If-None-Match') == tag:
                    res.status = b'304 Not Modified'
                    # Content-Type may be defined globally. It isn't valid on a
                    # 304, so discard it.
                    try:
                        del res.headers[b'Content-Type']
                    except KeyError:
                        pass
                    # Response body not allowed on 304.
                    res.setbodybytes(b'')
                    return res.sendresponse()

                res.headers[b'ETag'] = tag

            if cmd not in webcommands.__all__:
                msg = b'no such method: %s' % cmd
                raise ErrorResponse(HTTP_BAD_REQUEST, msg)
            else:
                # Set some globals appropriate for web handlers. Commands can
                # override easily enough.
                res.status = b'200 Script output follows'
                res.headers[b'Content-Type'] = ctype
                return getattr(webcommands, cmd)(rctx)

        except (error.LookupError, error.RepoLookupError) as err:
            msg = pycompat.bytestr(err)
            if util.safehasattr(err, b'name') and not isinstance(
                err, error.ManifestLookupError
            ):
                msg = b'revision not found: %s' % err.name

            res.status = b'404 Not Found'
            res.headers[b'Content-Type'] = ctype
            return rctx.sendtemplate(b'error', error=msg)
        except (error.RepoError, error.StorageError) as e:
            res.status = b'500 Internal Server Error'
            res.headers[b'Content-Type'] = ctype
            return rctx.sendtemplate(b'error', error=pycompat.bytestr(e))
        except error.Abort as e:
            res.status = b'403 Forbidden'
            res.headers[b'Content-Type'] = ctype
            return rctx.sendtemplate(b'error', error=pycompat.bytestr(e))
        except ErrorResponse as e:
            for k, v in e.headers:
                res.headers[k] = v
            res.status = statusmessage(e.code, pycompat.bytestr(e))
            res.headers[b'Content-Type'] = ctype
            return rctx.sendtemplate(b'error', error=pycompat.bytestr(e))

    def check_perm(self, rctx, req, op):
        for permhook in permhooks:
            permhook(rctx, req, op)

    See the repoview module for details.

    The option has been around undocumented since Mercurial 2.5, but no
    user ever asked about it. So we better keep it undocumented for now."""
    # experimental config: web.view
    viewconfig = repo.ui.config(b'web', b'view', untrusted=True)
    if viewconfig == b'all':
        return repo.unfiltered()
    elif viewconfig in repoview.filtertable:
        return repo.filtered(viewconfig)
    else:
        return repo.filtered(b'served')