# Copyright 21 May 2005 - (c) 2005 Jake Edge <jake@edge2.net>

# Copyright 2005-2007 Matt Mackall <mpm@selenic.com>

#

# This software may be used and distributed according to the terms of the

# GNU General Public License version 2 or any later version.

from __future__ import absolute_import

import contextlib

import struct

import sys

import threading

from .i18n import _

from . import (

    encoding,

    error,

    hook,

    pycompat,

    util,

    wireproto,

    wireprototypes,

)

stringio = util.stringio

urlerr = util.urlerr

urlreq = util.urlreq

HTTP_OK = 200

HGTYPE = 'application/mercurial-0.1'

HGTYPE2 = 'application/mercurial-0.2'

HGERRTYPE = 'application/hg-error'

SSHV1 = wireprototypes.SSHV1

SSHV2 = wireprototypes.SSHV2

def decodevaluefromheaders(wsgireq, headerprefix):

    """Decode a long value from multiple HTTP request headers.

    Returns the value as a bytes, not a str.

    """

    chunks = []

    i = 1

    prefix = headerprefix.upper().replace(r'-', r'_')

    while True:

        v = wsgireq.env.get(r'HTTP_%s_%d' % (prefix, i))

        if v is None:

            break

        chunks.append(pycompat.bytesurl(v))

        i += 1

    return ''.join(chunks)

class httpv1protocolhandler(wireprototypes.baseprotocolhandler):

    def __init__(self, wsgireq, ui, checkperm):

        self._wsgireq = wsgireq

        self._ui = ui

        self._checkperm = checkperm

    @property

    def name(self):

        return 'http-v1'

    def getargs(self, args):

        knownargs = self._args()

        data = {}

        keys = args.split()

        for k in keys:

            if k == '*':

                star = {}

                for key in knownargs.keys():

                    if key != 'cmd' and key not in keys:

                        star[key] = knownargs[key][0]

                data['*'] = star

            else:

                data[k] = knownargs[k][0]

        return [data[k] for k in keys]

    def _args(self):

        args = util.rapply(pycompat.bytesurl, self._wsgireq.form.copy())

        postlen = int(self._wsgireq.env.get(r'HTTP_X_HGARGS_POST', 0))

        if postlen:

            args.update(urlreq.parseqs(

                self._wsgireq.read(postlen), keep_blank_values=True))

            return args

        argvalue = decodevaluefromheaders(self._wsgireq, r'X-HgArg')

        args.update(urlreq.parseqs(argvalue, keep_blank_values=True))

        return args

    def forwardpayload(self, fp):

        if r'HTTP_CONTENT_LENGTH' in self._wsgireq.env:

            length = int(self._wsgireq.env[r'HTTP_CONTENT_LENGTH'])

        else:

            length = int(self._wsgireq.env[r'CONTENT_LENGTH'])

        # If httppostargs is used, we need to read Content-Length

        # minus the amount that was consumed by args.

        length -= int(self._wsgireq.env.get(r'HTTP_X_HGARGS_POST', 0))

        for s in util.filechunkiter(self._wsgireq, limit=length):

            fp.write(s)

    @contextlib.contextmanager

    def mayberedirectstdio(self):

        oldout = self._ui.fout

        olderr = self._ui.ferr

        out = util.stringio()

        try:

            self._ui.fout = out

            self._ui.ferr = out

            yield out

        finally:

            self._ui.fout = oldout

            self._ui.ferr = olderr

    def client(self):

        return 'remote:%s:%s:%s' % (

            self._wsgireq.env.get('wsgi.url_scheme') or 'http',

            urlreq.quote(self._wsgireq.env.get('REMOTE_HOST', '')),

            urlreq.quote(self._wsgireq.env.get('REMOTE_USER', '')))

    def addcapabilities(self, repo, caps):

        caps.append('httpheader=%d' %

                    repo.ui.configint('server', 'maxhttpheaderlen'))

        if repo.ui.configbool('experimental', 'httppostargs'):

            caps.append('httppostargs')

        # FUTURE advertise 0.2rx once support is implemented

        # FUTURE advertise minrx and mintx after consulting config option

        caps.append('httpmediatype=0.1rx,0.1tx,0.2tx')

        compengines = wireproto.supportedcompengines(repo.ui, util.SERVERROLE)

        if compengines:

            comptypes = ','.join(urlreq.quote(e.wireprotosupport().name)

                                 for e in compengines)

            caps.append('compression=%s' % comptypes)

        return caps

    def checkperm(self, perm):

        return self._checkperm(perm)

