comparison mercurial/wireprotoserver.py @ 43077:687b865b95ad

formatting: byteify all mercurial/ and hgext/ string literals

Done with

    python3.7 contrib/byteify-strings.py -i $(hg files 'set:mercurial/**.py - mercurial/thirdparty/** + hgext/**.py - hgext/fsmonitor/pywatchman/** - mercurial/__init__.py')
    black -l 80 -t py33 -S $(hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**" - hgext/fsmonitor/pywatchman/**')

# skip-blame mass-reformatting only

Differential Revision: https://phab.mercurial-scm.org/D6972

43076:2372284d9457

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

        if v is None:
            break
        chunks.append(pycompat.bytesurl(v))
        i += 1

    return ''.join(chunks)


@interfaceutil.implementer(wireprototypes.baseprotocolhandler)
class httpv1protocolhandler(object):
    def __init__(self, req, ui, checkperm):

        self._checkperm = checkperm
        self._protocaps = None

    @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):

        return args

    def getprotocaps(self):
        if self._protocaps is None:
            value = decodevaluefromheaders(self._req, b'X-HgProto')
            self._protocaps = set(value.split(' '))
        return self._protocaps

    def getpayload(self):
        # Existing clients *always* send Content-Length.
        length = int(self._req.headers[b'Content-Length'])

        finally:
            self._ui.fout = oldout
            self._ui.ferr = olderr

    def client(self):
        return 'remote:%s:%s:%s' % (
            self._req.urlscheme,
            urlreq.quote(self._req.remotehost or ''),
            urlreq.quote(self._req.remoteuser or ''),
        )

    def addcapabilities(self, repo, caps):
        caps.append(b'batch')

        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 = wireprototypes.supportedcompengines(
            repo.ui, compression.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)

    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.
    if 'cmd' not in req.qsparams:
        return False

    cmd = req.qsparams['cmd']

    # 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

    # repo. e.g. ``/?cmd=foo``, ``/repo?cmd=foo``. URL paths within the repo
    # like ``/blah?cmd=foo`` are not allowed. So don't recognize the request
    # in this case. We send an HTTP 404 for backwards compatibility reasons.
    if req.dispatchpath:
        res.status = hgwebcommon.statusmessage(404)
        res.headers['Content-Type'] = HGTYPE
        # TODO This is not a good response to issue for this request. This
        # is mostly for BC for now.
        res.setbodybytes('0\n%s\n' % b'Not Found')
        return True

    proto = httpv1protocolhandler(
        req, repo.ui, lambda perm: checkperm(rctx, req, perm)
    )

        for k, v in e.headers:
            res.headers[k] = v
        res.status = hgwebcommon.statusmessage(e.code, pycompat.bytestr(e))
        # TODO This response body assumes the failed command was
        # "unbundle." That assumption is not always valid.
        res.setbodybytes('0\n%s\n' % pycompat.bytestr(e))

    return True


def _availableapis(repo):
    apis = set()

    # Registered APIs are made available via config options of the name of
    # the protocol.
    for k, v in API_HANDLERS.items():
        section, option = v['config']
        if repo.ui.configbool(section, option):
            apis.add(k)

    return apis

    repo = rctx.repo

    # This whole URL space is experimental for now. But we want to
    # reserve the URL space. So, 404 all URLs if the feature isn't enabled.
    if not repo.ui.configbool('experimental', 'web.apiserver'):
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(_('Experimental API server endpoint not enabled'))
        return

    # The URL space is /api/<protocol>/*. The structure of URLs under varies
    # by <protocol>.

    if req.dispatchparts == [b'api']:
        res.status = b'200 OK'
        res.headers[b'Content-Type'] = b'text/plain'
        lines = [
            _(
                'APIs can be accessed at /api/<name>, where <name> can be '
                'one of the following:\n'
            )
        ]
        if availableapis:
            lines.extend(sorted(availableapis))
        else:
            lines.append(_('(no available APIs)\n'))
        res.setbodybytes(b'\n'.join(lines))
        return

    proto = req.dispatchparts[1]

    if proto not in API_HANDLERS:
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(
            _('Unknown API: %s\nKnown APIs: %s')
            % (proto, b', '.join(sorted(availableapis)))
        )
        return

    if proto not in availableapis:
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(_('API %s not enabled\n') % proto)
        return

    API_HANDLERS[proto]['handler'](
        rctx, req, res, checkperm, req.dispatchparts[2:]
    )


