mercurial/hgweb/hgwebdir_mod.py

36811:8e1556ac01bb

    get_mtime,
    ismember,
    paritygen,
    staticfile,
)
from .request import wsgirequest

from .. import (
    configitems,
    encoding,
    error,

    util,
)

from . import (
    hgweb_mod,
    webutil,
    wsgicgi,
)
from ..utils import dateutil

            raise RuntimeError("This function is only intended to be "
                               "called while running as a CGI script.")
        wsgicgi.launch(self)

    def __call__(self, env, respond):
        req = wsgirequest(env, respond)
        return self.run_wsgi(req)

    def read_allowed(self, ui, req):
        """Check allow_read and deny_read config options of a repo's ui object
        to determine user permissions.  By default, with neither option set (or
        both empty), allow all users to read the repo.  There are two ways a
        user can be denied read access:  (1) deny_read is not empty, and the
        user is unauthenticated or deny_read contains user (or *), and (2)
        allow_read is not empty and the user is not in allow_read.  Return True
        if user is allowed to read the repo, else return False."""

        user = req.env.get('REMOTE_USER')

        deny_read = ui.configlist('web', 'deny_read', untrusted=True)
        if deny_read and (not user or ismember(ui, user, deny_read)):
            return False

        if (not allow_read) or ismember(ui, user, allow_read):
            return True

        return False

    def run_wsgi(self, req):
        profile = self.ui.configbool('profiling', 'enabled')
        with profiling.profile(self.ui, enabled=profile):
            for r in self._runwsgi(req):
                yield r

    def _runwsgi(self, req):
        try:
            self.refresh()

            csp, nonce = cspvalues(self.ui)
            if csp:
                req.headers.append(('Content-Security-Policy', csp))

            virtual = req.env.get("PATH_INFO", "").strip('/')
            tmpl = self.templater(req, nonce)
            ctype = tmpl('mimetype', encoding=encoding.encoding)
            ctype = templater.stringify(ctype)

            # a static file
            if virtual.startswith('static/') or 'static' in req.form:
                if virtual.startswith('static/'):
                    fname = virtual[7:]
                else:
                    fname = req.form['static'][0]
                static = self.ui.config("web", "static", None,
                                        untrusted=False)
                if not static:
                    tp = self.templatepath or templater.templatepaths()
                    if isinstance(tp, str):
                        tp = [tp]
                    static = [os.path.join(p, 'static') for p in tp]
                staticfile(static, fname, req)
                return []

            # top-level index

            repos = dict(self.repos)

            if (not virtual or virtual == 'index') and virtual not in repos:
                req.respond(HTTP_OK, ctype)
                return self.makeindex(req, tmpl)

            # nested indexes and hgwebs

            if virtual.endswith('/index') and virtual not in repos:
                subdir = virtual[:-len('index')]
                if any(r.startswith(subdir) for r in repos):
                    req.respond(HTTP_OK, ctype)
                    return self.makeindex(req, tmpl, subdir)

            def _virtualdirs():
                # Check the full virtual path, each parent, and the root ('')
                if virtual != '':
                    yield virtual

                yield ''

            for virtualrepo in _virtualdirs():
                real = repos.get(virtualrepo)
                if real:
                    req.env['REPO_NAME'] = virtualrepo
                    try:
                        # ensure caller gets private copy of ui
                        repo = hg.repository(self.ui.copy(), real)
                        return hgweb_mod.hgweb(repo).run_wsgi(req)
                    except IOError as inst:
                        msg = encoding.strtolocal(inst.strerror)
                        raise ErrorResponse(HTTP_SERVER_ERROR, msg)
                    except error.RepoError as inst:
                        raise ErrorResponse(HTTP_SERVER_ERROR, bytes(inst))

            # browse subdirectories
            subdir = virtual + '/'
            if [r for r in repos if r.startswith(subdir)]:
                req.respond(HTTP_OK, ctype)
                return self.makeindex(req, tmpl, subdir)

            # prefixes not found
            req.respond(HTTP_NOT_FOUND, ctype)
            return tmpl("notfound", repo=virtual)

        except ErrorResponse as err:
            req.respond(err, ctype)
            return tmpl('error', error=err.message or '')
        finally:
            tmpl = None

    def makeindex(self, req, tmpl, subdir=""):

        def archivelist(ui, nodeid, url):
            allowed = ui.configlist("web", "allow_archive", untrusted=True)
            archives = []
            for typ, spec in hgweb_mod.archivespecs.iteritems():

                        except (IOError, error.RepoError):
                            pass

                parts = [name]
                parts.insert(0, '/' + subdir.rstrip('/'))
                if req.env['SCRIPT_NAME']:
                    parts.insert(0, req.env['SCRIPT_NAME'])
                url = re.sub(r'/+', '/', '/'.join(parts) + '/')

                # show either a directory entry or a repository
                if directory:
                    # get the directory's time information

                    return u.config(section, name, default, untrusted=True)

                if u.configbool("web", "hidden", untrusted=True):
                    continue

                if not self.read_allowed(u, req):
                    continue

                # update time with local timezone
                try:
                    r = hg.repository(self.ui, path)

                yield row

        self.refresh()
        sortable = ["name", "description", "contact", "lastchange"]
        sortcolumn, descending = sortdefault
        if 'sort' in req.form:
            sortcolumn = req.form['sort'][0]
            descending = sortcolumn.startswith('-')
            if descending:
                sortcolumn = sortcolumn[1:]
            if sortcolumn not in sortable:
                sortcolumn = ""

                 "%s%s" % ((not descending and column == sortcolumn)
                            and "-" or "", column))
                for column in sortable]

        self.refresh()
        self.updatereqenv(req.env)

        return tmpl("index", entries=entries, subdir=subdir,
                    pathdef=hgweb_mod.makebreadcrumb('/' + subdir, self.prefix),
                    sortcolumn=sortcolumn, descending=descending,
                    **dict(sort))

    def templater(self, req, nonce):

        def motd(**map):
            if self.motd is not None:
                yield self.motd
            else:
                yield config('web', 'motd')

        def config(section, name, default=uimod._unset, untrusted=True):
            return self.ui.config(section, name, default, untrusted)

        self.updatereqenv(req.env)

        url = req.env.get('SCRIPT_NAME', '')
        if not url.endswith('/'):
            url += '/'

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

        start = r'&' if url[-1] == r'?' else r'?'

36812:b9b968e21f78

    get_mtime,
    ismember,
    paritygen,
    staticfile,
)

from .. import (
    configitems,
    encoding,
    error,

    util,
)

from . import (
    hgweb_mod,
    request as requestmod,
    webutil,
    wsgicgi,
)
from ..utils import dateutil

            raise RuntimeError("This function is only intended to be "
                               "called while running as a CGI script.")
        wsgicgi.launch(self)

    def __call__(self, env, respond):
        wsgireq = requestmod.wsgirequest(env, respond)
        return self.run_wsgi(wsgireq)

    def read_allowed(self, ui, wsgireq):
        """Check allow_read and deny_read config options of a repo's ui object
        to determine user permissions.  By default, with neither option set (or
        both empty), allow all users to read the repo.  There are two ways a
        user can be denied read access:  (1) deny_read is not empty, and the
        user is unauthenticated or deny_read contains user (or *), and (2)
        allow_read is not empty and the user is not in allow_read.  Return True
        if user is allowed to read the repo, else return False."""

        user = wsgireq.env.get('REMOTE_USER')

        deny_read = ui.configlist('web', 'deny_read', untrusted=True)
        if deny_read and (not user or ismember(ui, user, deny_read)):
            return False

        if (not allow_read) or ismember(ui, user, allow_read):
            return True

        return False

    def run_wsgi(self, wsgireq):
        profile = self.ui.configbool('profiling', 'enabled')
        with profiling.profile(self.ui, enabled=profile):
            for r in self._runwsgi(wsgireq):
                yield r

    def _runwsgi(self, wsgireq):
        try:
            self.refresh()

            csp, nonce = cspvalues(self.ui)
            if csp:
                wsgireq.headers.append(('Content-Security-Policy', csp))

            virtual = wsgireq.env.get("PATH_INFO", "").strip('/')
            tmpl = self.templater(wsgireq, nonce)
            ctype = tmpl('mimetype', encoding=encoding.encoding)
            ctype = templater.stringify(ctype)

            # a static file
            if virtual.startswith('static/') or 'static' in wsgireq.form:
                if virtual.startswith('static/'):
                    fname = virtual[7:]
                else:
                    fname = wsgireq.form['static'][0]
                static = self.ui.config("web", "static", None,
                                        untrusted=False)
                if not static:
                    tp = self.templatepath or templater.templatepaths()
                    if isinstance(tp, str):
                        tp = [tp]
                    static = [os.path.join(p, 'static') for p in tp]
                staticfile(static, fname, wsgireq)
                return []

            # top-level index

            repos = dict(self.repos)

            if (not virtual or virtual == 'index') and virtual not in repos:
                wsgireq.respond(HTTP_OK, ctype)
                return self.makeindex(wsgireq, tmpl)

            # nested indexes and hgwebs

            if virtual.endswith('/index') and virtual not in repos:
                subdir = virtual[:-len('index')]
                if any(r.startswith(subdir) for r in repos):
                    wsgireq.respond(HTTP_OK, ctype)
                    return self.makeindex(wsgireq, tmpl, subdir)

            def _virtualdirs():
                # Check the full virtual path, each parent, and the root ('')
                if virtual != '':
                    yield virtual

                yield ''

            for virtualrepo in _virtualdirs():
                real = repos.get(virtualrepo)
                if real:
                    wsgireq.env['REPO_NAME'] = virtualrepo
                    try:
                        # ensure caller gets private copy of ui
                        repo = hg.repository(self.ui.copy(), real)
                        return hgweb_mod.hgweb(repo).run_wsgi(wsgireq)
                    except IOError as inst:
                        msg = encoding.strtolocal(inst.strerror)
                        raise ErrorResponse(HTTP_SERVER_ERROR, msg)
                    except error.RepoError as inst:
                        raise ErrorResponse(HTTP_SERVER_ERROR, bytes(inst))

            # browse subdirectories
            subdir = virtual + '/'
            if [r for r in repos if r.startswith(subdir)]:
                wsgireq.respond(HTTP_OK, ctype)
                return self.makeindex(wsgireq, tmpl, subdir)

            # prefixes not found
            wsgireq.respond(HTTP_NOT_FOUND, ctype)
            return tmpl("notfound", repo=virtual)

        except ErrorResponse as err:
            wsgireq.respond(err, ctype)
            return tmpl('error', error=err.message or '')
        finally:
            tmpl = None

    def makeindex(self, wsgireq, tmpl, subdir=""):

        def archivelist(ui, nodeid, url):
            allowed = ui.configlist("web", "allow_archive", untrusted=True)
            archives = []
            for typ, spec in hgweb_mod.archivespecs.iteritems():

                        except (IOError, error.RepoError):
                            pass

                parts = [name]
                parts.insert(0, '/' + subdir.rstrip('/'))
                if wsgireq.env['SCRIPT_NAME']:
                    parts.insert(0, wsgireq.env['SCRIPT_NAME'])
                url = re.sub(r'/+', '/', '/'.join(parts) + '/')

                # show either a directory entry or a repository
                if directory:
                    # get the directory's time information

                    return u.config(section, name, default, untrusted=True)

                if u.configbool("web", "hidden", untrusted=True):
                    continue

                if not self.read_allowed(u, wsgireq):
                    continue

                # update time with local timezone
                try:
                    r = hg.repository(self.ui, path)

                yield row

        self.refresh()
        sortable = ["name", "description", "contact", "lastchange"]
        sortcolumn, descending = sortdefault
        if 'sort' in wsgireq.form:
            sortcolumn = wsgireq.form['sort'][0]
            descending = sortcolumn.startswith('-')
            if descending:
                sortcolumn = sortcolumn[1:]
            if sortcolumn not in sortable:
                sortcolumn = ""

                 "%s%s" % ((not descending and column == sortcolumn)
                            and "-" or "", column))
                for column in sortable]

        self.refresh()
        self.updatereqenv(wsgireq.env)

        return tmpl("index", entries=entries, subdir=subdir,
                    pathdef=hgweb_mod.makebreadcrumb('/' + subdir, self.prefix),
                    sortcolumn=sortcolumn, descending=descending,
                    **dict(sort))

    def templater(self, wsgireq, nonce):

        def motd(**map):
            if self.motd is not None:
                yield self.motd
            else:
                yield config('web', 'motd')

        def config(section, name, default=uimod._unset, untrusted=True):
            return self.ui.config(section, name, default, untrusted)

        self.updatereqenv(wsgireq.env)

        url = wsgireq.env.get('SCRIPT_NAME', '')
        if not url.endswith('/'):
            url += '/'

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

        start = r'&' if url[-1] == r'?' else r'?'