# This method exists mostly so that extensions like remotefilelog can

# disable a kludgey legacy method only over http. As of early 2018,

# there are no other known users, so with any luck we can discard this

# hook if remotefilelog becomes a first-party extension.

def iscmd(cmd):

    return cmd in wireproto.commands

    """Parse the HTTP request for a wire protocol request.

    If the current request appears to be a wire protocol request, this

    function returns a dict with details about that request, including

    an ``abstractprotocolserver`` instance suitable for handling the

    request. Otherwise, ``None`` is returned.

    ``wsgireq`` is a ``wsgirequest`` instance.

    """

    repo = rctx.repo

    # HTTP version 1 wire protocol requests are denoted by a "cmd" query

    # string parameter. If it isn't present, this isn't a wire protocol

    # request.

        return None

    # The "cmd" request parameter is used by both the wire protocol and hgweb.

    # While not all wire protocol commands are available for all transports,

    # if we see a "cmd" value that resembles a known wire protocol command, we

    # route it to a protocol handler. This is better than routing possible

    # wire protocol requests to hgweb because it prevents hgweb from using

    # known wire protocol commands and it is less confusing for machine

    # clients.

    if not iscmd(cmd):

        return None

    proto = httpv1protocolhandler(wsgireq, repo.ui,

                                  lambda perm: checkperm(rctx, wsgireq, perm))

    return {

        'cmd': cmd,

        'proto': proto,

        'dispatch': lambda: _callhttp(repo, wsgireq, proto, cmd),

        'handleerror': lambda ex: _handlehttperror(ex, wsgireq, cmd),

    }

def _httpresponsetype(ui, wsgireq, prefer_uncompressed):

    """Determine the appropriate response type and compression settings.

    Returns a tuple of (mediatype, compengine, engineopts).

    """

    # Determine the response media type and compression engine based

    # on the request parameters.

    protocaps = decodevaluefromheaders(wsgireq, r'X-HgProto').split(' ')

    if '0.2' in protocaps:

        # All clients are expected to support uncompressed data.

        if prefer_uncompressed:

            return HGTYPE2, util._noopengine(), {}

        # Default as defined by wire protocol spec.

        compformats = ['zlib', 'none']

        for cap in protocaps:

            if cap.startswith('comp='):

                compformats = cap[5:].split(',')

                break

        # Now find an agreed upon compression format.

        for engine in wireproto.supportedcompengines(ui, util.SERVERROLE):

            if engine.wireprotosupport().name in compformats:

                opts = {}

                level = ui.configint('server', '%slevel' % engine.name())

                if level is not None:

                    opts['level'] = level

                return HGTYPE2, engine, opts

        # No mutually supported compression format. Fall back to the

        # legacy protocol.

    # Don't allow untrusted settings because disabling compression or

    # setting a very high compression level could lead to flooding

    # the server's network or CPU.

    opts = {'level': ui.configint('server', 'zliblevel')}

    return HGTYPE, util.compengines['zlib'], opts