# Maps API name to metadata so custom API can be registered.

# apidescriptor
#    Callable receiving (req, repo) that is called to obtain an API
#    descriptor for this service. The response must be serializable to CBOR.
API_HANDLERS = {
    wireprotov2server.HTTP_WIREPROTO_V2: {
        'config': ('experimental', 'web.api.http-v2'),
        'handler': wireprotov2server.handlehttpv2request,
        'apidescriptor': wireprotov2server.httpv2apidescriptor,
    },
}


def _httpresponsetype(ui, proto, prefer_uncompressed):

    Returns a tuple of (mediatype, compengine, engineopts).
    """
    # Determine the response media type and compression engine based
    # on the request parameters.

    if '0.2' in proto.getprotocaps():
        # All clients are expected to support uncompressed data.
        if prefer_uncompressed:
            return HGTYPE2, compression._noopengine(), {}

        # Now find an agreed upon compression format.

        for engine in wireprototypes.supportedcompengines(
            ui, compression.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 processcapabilitieshandshake(repo, req, res, proto):
    """Called during a ?cmd=capabilities request.

    If the client is advertising support for a newer protocol, we send
    a CBOR response with information about available services. If no
    advertised services are available, we don't handle the request.
    """
    # Fall back to old behavior unless the API server is enabled.
    if not repo.ui.configbool('experimental', 'web.apiserver'):
        return False

    clientapis = decodevaluefromheaders(req, b'X-HgUpgrade')
    protocaps = decodevaluefromheaders(req, b'X-HgProto')
    if not clientapis or not protocaps:
        return False

    # We currently only support CBOR responses.
    protocaps = set(protocaps.split(' '))
    if b'cbor' not in protocaps:
        return False

    descriptors = {}

    for api in sorted(set(clientapis.split()) & _availableapis(repo)):
        handler = API_HANDLERS[api]

        descriptorfn = handler.get('apidescriptor')
        if not descriptorfn:
            continue

        descriptors[api] = descriptorfn(req, repo)

    v1caps = wireprotov1server.dispatch(repo, proto, 'capabilities')
    assert isinstance(v1caps, wireprototypes.bytesresponse)

    m = {
        # TODO allow this to be configurable.
        'apibase': 'api/',
        'apis': descriptors,
        'v1capabilities': v1caps.data,
    }

    res.status = b'200 OK'
    res.headers[b'Content-Type'] = b'application/mercurial-cbor'
    res.setbodybytes(b''.join(cborutil.streamencode(m)))

    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

    def setresponse(code, contenttype, bodybytes=None, bodygen=None):
        if code == HTTP_OK:
            res.status = '200 Script output follows'
        else:
            res.status = hgwebcommon.statusmessage(code)

        res.headers['Content-Type'] = contenttype

        if bodybytes is not None:
            res.setbodybytes(bodybytes)
        if bodygen is not None:
            res.setbodygen(bodygen)

    if not wireprotov1server.commands.commandavailable(cmd, proto):
        setresponse(
            HTTP_OK,
            HGERRTYPE,
            _('requested wire protocol command is not available over ' 'HTTP'),
        )
        return

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

    # Possibly handle a modern client wanting to switch protocols.
    if cmd == 'capabilities' and processcapabilitieshandshake(
        repo, req, res, proto
    ):

        return

        if mediatype == HGTYPE2:
            gen = genversion2(gen, engine, engineopts)

        setresponse(HTTP_OK, mediatype, bodygen=gen)
    elif isinstance(rsp, wireprototypes.pushres):
        rsp = '%d\n%s' % (rsp.res, rsp.output)
        setresponse(HTTP_OK, HGTYPE, bodybytes=rsp)
    elif isinstance(rsp, wireprototypes.pusherr):
        rsp = '0\n%s\n' % rsp.res
        res.drain = True
        setresponse(HTTP_OK, HGTYPE, bodybytes=rsp)
    elif isinstance(rsp, wireprototypes.ooberror):
        setresponse(HTTP_OK, HGERRTYPE, bodybytes=rsp.message)
    else:
        raise error.ProgrammingError('hgweb.protocol internal failure', rsp)


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

        keys = args.split()
        for n in pycompat.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 pycompat.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]

    @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):
        if self.name == wireprototypes.SSHV1:
            caps.append(b'protocaps')
        caps.append(b'batch')

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

                    _sshv1respondooberror(
                        fout,
                        ui.ferr,
                        b'cannot upgrade protocols multiple ' b'times',
                    )
                    state = 'shutdown'
                    continue

                state = 'upgrade-initial'
                continue

            available = wireprotov1server.commands.commandavailable(
                request, proto
            )

                _sshv1respondbytes(fout, rsp.res)
            elif isinstance(rsp, wireprototypes.ooberror):
                _sshv1respondooberror(fout, ui.ferr, rsp.message)
            else:
                raise error.ProgrammingError(
                    'unhandled response type from '
                    'wire protocol command: %s' % rsp
                )

        # For now, protocol version 2 serving just goes back to version 1.
        elif state == 'protov2-serving':
            state = 'protov1-serving'
            continue

        elif state == 'upgrade-initial':
            # We should never transition into this state if we've switched
            # protocols.
            assert not protoswitched
            assert proto.name == wireprototypes.SSHV1

            # We treat this as an unknown command.
            try:
                token, caps = request.split(b' ')[1:]
            except ValueError:
                _sshv1respondbytes(fout, b'')
                state = 'protov1-serving'
                continue

            # Send empty response if we don't support upgrading protocols.
            if not ui.configbool('experimental', 'sshserver.support-v2'):
                _sshv1respondbytes(fout, b'')
                state = 'protov1-serving'
                continue

            try:
                caps = urlreq.parseqs(caps)
            except ValueError:
                _sshv1respondbytes(fout, b'')
                state = 'protov1-serving'
                continue

            # We don't see an upgrade request to protocol version 2. Ignore
            # the upgrade request.
            wantedprotos = caps.get(b'proto', [b''])[0]
            if SSHV2 not in wantedprotos:
                _sshv1respondbytes(fout, b'')
                state = 'protov1-serving'
                continue

            # It looks like we can honor this upgrade request to protocol 2.
            # Filter the rest of the handshake protocol request lines.
            state = 'upgrade-v2-filter-legacy-handshake'
            continue

        elif state == 'upgrade-v2-filter-legacy-handshake':
            # Client should have sent legacy handshake after an ``upgrade``
            # request. Expected lines:
            #
            #    hello
            #    between

                        fout,
                        ui.ferr,
                        b'malformed handshake protocol: ' b'missing %s' % line,
                    )
                    ok = False
                    state = 'shutdown'
                    break

            if not ok:
                continue

                    fout,
                    ui.ferr,
                    b'malformed handshake protocol: '
                    b'missing between argument value',
                )
                state = 'shutdown'
                continue

            state = 'upgrade-v2-finish'
            continue

        elif state == 'upgrade-v2-finish':
            # Send the upgrade response.
            fout.write(b'upgraded %s %s\n' % (token, SSHV2))
            servercaps = wireprotov1server.capabilities(repo, proto)
            rsp = b'capabilities: %s' % servercaps.data
            fout.write(b'%d\n%s\n' % (len(rsp), rsp))
            fout.flush()

            proto = sshv2protocolhandler(ui, fin, fout)
            protoswitched = True

            state = 'protov2-serving'
            continue

        elif state == 'shutdown':
            break

        else:
            raise error.ProgrammingError(
                'unhandled ssh server state: %s' % state
            )


class sshserver(object):
    def __init__(self, ui, repo, logfh=None):

        self._fin, self._fout = ui.protectfinout()

        # Log write I/O to stdout and stderr if configured.
        if logfh:
            self._fout = util.makeloggingfileobject(
                logfh, self._fout, 'o', logdata=True
            )
            ui.ferr = util.makeloggingfileobject(
                logfh, ui.ferr, 'e', logdata=True
            )

    def serve_forever(self):
        self.serveuntil(threading.Event())
        self._ui.restorefinout(self._fin, self._fout)

43077:687b865b95ad

urlerr = util.urlerr
urlreq = util.urlreq

HTTP_OK = 200

HGTYPE = b'application/mercurial-0.1'
HGTYPE2 = b'application/mercurial-0.2'
HGERRTYPE = b'application/hg-error'

SSHV1 = wireprototypes.SSHV1
SSHV2 = wireprototypes.SSHV2

        if v is None:
            break
        chunks.append(pycompat.bytesurl(v))
        i += 1

    return b''.join(chunks)


@interfaceutil.implementer(wireprototypes.baseprotocolhandler)
class httpv1protocolhandler(object):
    def __init__(self, req, ui, checkperm):

        self._checkperm = checkperm
        self._protocaps = None

    @property
    def name(self):
        return b'http-v1'

    def getargs(self, args):
        knownargs = self._args()
        data = {}
        keys = args.split()
        for k in keys:
            if k == b'*':
                star = {}
                for key in knownargs.keys():
                    if key != b'cmd' and key not in keys:
                        star[key] = knownargs[key][0]
                data[b'*'] = star
            else:
                data[k] = knownargs[k][0]
        return [data[k] for k in keys]

    def _args(self):

        return args

    def getprotocaps(self):
        if self._protocaps is None:
            value = decodevaluefromheaders(self._req, b'X-HgProto')
            self._protocaps = set(value.split(b' '))
        return self._protocaps

    def getpayload(self):
        # Existing clients *always* send Content-Length.
        length = int(self._req.headers[b'Content-Length'])

        finally:
            self._ui.fout = oldout
            self._ui.ferr = olderr

    def client(self):
        return b'remote:%s:%s:%s' % (
            self._req.urlscheme,
            urlreq.quote(self._req.remotehost or b''),
            urlreq.quote(self._req.remoteuser or b''),
        )

    def addcapabilities(self, repo, caps):
        caps.append(b'batch')

        caps.append(
            b'httpheader=%d' % repo.ui.configint(b'server', b'maxhttpheaderlen')
        )
        if repo.ui.configbool(b'experimental', b'httppostargs'):
            caps.append(b'httppostargs')

        # FUTURE advertise 0.2rx once support is implemented
        # FUTURE advertise minrx and mintx after consulting config option
        caps.append(b'httpmediatype=0.1rx,0.1tx,0.2tx')

        compengines = wireprototypes.supportedcompengines(
            repo.ui, compression.SERVERROLE
        )
        if compengines:
            comptypes = b','.join(
                urlreq.quote(e.wireprotosupport().name) for e in compengines
            )
            caps.append(b'compression=%s' % comptypes)

        return caps

    def checkperm(self, perm):
        return self._checkperm(perm)

    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.
    if b'cmd' not in req.qsparams:
        return False

    cmd = req.qsparams[b'cmd']

    # 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

    # repo. e.g. ``/?cmd=foo``, ``/repo?cmd=foo``. URL paths within the repo
    # like ``/blah?cmd=foo`` are not allowed. So don't recognize the request
    # in this case. We send an HTTP 404 for backwards compatibility reasons.
    if req.dispatchpath:
        res.status = hgwebcommon.statusmessage(404)
        res.headers[b'Content-Type'] = HGTYPE
        # TODO This is not a good response to issue for this request. This
        # is mostly for BC for now.
        res.setbodybytes(b'0\n%s\n' % b'Not Found')
        return True

    proto = httpv1protocolhandler(
        req, repo.ui, lambda perm: checkperm(rctx, req, perm)
    )

        for k, v in e.headers:
            res.headers[k] = v
        res.status = hgwebcommon.statusmessage(e.code, pycompat.bytestr(e))
        # TODO This response body assumes the failed command was
        # "unbundle." That assumption is not always valid.
        res.setbodybytes(b'0\n%s\n' % pycompat.bytestr(e))

    return True


def _availableapis(repo):
    apis = set()

    # Registered APIs are made available via config options of the name of
    # the protocol.
    for k, v in API_HANDLERS.items():
        section, option = v[b'config']
        if repo.ui.configbool(section, option):
            apis.add(k)

    return apis

    repo = rctx.repo

    # This whole URL space is experimental for now. But we want to
    # reserve the URL space. So, 404 all URLs if the feature isn't enabled.
    if not repo.ui.configbool(b'experimental', b'web.apiserver'):
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(_(b'Experimental API server endpoint not enabled'))
        return

    # The URL space is /api/<protocol>/*. The structure of URLs under varies
    # by <protocol>.

    if req.dispatchparts == [b'api']:
        res.status = b'200 OK'
        res.headers[b'Content-Type'] = b'text/plain'
        lines = [
            _(
                b'APIs can be accessed at /api/<name>, where <name> can be '
                b'one of the following:\n'
            )
        ]
        if availableapis:
            lines.extend(sorted(availableapis))
        else:
            lines.append(_(b'(no available APIs)\n'))
        res.setbodybytes(b'\n'.join(lines))
        return

    proto = req.dispatchparts[1]

    if proto not in API_HANDLERS:
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(
            _(b'Unknown API: %s\nKnown APIs: %s')
            % (proto, b', '.join(sorted(availableapis)))
        )
        return

    if proto not in availableapis:
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(_(b'API %s not enabled\n') % proto)
        return

    API_HANDLERS[proto][b'handler'](
        rctx, req, res, checkperm, req.dispatchparts[2:]
    )


# Maps API name to metadata so custom API can be registered.

# apidescriptor
#    Callable receiving (req, repo) that is called to obtain an API
#    descriptor for this service. The response must be serializable to CBOR.
API_HANDLERS = {
    wireprotov2server.HTTP_WIREPROTO_V2: {
        b'config': (b'experimental', b'web.api.http-v2'),
        b'handler': wireprotov2server.handlehttpv2request,
        b'apidescriptor': wireprotov2server.httpv2apidescriptor,
    },
}


def _httpresponsetype(ui, proto, prefer_uncompressed):

    Returns a tuple of (mediatype, compengine, engineopts).
    """
    # Determine the response media type and compression engine based
    # on the request parameters.

    if b'0.2' in proto.getprotocaps():
        # All clients are expected to support uncompressed data.
        if prefer_uncompressed:
            return HGTYPE2, compression._noopengine(), {}

        # Now find an agreed upon compression format.

        for engine in wireprototypes.supportedcompengines(
            ui, compression.SERVERROLE
        ):
            if engine.wireprotosupport().name in compformats:
                opts = {}
                level = ui.configint(b'server', b'%slevel' % engine.name())
                if level is not None:
                    opts[b'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 = {b'level': ui.configint(b'server', b'zliblevel')}
    return HGTYPE, util.compengines[b'zlib'], opts


def processcapabilitieshandshake(repo, req, res, proto):
    """Called during a ?cmd=capabilities request.

    If the client is advertising support for a newer protocol, we send
    a CBOR response with information about available services. If no
    advertised services are available, we don't handle the request.
    """
    # Fall back to old behavior unless the API server is enabled.
    if not repo.ui.configbool(b'experimental', b'web.apiserver'):
        return False

    clientapis = decodevaluefromheaders(req, b'X-HgUpgrade')
    protocaps = decodevaluefromheaders(req, b'X-HgProto')
    if not clientapis or not protocaps:
        return False

    # We currently only support CBOR responses.
    protocaps = set(protocaps.split(b' '))
    if b'cbor' not in protocaps:
        return False

    descriptors = {}

    for api in sorted(set(clientapis.split()) & _availableapis(repo)):
        handler = API_HANDLERS[api]

        descriptorfn = handler.get(b'apidescriptor')
        if not descriptorfn:
            continue

        descriptors[api] = descriptorfn(req, repo)

    v1caps = wireprotov1server.dispatch(repo, proto, b'capabilities')
    assert isinstance(v1caps, wireprototypes.bytesresponse)

    m = {
        # TODO allow this to be configurable.
        b'apibase': b'api/',
        b'apis': descriptors,
        b'v1capabilities': v1caps.data,
    }

    res.status = b'200 OK'
    res.headers[b'Content-Type'] = b'application/mercurial-cbor'
    res.setbodybytes(b''.join(cborutil.streamencode(m)))

    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'B', len(name))
        yield name

        for chunk in gen:
            yield chunk

    def setresponse(code, contenttype, bodybytes=None, bodygen=None):
        if code == HTTP_OK:
            res.status = b'200 Script output follows'
        else:
            res.status = hgwebcommon.statusmessage(code)

        res.headers[b'Content-Type'] = contenttype

        if bodybytes is not None:
            res.setbodybytes(bodybytes)
        if bodygen is not None:
            res.setbodygen(bodygen)

    if not wireprotov1server.commands.commandavailable(cmd, proto):
        setresponse(
            HTTP_OK,
            HGERRTYPE,
            _(
                b'requested wire protocol command is not available over '
                b'HTTP'
            ),
        )
        return

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

    # Possibly handle a modern client wanting to switch protocols.
    if cmd == b'capabilities' and processcapabilitieshandshake(
        repo, req, res, proto
    ):

        return

        if mediatype == HGTYPE2:
            gen = genversion2(gen, engine, engineopts)

        setresponse(HTTP_OK, mediatype, bodygen=gen)
    elif isinstance(rsp, wireprototypes.pushres):
        rsp = b'%d\n%s' % (rsp.res, rsp.output)
        setresponse(HTTP_OK, HGTYPE, bodybytes=rsp)
    elif isinstance(rsp, wireprototypes.pusherr):
        rsp = b'0\n%s\n' % rsp.res
        res.drain = True
        setresponse(HTTP_OK, HGTYPE, bodybytes=rsp)
    elif isinstance(rsp, wireprototypes.ooberror):
        setresponse(HTTP_OK, HGERRTYPE, bodybytes=rsp.message)
    else:
        raise error.ProgrammingError(b'hgweb.protocol internal failure', rsp)


