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

import gzip
import shutil
import logging
import tempfile
import urllib.parse

from webob.exc import HTTPNotFound

import rhodecode
from rhodecode.apps._base import ADMIN_PREFIX
from rhodecode.lib.middleware.utils import get_path_info
from rhodecode.lib.middleware.appenlight import wrap_in_appenlight_if_enabled
from rhodecode.lib.middleware.simplegit import SimpleGit, GIT_PROTO_PAT
from rhodecode.lib.middleware.simplehg import SimpleHg
from rhodecode.lib.middleware.simplesvn import SimpleSvn
from rhodecode.lib.str_utils import safe_str
from rhodecode.model.settings import VcsSettingsModel


log = logging.getLogger(__name__)

VCS_TYPE_KEY = "_rc_vcs_type"
VCS_TYPE_SKIP = "_rc_vcs_skip"


def is_git(environ):
    """
    Returns True if requests should be handled by GIT wsgi middleware
    """
    path_info = get_path_info(environ)
    is_git_path = GIT_PROTO_PAT.match(path_info)
    log.debug("request path: `%s` detected as GIT PROTOCOL %s", path_info, is_git_path is not None)

    return is_git_path


def is_hg(environ):
    """
    Returns True if requests target is mercurial server - header
    ``HTTP_ACCEPT`` of such request would start with ``application/mercurial``.
    """
    is_hg_path = False

    http_accept = environ.get("HTTP_ACCEPT")

    if http_accept and http_accept.startswith("application/mercurial"):
        query = urllib.parse.parse_qs(environ["QUERY_STRING"])
        if "cmd" in query:
            is_hg_path = True

    path_info = get_path_info(environ)
    log.debug("request path: `%s` detected as HG PROTOCOL %s", path_info, is_hg_path)

    return is_hg_path


def is_svn(environ):
    """
    Returns True if requests target is Subversion server
    """

    http_dav = environ.get("HTTP_DAV", "")
    magic_path_segment = rhodecode.CONFIG.get("rhodecode_subversion_magic_path", "/!svn")
    path_info = get_path_info(environ)
    req_method = environ["REQUEST_METHOD"]

    is_svn_path = (
        "subversion" in http_dav or magic_path_segment in path_info or req_method in ["PROPFIND", "PROPPATCH", "HEAD"]
    )
    log.debug("request path: `%s` detected as SVN PROTOCOL %s", path_info, is_svn_path)

    return is_svn_path


class GunzipMiddleware(object):
    """
    WSGI middleware that unzips gzip-encoded requests before
    passing on to the underlying application.
    """

    def __init__(self, application):
        self.app = application

    def __call__(self, environ, start_response):
        accepts_encoding_header = safe_str(environ.get("HTTP_CONTENT_ENCODING", ""))

        if "gzip" in accepts_encoding_header:
            log.debug("gzip detected, now running gunzip wrapper")
            wsgi_input = environ["wsgi.input"]

            if not hasattr(environ["wsgi.input"], "seek"):
                # The gzip implementation in the standard library of Python 2.x
                # requires the '.seek()' and '.tell()' methods to be available
                # on the input stream.  Read the data into a temporary file to
                # work around this limitation.

                wsgi_input = tempfile.SpooledTemporaryFile(64 * 1024 * 1024)
                shutil.copyfileobj(environ["wsgi.input"], wsgi_input)
                wsgi_input.seek(0)

            environ["wsgi.input"] = gzip.GzipFile(fileobj=wsgi_input, mode="r")
            # since we "Ungzipped" the content we say now it's no longer gzip
            # content encoding
            del environ["HTTP_CONTENT_ENCODING"]

            # content length has changes ? or i'm not sure
            if "CONTENT_LENGTH" in environ:
                del environ["CONTENT_LENGTH"]
        else:
            log.debug("content not gzipped, gzipMiddleware passing request further")
        return self.app(environ, start_response)


def is_vcs_call(environ):
    if VCS_TYPE_KEY in environ:
        raw_type = environ[VCS_TYPE_KEY]
        return raw_type and raw_type != VCS_TYPE_SKIP
    return False