def _callhttp(repo, wsgireq, proto, cmd):

    def genversion2(gen, engine, engineopts):

        # application/mercurial-0.2 always sends a payload header

        # identifying the compression engine.

        name = engine.wireprotosupport().name

        assert 0 < len(name) < 256

        yield struct.pack('B', len(name))

        yield name

        for chunk in gen:

            yield chunk

    if not wireproto.commands.commandavailable(cmd, proto):

        wsgireq.respond(HTTP_OK, HGERRTYPE,

                        body=_('requested wire protocol command is not '

                               'available over HTTP'))

        return []

    proto.checkperm(wireproto.commands[cmd].permission)

    rsp = wireproto.dispatch(repo, proto, cmd)

    if isinstance(rsp, bytes):

        wsgireq.respond(HTTP_OK, HGTYPE, body=rsp)

        return []

    elif isinstance(rsp, wireprototypes.bytesresponse):

        wsgireq.respond(HTTP_OK, HGTYPE, body=rsp.data)

        return []

    elif isinstance(rsp, wireprototypes.streamreslegacy):

        gen = rsp.gen

        wsgireq.respond(HTTP_OK, HGTYPE)

        return gen

    elif isinstance(rsp, wireprototypes.streamres):

        gen = rsp.gen

        # This code for compression should not be streamres specific. It

        # is here because we only compress streamres at the moment.

        mediatype, engine, engineopts = _httpresponsetype(

            repo.ui, wsgireq, rsp.prefer_uncompressed)

        gen = engine.compressstream(gen, engineopts)

        if mediatype == HGTYPE2:

            gen = genversion2(gen, engine, engineopts)

        wsgireq.respond(HTTP_OK, mediatype)

        return gen

    elif isinstance(rsp, wireprototypes.pushres):

        rsp = '%d\n%s' % (rsp.res, rsp.output)

        wsgireq.respond(HTTP_OK, HGTYPE, body=rsp)

        return []

    elif isinstance(rsp, wireprototypes.pusherr):

        # This is the httplib workaround documented in _handlehttperror().

        wsgireq.drain()

        rsp = '0\n%s\n' % rsp.res

        wsgireq.respond(HTTP_OK, HGTYPE, body=rsp)

        return []

    elif isinstance(rsp, wireprototypes.ooberror):

        rsp = rsp.message

        wsgireq.respond(HTTP_OK, HGERRTYPE, body=rsp)

        return []

    raise error.ProgrammingError('hgweb.protocol internal failure', rsp)

def _handlehttperror(e, wsgireq, cmd):

    """Called when an ErrorResponse is raised during HTTP request processing."""

    # Clients using Python's httplib are stateful: the HTTP client

    # won't process an HTTP response until all request data is

    # sent to the server. The intent of this code is to ensure

    # we always read HTTP request data from the client, thus

    # ensuring httplib transitions to a state that allows it to read

    # the HTTP response. In other words, it helps prevent deadlocks

    # on clients using httplib.

    if (wsgireq.env[r'REQUEST_METHOD'] == r'POST' and

        # But not if Expect: 100-continue is being used.

        (wsgireq.env.get('HTTP_EXPECT',

                         '').lower() != '100-continue') or

        # Or the non-httplib HTTP library is being advertised by

        # the client.

        wsgireq.env.get('X-HgHttp2', '')):

        wsgireq.drain()

    else:

        wsgireq.headers.append((r'Connection', r'Close'))

    # TODO This response body assumes the failed command was

    # "unbundle." That assumption is not always valid.

    wsgireq.respond(e, HGTYPE, body='0\n%s\n' % pycompat.bytestr(e))

    return ''

def _sshv1respondbytes(fout, value):

    """Send a bytes response for protocol version 1."""

    fout.write('%d\n' % len(value))

    fout.write(value)

    fout.flush()

def _sshv1respondstream(fout, source):

    write = fout.write

    for chunk in source.gen:

        write(chunk)

    fout.flush()

def _sshv1respondooberror(fout, ferr, rsp):

    ferr.write(b'%s\n-\n' % rsp)

    ferr.flush()

    fout.write(b'\n')

    fout.flush()

