|
|
# 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)
|
|
|
|