def detect_vcs_request(environ, backends):
    checks = {
        "hg": (is_hg, SimpleHg),
        "git": (is_git, SimpleGit),
        "svn": (is_svn, SimpleSvn),
    }
    handler = None
    # List of path views first chunk we don't do any checks
    white_list = [
        # favicon often requested by browsers
        "favicon.ico",
        # static files no detection
        "_static++",
        # debug-toolbar
        "_debug_toolbar++",
        # e.g /_file_store/download
        "_file_store++",
        # login
        f"{ADMIN_PREFIX}/login",
        f"{ADMIN_PREFIX}/logout",
        # 2fa
        f"{ADMIN_PREFIX}/check_2fa",
        f"{ADMIN_PREFIX}/setup_2fa",
        # _admin/api is safe too
        f"{ADMIN_PREFIX}/api",
        # _admin/gist is safe too
        f"{ADMIN_PREFIX}/gists++",
        # _admin/my_account is safe too
        f"{ADMIN_PREFIX}/my_account++",
        # skip ops ping, status
        f"{ADMIN_PREFIX}/ops/ping",
        f"{ADMIN_PREFIX}/ops/status",
        # full channelstream connect should be VCS skipped
        f"{ADMIN_PREFIX}/channelstream/connect",
        "++/repo_creating_check",
    ]

    path_info = get_path_info(environ)
    path_url = path_info.lstrip("/")
    req_method = environ.get("REQUEST_METHOD")

    for item in white_list:
        item = item.lstrip("/")

        if item.endswith("++") and path_url.startswith(item[:-2]):
            log.debug("path `%s` in whitelist (match:%s), skipping...", path_url, item)
            return handler
        if item.startswith("++") and path_url.endswith(item[2:]):
            log.debug("path `%s` in whitelist (match:%s), skipping...", path_url, item)
            return handler
        if item == path_url:
            log.debug("path `%s` in whitelist (match:%s), skipping...", path_url, item)
            return handler

    if VCS_TYPE_KEY in environ:
        raw_type = environ[VCS_TYPE_KEY]
        if raw_type == VCS_TYPE_SKIP:
            log.debug("got `skip` marker for vcs detection, skipping...")
            return handler

        _check, handler = checks.get(raw_type) or [None, None]
        if handler:
            log.debug("got handler:%s from environ", handler)

    if not handler:
        log.debug(
            "request start: checking if request for `%s:%s` is of VCS type in order: %s", req_method, path_url, backends
        )
        for vcs_type in backends:
            vcs_check, _handler = checks[vcs_type]
            if vcs_check(environ):
                log.debug("vcs handler found %s", _handler)
                handler = _handler
                break

    return handler


class VCSMiddleware(object):
    def __init__(self, app, registry, config, appenlight_client):
        self.application = app
        self.registry = registry
        self.config = config
        self.appenlight_client = appenlight_client
        self.use_gzip = True
        # order in which we check the middlewares, based on vcs.backends config
        self.check_middlewares = config["vcs.backends"]

    def vcs_config(self, repo_name=None):
        """
        returns serialized VcsSettings
        """
        try:
            return VcsSettingsModel(repo=repo_name).get_ui_settings_as_config_obj()
        except Exception:
            pass

    def wrap_in_gzip_if_enabled(self, app, config):
        if self.use_gzip:
            app = GunzipMiddleware(app)
        return app

    def _get_handler_app(self, environ):
        app = None
        log.debug("VCSMiddleware: detecting vcs type.")
        handler = detect_vcs_request(environ, self.check_middlewares)
        if handler:
            app = handler(self.config, self.registry)

        return app

    def __call__(self, environ, start_response):
        # check if we handle one of interesting protocols, optionally extract
        # specific vcsSettings and allow changes of how things are wrapped
        vcs_handler = self._get_handler_app(environ)
        if vcs_handler:
            # translate the _REPO_ID into real repo NAME for usage
            # in middleware

            path_info = get_path_info(environ)
            environ["PATH_INFO"] = vcs_handler._get_by_id(path_info)

            # Set acl, url and vcs repo names.
            vcs_handler.set_repo_names(environ)

            # register repo config back to the handler
            vcs_conf = self.vcs_config(vcs_handler.acl_repo_name)
            # maybe damaged/non existent settings. We still want to
            # pass that point to validate on is_valid_and_existing_repo
            # and return proper HTTP Code back to client
            if vcs_conf:
                vcs_handler.repo_vcs_config = vcs_conf

            # check for type, presence in database and on filesystem
            if not vcs_handler.is_valid_and_existing_repo(
                vcs_handler.acl_repo_name, vcs_handler.base_path, vcs_handler.SCM
            ):
                return HTTPNotFound()(environ, start_response)

            environ["REPO_NAME"] = vcs_handler.url_repo_name

            # Wrap handler in middlewares if they are enabled.
            vcs_handler = self.wrap_in_gzip_if_enabled(vcs_handler, self.config)
            vcs_handler, _ = wrap_in_appenlight_if_enabled(vcs_handler, self.config, self.appenlight_client)

            return vcs_handler(environ, start_response)

        return self.application(environ, start_response)