class sshv1protocolhandler(wireprototypes.baseprotocolhandler):

    """Handler for requests services via version 1 of SSH protocol."""

    def __init__(self, ui, fin, fout):

        self._ui = ui

        self._fin = fin

        self._fout = fout

    @property

    def name(self):

        return wireprototypes.SSHV1

    def getargs(self, args):

        data = {}

        keys = args.split()

        for n in xrange(len(keys)):

            argline = self._fin.readline()[:-1]

            arg, l = argline.split()

            if arg not in keys:

                raise error.Abort(_("unexpected parameter %r") % arg)

            if arg == '*':

                star = {}

                for k in xrange(int(l)):

                    argline = self._fin.readline()[:-1]

                    arg, l = argline.split()

                    val = self._fin.read(int(l))

                    star[arg] = val

                data['*'] = star

            else:

                val = self._fin.read(int(l))

                data[arg] = val

        return [data[k] for k in keys]

    def forwardpayload(self, fpout):

        # We initially send an empty response. This tells the client it is

        # OK to start sending data. If a client sees any other response, it

        # interprets it as an error.

        _sshv1respondbytes(self._fout, b'')

        # The file is in the form:

        #

        # <chunk size>\n<chunk>

        # ...

        # 0\n

        count = int(self._fin.readline())

        while count:

            fpout.write(self._fin.read(count))

            count = int(self._fin.readline())

    @contextlib.contextmanager

    def mayberedirectstdio(self):

        yield None

    def client(self):

        client = encoding.environ.get('SSH_CLIENT', '').split(' ', 1)[0]

        return 'remote:ssh:' + client

    def addcapabilities(self, repo, caps):

        return caps

    def checkperm(self, perm):

        pass

class sshv2protocolhandler(sshv1protocolhandler):

    """Protocol handler for version 2 of the SSH protocol."""

    @property

    def name(self):

        return wireprototypes.SSHV2

def _runsshserver(ui, repo, fin, fout, ev):

    # This function operates like a state machine of sorts. The following

    # states are defined:

    #

    # protov1-serving

    #    Server is in protocol version 1 serving mode. Commands arrive on

    #    new lines. These commands are processed in this state, one command

    #    after the other.

    #

    # protov2-serving

    #    Server is in protocol version 2 serving mode.

    #

    # upgrade-initial

    #    The server is going to process an upgrade request.

    #

    # upgrade-v2-filter-legacy-handshake

    #    The protocol is being upgraded to version 2. The server is expecting

    #    the legacy handshake from version 1.

    #

    # upgrade-v2-finish

    #    The upgrade to version 2 of the protocol is imminent.

    #

    # shutdown

    #    The server is shutting down, possibly in reaction to a client event.

    #

    # And here are their transitions:

    #

    # protov1-serving -> shutdown

    #    When server receives an empty request or encounters another

    #    error.

    #

    # protov1-serving -> upgrade-initial

    #    An upgrade request line was seen.

    #

    # upgrade-initial -> upgrade-v2-filter-legacy-handshake

    #    Upgrade to version 2 in progress. Server is expecting to

    #    process a legacy handshake.

    #

    # upgrade-v2-filter-legacy-handshake -> shutdown

    #    Client did not fulfill upgrade handshake requirements.

    #

    # upgrade-v2-filter-legacy-handshake -> upgrade-v2-finish

    #    Client fulfilled version 2 upgrade requirements. Finishing that

    #    upgrade.

    #

    # upgrade-v2-finish -> protov2-serving

    #    Protocol upgrade to version 2 complete. Server can now speak protocol

    #    version 2.

    #

    # protov2-serving -> protov1-serving

    #    Ths happens by default since protocol version 2 is the same as

    #    version 1 except for the handshake.

    state = 'protov1-serving'

    proto = sshv1protocolhandler(ui, fin, fout)

    protoswitched = False

    while not ev.is_set():

        if state == 'protov1-serving':

            # Commands are issued on new lines.

            request = fin.readline()[:-1]

            # Empty lines signal to terminate the connection.

            if not request:

                state = 'shutdown'

                continue

            # It looks like a protocol upgrade request. Transition state to

            # handle it.

            if request.startswith(b'upgrade '):

                if protoswitched: