unrhodecode/rhodecode/lib/middleware/simplehg.py

# Copyright (C) 2010-2024 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/

"""
SimpleHG middleware for handling mercurial protocol request
(push/clone etc.). It's implemented with basic auth function
"""

import copy
import logging
import urllib.parse
import urllib.request
import urllib.parse
import urllib.error

from rhodecode.lib import utils
from rhodecode.lib.ext_json import json
from rhodecode.lib.middleware import simplevcs
from rhodecode.lib.middleware.utils import get_path_info
from rhodecode.lib.str_utils import safe_str

log = logging.getLogger(__name__)


class SimpleHg(simplevcs.SimpleVCS):
    SCM = "hg"

    def _get_repository_name(self, environ):
        """
        Gets repository name out of PATH_INFO header

        :param environ: environ where PATH_INFO is stored
        """
        repo_name = get_path_info(environ)
        if repo_name and repo_name.startswith("/"):
            # remove only the first leading /
            repo_name = repo_name[1:]
        return repo_name.rstrip("/")

    _ACTION_MAPPING = {
        "between": "pull",
        "branches": "pull",
        "branchmap": "pull",
        "capabilities": "pull",
        "changegroup": "pull",
        "changegroupsubset": "pull",
        "changesetdata": "pull",
        "clonebundles": "pull",
        "clonebundles_manifest": "pull",
        "debugwireargs": "pull",
        "filedata": "pull",
        "getbundle": "pull",
        "heads": "pull",
        "hello": "pull",
        "known": "pull",
        "listkeys": "pull",
        "lookup": "pull",
        "manifestdata": "pull",
        "narrow_widen": "pull",
        "protocaps": "pull",
        "stream_out": "pull",
        # largefiles
        "getlfile": "pull",
        "putlfile": "push",
        "statlfile": "pull",
        "lheads": "pull",
        # evolve
        "evoext_obshashrange_v1": "pull",
        "evoext_obshash": "pull",
        "evoext_obshash1": "pull",
        "unbundle": "push",
        "pushkey": "push",
    }

    @classmethod
    def _get_xarg_headers(cls, environ):
        i = 1
        chunks = []  # gather chunks stored in multiple 'hgarg_N'
        while True:
            head = environ.get(f"HTTP_X_HGARG_{i}")
            if not head:
                break
            i += 1
            chunks.append(urllib.parse.unquote_plus(head))
        full_arg = "".join(chunks)
        pref = "cmds="
        if full_arg.startswith(pref):
            # strip the cmds= header defining our batch commands
            full_arg = full_arg[len(pref) :]
        cmds = full_arg.split(";")
        return cmds

    @classmethod
    def _get_batch_cmd(cls, environ):
        """
        Handle batch command send commands. Those are ';' separated commands
        sent by batch command that server needs to execute. We need to extract
        those, and map them to our ACTION_MAPPING to get all push/pull commands
        specified in the batch
        """
        default = "push"
        batch_cmds = []

        try:
            httppostargs_enabled = True
            post_args_size = environ.get("HTTP_X_HGARGS_POST")
            if post_args_size and httppostargs_enabled:
                # a new proto when httppostargs is enabled
                response_data = copy.copy(environ["wsgi.input"])
                cmds = [safe_str(response_data.read(post_args_size))]
            else:
                # old way... from headers
                cmds = cls._get_xarg_headers(environ)

            for pair in cmds:
                parts = pair.split(" ", 1)
                if len(parts) != 2:
                    continue
                # entry should be in a format `key ARGS`
                cmd, args = parts
                action = cls._ACTION_MAPPING.get(cmd, default)
                batch_cmds.append(action)
        except Exception:
            log.exception("Failed to extract batch commands operations")

        # in case we failed, (e.g malformed data) assume it's PUSH sub-command
        # for safety
        return batch_cmds or [default]

    def _get_action(self, environ):
        """
        Maps mercurial request commands into a pull or push command.
        In case of unknown/unexpected data, it returns 'push' to be safe.

        :param environ:
        """
        default = "push"
        query = urllib.parse.parse_qs(environ["QUERY_STRING"], keep_blank_values=True)

        if "cmd" in query:
            cmd = query["cmd"][0]
            if cmd == "batch":
                cmds = self._get_batch_cmd(environ)
                if "push" in cmds:
                    return "push"
                else:
                    return "pull"
            return self._ACTION_MAPPING.get(cmd, default)

        return default

    def _create_wsgi_app(self, repo_path, repo_name, config):
        return self.scm_app.create_hg_wsgi_app(repo_path, repo_name, config)

    def _create_config(self, extras, repo_name, scheme="http"):
        config = utils.make_db_config(repo=repo_name)
        config.set("rhodecode", "RC_SCM_DATA", json.dumps(extras))

        return config.serialize()