# RhodeCode VCSServer provides access to different vcs backends via network. # Copyright (C) 2014-2023 RhodeCode GmbH # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 3 of the License, or # (at your option) any later version. # # 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 General Public License # along with this program; if not, write to the Free Software Foundation, # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA import os import logging import itertools import mercurial import mercurial.error import mercurial.wireprotoserver import mercurial.hgweb.common import mercurial.hgweb.hgweb_mod import webob.exc from vcsserver import pygrack, exceptions, settings, git_lfs from vcsserver.str_utils import ascii_bytes, safe_bytes log = logging.getLogger(__name__) # propagated from mercurial documentation HG_UI_SECTIONS = [ 'alias', 'auth', 'decode/encode', 'defaults', 'diff', 'email', 'extensions', 'format', 'merge-patterns', 'merge-tools', 'hooks', 'http_proxy', 'smtp', 'patch', 'paths', 'profiling', 'server', 'trusted', 'ui', 'web', ] class HgWeb(mercurial.hgweb.hgweb_mod.hgweb): """Extension of hgweb that simplifies some functions.""" def _get_view(self, repo): """Views are not supported.""" return repo def loadsubweb(self): """The result is only used in the templater method which is not used.""" return None def run(self): """Unused function so raise an exception if accidentally called.""" raise NotImplementedError def templater(self, req): """Function used in an unreachable code path. This code is unreachable because we guarantee that the HTTP request, corresponds to a Mercurial command. See the is_hg method. So, we are never going to get a user-visible url. """ raise NotImplementedError def archivelist(self, nodeid): """Unused function so raise an exception if accidentally called.""" raise NotImplementedError def __call__(self, environ, start_response): """Run the WSGI application. This may be called by multiple threads. """ from mercurial.hgweb import request as requestmod req = requestmod.parserequestfromenv(environ) res = requestmod.wsgiresponse(req, start_response) gen = self.run_wsgi(req, res) first_chunk = None try: data = next(gen) def first_chunk(): yield data except StopIteration: pass if first_chunk: return itertools.chain(first_chunk(), gen) return gen def _runwsgi(self, req, res, repo): cmd = req.qsparams.get(b'cmd', '') if not mercurial.wireprotoserver.iscmd(cmd): # NOTE(marcink): for unsupported commands, we return bad request # internally from HG log.warning('cmd: `%s` is not supported by the mercurial wireprotocol v1', cmd) from mercurial.hgweb.common import statusmessage res.status = statusmessage(mercurial.hgweb.common.HTTP_BAD_REQUEST) res.setbodybytes(b'') return res.sendresponse() return super()._runwsgi(req, res, repo) def sanitize_hg_ui(baseui): # NOTE(marcink): since python3 hgsubversion is deprecated. # From old installations we might still have this set enabled # we explicitly remove this now here to make sure it wont propagate further if baseui.config(b'extensions', b'hgsubversion') is not None: for cfg in (baseui._ocfg, baseui._tcfg, baseui._ucfg): if b'extensions' in cfg: if b'hgsubversion' in cfg[b'extensions']: del cfg[b'extensions'][b'hgsubversion'] def make_hg_ui_from_config(repo_config): baseui = mercurial.ui.ui() # clean the baseui object baseui._ocfg = mercurial.config.config() baseui._ucfg = mercurial.config.config() baseui._tcfg = mercurial.config.config() for section, option, value in repo_config: baseui.setconfig( ascii_bytes(section, allow_bytes=True), ascii_bytes(option, allow_bytes=True), ascii_bytes(value, allow_bytes=True)) # make our hgweb quiet so it doesn't print output baseui.setconfig(b'ui', b'quiet', b'true') return baseui def update_hg_ui_from_hgrc(baseui, repo_path): path = os.path.join(repo_path, '.hg', 'hgrc') if not os.path.isfile(path): log.debug('hgrc file is not present at %s, skipping...', path) return log.debug('reading hgrc from %s', path) cfg = mercurial.config.config() cfg.read(ascii_bytes(path)) for section in HG_UI_SECTIONS: for k, v in cfg.items(section): log.debug('settings ui from file: [%s] %s=%s', section, k, v) baseui.setconfig( ascii_bytes(section, allow_bytes=True), ascii_bytes(k, allow_bytes=True), ascii_bytes(v, allow_bytes=True)) def create_hg_wsgi_app(repo_path, repo_name, config): """ Prepares a WSGI application to handle Mercurial requests. :param config: is a list of 3-item tuples representing a ConfigObject (it is the serialized version of the config object). """ log.debug("Creating Mercurial WSGI application") baseui = make_hg_ui_from_config(config) update_hg_ui_from_hgrc(baseui, repo_path) sanitize_hg_ui(baseui) try: return HgWeb(safe_bytes(repo_path), name=safe_bytes(repo_name), baseui=baseui) except mercurial.error.RequirementError as e: raise exceptions.RequirementException(e)(e) class GitHandler: """ Handler for Git operations like push/pull etc """ def __init__(self, repo_location, repo_name, git_path, update_server_info, extras): if not os.path.isdir(repo_location): raise OSError(repo_location) self.content_path = repo_location self.repo_name = repo_name self.repo_location = repo_location self.extras = extras self.git_path = git_path self.update_server_info = update_server_info def __call__(self, environ, start_response): app = webob.exc.HTTPNotFound() candidate_paths = ( self.content_path, os.path.join(self.content_path, '.git')) for content_path in candidate_paths: try: app = pygrack.GitRepository( self.repo_name, content_path, self.git_path, self.update_server_info, self.extras) break except OSError: continue return app(environ, start_response) def create_git_wsgi_app(repo_path, repo_name, config): """ Creates a WSGI application to handle Git requests. :param config: is a dictionary holding the extras. """ git_path = settings.GIT_EXECUTABLE update_server_info = config.pop('git_update_server_info') app = GitHandler( repo_path, repo_name, git_path, update_server_info, config) return app class GitLFSHandler: """ Handler for Git LFS operations """ def __init__(self, repo_location, repo_name, git_path, update_server_info, extras): if not os.path.isdir(repo_location): raise OSError(repo_location) self.content_path = repo_location self.repo_name = repo_name self.repo_location = repo_location self.extras = extras self.git_path = git_path self.update_server_info = update_server_info def get_app(self, git_lfs_enabled, git_lfs_store_path, git_lfs_http_scheme): app = git_lfs.create_app(git_lfs_enabled, git_lfs_store_path, git_lfs_http_scheme) return app def create_git_lfs_wsgi_app(repo_path, repo_name, config): git_path = settings.GIT_EXECUTABLE update_server_info = config.pop('git_update_server_info') git_lfs_enabled = config.pop('git_lfs_enabled') git_lfs_store_path = config.pop('git_lfs_store_path') git_lfs_http_scheme = config.pop('git_lfs_http_scheme', 'http') app = GitLFSHandler( repo_path, repo_name, git_path, update_server_info, config) return app.get_app(git_lfs_enabled, git_lfs_store_path, git_lfs_http_scheme)