def _sshv1respondbytes(fout, value):
    """Send a bytes response for protocol version 1."""
    fout.write(b'%d\n' % len(value))
    fout.write(value)
    fout.flush()


def _sshv1respondstream(fout, source):

        keys = args.split()
        for n in pycompat.xrange(len(keys)):
            argline = self._fin.readline()[:-1]
            arg, l = argline.split()
            if arg not in keys:
                raise error.Abort(_(b"unexpected parameter %r") % arg)
            if arg == b'*':
                star = {}
                for k in pycompat.xrange(int(l)):
                    argline = self._fin.readline()[:-1]
                    arg, l = argline.split()
                    val = self._fin.read(int(l))
                    star[arg] = val
                data[b'*'] = star
            else:
                val = self._fin.read(int(l))
                data[arg] = val
        return [data[k] for k in keys]

    @contextlib.contextmanager
    def mayberedirectstdio(self):
        yield None

    def client(self):
        client = encoding.environ.get(b'SSH_CLIENT', b'').split(b' ', 1)[0]
        return b'remote:ssh:' + client

    def addcapabilities(self, repo, caps):
        if self.name == wireprototypes.SSHV1:
            caps.append(b'protocaps')
        caps.append(b'batch')

    #
    # protov2-serving -> protov1-serving
    #    Ths happens by default since protocol version 2 is the same as
    #    version 1 except for the handshake.

    state = b'protov1-serving'
    proto = sshv1protocolhandler(ui, fin, fout)
    protoswitched = False

    while not ev.is_set():
        if state == b'protov1-serving':
            # Commands are issued on new lines.
            request = fin.readline()[:-1]

            # Empty lines signal to terminate the connection.
            if not request:
                state = b'shutdown'
                continue

            # It looks like a protocol upgrade request. Transition state to
            # handle it.
            if request.startswith(b'upgrade '):

                    _sshv1respondooberror(
                        fout,
                        ui.ferr,
                        b'cannot upgrade protocols multiple ' b'times',
                    )
                    state = b'shutdown'
                    continue

                state = b'upgrade-initial'
                continue

            available = wireprotov1server.commands.commandavailable(
                request, proto
            )

                _sshv1respondbytes(fout, rsp.res)
            elif isinstance(rsp, wireprototypes.ooberror):
                _sshv1respondooberror(fout, ui.ferr, rsp.message)
            else:
                raise error.ProgrammingError(
                    b'unhandled response type from '
                    b'wire protocol command: %s' % rsp
                )

        # For now, protocol version 2 serving just goes back to version 1.
        elif state == b'protov2-serving':
            state = b'protov1-serving'
            continue

        elif state == b'upgrade-initial':
            # We should never transition into this state if we've switched
            # protocols.
            assert not protoswitched
            assert proto.name == wireprototypes.SSHV1

            # We treat this as an unknown command.
            try:
                token, caps = request.split(b' ')[1:]
            except ValueError:
                _sshv1respondbytes(fout, b'')
                state = b'protov1-serving'
                continue

            # Send empty response if we don't support upgrading protocols.
            if not ui.configbool(b'experimental', b'sshserver.support-v2'):
                _sshv1respondbytes(fout, b'')
                state = b'protov1-serving'
                continue

            try:
                caps = urlreq.parseqs(caps)
            except ValueError:
                _sshv1respondbytes(fout, b'')
                state = b'protov1-serving'
                continue

            # We don't see an upgrade request to protocol version 2. Ignore
            # the upgrade request.
            wantedprotos = caps.get(b'proto', [b''])[0]
            if SSHV2 not in wantedprotos:
                _sshv1respondbytes(fout, b'')
                state = b'protov1-serving'
                continue

            # It looks like we can honor this upgrade request to protocol 2.
            # Filter the rest of the handshake protocol request lines.
            state = b'upgrade-v2-filter-legacy-handshake'
            continue

        elif state == b'upgrade-v2-filter-legacy-handshake':
            # Client should have sent legacy handshake after an ``upgrade``
            # request. Expected lines:
            #
            #    hello
            #    between

                        fout,
                        ui.ferr,
                        b'malformed handshake protocol: ' b'missing %s' % line,
                    )
                    ok = False
                    state = b'shutdown'
                    break

            if not ok:
                continue

                    fout,
                    ui.ferr,
                    b'malformed handshake protocol: '
                    b'missing between argument value',
                )
                state = b'shutdown'
                continue

            state = b'upgrade-v2-finish'
            continue

        elif state == b'upgrade-v2-finish':
            # Send the upgrade response.
            fout.write(b'upgraded %s %s\n' % (token, SSHV2))
            servercaps = wireprotov1server.capabilities(repo, proto)
            rsp = b'capabilities: %s' % servercaps.data
            fout.write(b'%d\n%s\n' % (len(rsp), rsp))
            fout.flush()

            proto = sshv2protocolhandler(ui, fin, fout)
            protoswitched = True

            state = b'protov2-serving'
            continue

        elif state == b'shutdown':
            break

        else:
            raise error.ProgrammingError(
                b'unhandled ssh server state: %s' % state
            )


class sshserver(object):
    def __init__(self, ui, repo, logfh=None):

        self._fin, self._fout = ui.protectfinout()

        # Log write I/O to stdout and stderr if configured.
        if logfh:
            self._fout = util.makeloggingfileobject(
                logfh, self._fout, b'o', logdata=True
            )
            ui.ferr = util.makeloggingfileobject(
                logfh, ui.ferr, b'e', logdata=True
            )

    def serve_forever(self):
        self.serveuntil(threading.Event())
        self._ui.restorefinout(self._fin, self._fout)