sshpeer.py
726 lines
| 24.0 KiB
| text/x-python
|
PythonLexer
/ mercurial / sshpeer.py
|
r17192 | # sshpeer.py - ssh repository proxy class for mercurial | ||
# | ||||
|
r47575 | # Copyright 2005, 2006 Olivia Mackall <olivia@selenic.com> | ||
|
r17192 | # | ||
# This software may be used and distributed according to the terms of the | ||||
# GNU General Public License version 2 or any later version. | ||||
|
r25975 | from __future__ import absolute_import | ||
|
r17192 | import re | ||
|
r35994 | import uuid | ||
|
r25975 | |||
from .i18n import _ | ||||
|
r43359 | from .pycompat import getattr | ||
|
r25975 | from . import ( | ||
error, | ||||
|
r33100 | pycompat, | ||
|
r25975 | util, | ||
|
r35994 | wireprotoserver, | ||
|
r36553 | wireprototypes, | ||
|
r37632 | wireprotov1peer, | ||
|
r37803 | wireprotov1server, | ||
|
r25975 | ) | ||
|
r37138 | from .utils import ( | ||
procutil, | ||||
|
r38494 | stringutil, | ||
r47669 | urlutil, | |||
|
r37138 | ) | ||
|
r17192 | |||
|
r43346 | |||
|
r17192 | def _serverquote(s): | ||
|
r35475 | """quote a string for the remote shell ... which we assume is sh""" | ||
|
r23671 | if not s: | ||
return s | ||||
|
r43347 | if re.match(b'[a-zA-Z0-9@%_+=:,./-]*$', s): | ||
|
r17192 | return s | ||
|
r43347 | return b"'%s'" % s.replace(b"'", b"'\\''") | ||
|
r17192 | |||
|
r43346 | |||
|
r45387 | def _forwardoutput(ui, pipe, warn=False): | ||
|
r25244 | """display all data currently available on pipe as remote output. | ||
This is non blocking.""" | ||||
|
r47429 | if pipe and not pipe.closed: | ||
|
r37138 | s = procutil.readpipe(pipe) | ||
|
r36851 | if s: | ||
|
r45387 | display = ui.warn if warn else ui.status | ||
|
r36851 | for l in s.splitlines(): | ||
|
r45387 | display(_(b"remote: "), l, b'\n') | ||
|
r25244 | |||
|
r43346 | |||
|
r25421 | class doublepipe(object): | ||
"""Operate a side-channel pipe in addition of a main one | ||||
The side-channel pipe contains server output to be forwarded to the user | ||||
input. The double pipe will behave as the "main" pipe, but will ensure the | ||||
content of the "side" pipe is properly processed while we wait for blocking | ||||
call on the "main" pipe. | ||||
If large amounts of data are read from "main", the forward will cease after | ||||
the first bytes start to appear. This simplifies the implementation | ||||
without affecting actual output of sshpeer too much as we rarely issue | ||||
large read for data not yet emitted by the server. | ||||
The main pipe is expected to be a 'bufferedinputpipe' from the util module | ||||
|
r31953 | that handle all the os specific bits. This class lives in this module | ||
|
r26781 | because it focus on behavior specific to the ssh protocol.""" | ||
|
r25421 | |||
def __init__(self, ui, main, side): | ||||
self._ui = ui | ||||
self._main = main | ||||
self._side = side | ||||
def _wait(self): | ||||
"""wait until some data are available on main or side | ||||
return a pair of boolean (ismainready, issideready) | ||||
(This will only wait for data if the setup is supported by `util.poll`) | ||||
""" | ||||
|
r43346 | if ( | ||
isinstance(self._main, util.bufferedinputpipe) | ||||
and self._main.hasbuffer | ||||
): | ||||
|
r36387 | # Main has data. Assume side is worth poking at. | ||
return True, True | ||||
|
r25421 | fds = [self._main.fileno(), self._side.fileno()] | ||
try: | ||||
act = util.poll(fds) | ||||
except NotImplementedError: | ||||
# non supported yet case, assume all have data. | ||||
act = fds | ||||
return (self._main.fileno() in act, self._side.fileno() in act) | ||||
|
r25456 | def write(self, data): | ||
|
r43347 | return self._call(b'write', data) | ||
|
r25456 | |||
|
r25421 | def read(self, size): | ||
|
r43347 | r = self._call(b'read', size) | ||
|
r32062 | if size != 0 and not r: | ||
# We've observed a condition that indicates the | ||||
# stdout closed unexpectedly. Check stderr one | ||||
# more time and snag anything that's there before | ||||
# letting anyone know the main part of the pipe | ||||
# closed prematurely. | ||||
_forwardoutput(self._ui, self._side) | ||||
return r | ||||
|
r25421 | |||
|
r38735 | def unbufferedread(self, size): | ||
|
r43347 | r = self._call(b'unbufferedread', size) | ||
|
r38735 | if size != 0 and not r: | ||
# We've observed a condition that indicates the | ||||
# stdout closed unexpectedly. Check stderr one | ||||
# more time and snag anything that's there before | ||||
# letting anyone know the main part of the pipe | ||||
# closed prematurely. | ||||
_forwardoutput(self._ui, self._side) | ||||
return r | ||||
|
r25421 | def readline(self): | ||
|
r43347 | return self._call(b'readline') | ||
|
r25421 | |||
|
r25455 | def _call(self, methname, data=None): | ||
|
r46554 | """call <methname> on "main", forward output of "side" while blocking""" | ||
|
r25455 | # data can be '' or 0 | ||
if (data is not None and not data) or self._main.closed: | ||||
|
r25421 | _forwardoutput(self._ui, self._side) | ||
|
r43347 | return b'' | ||
|
r25421 | while True: | ||
mainready, sideready = self._wait() | ||||
if sideready: | ||||
_forwardoutput(self._ui, self._side) | ||||
if mainready: | ||||
meth = getattr(self._main, methname) | ||||
|
r25455 | if data is None: | ||
|
r25421 | return meth() | ||
else: | ||||
|
r25455 | return meth(data) | ||
|
r25421 | |||
def close(self): | ||||
return self._main.close() | ||||
|
r47416 | @property | ||
def closed(self): | ||||
return self._main.closed | ||||
|
r25456 | def flush(self): | ||
return self._main.flush() | ||||
|
r43346 | |||
|
r47418 | def _cleanuppipes(ui, pipei, pipeo, pipee, warn): | ||
|
r35951 | """Clean up pipes used by an SSH connection.""" | ||
|
r47418 | didsomething = False | ||
if pipeo and not pipeo.closed: | ||||
didsomething = True | ||||
|
r35951 | pipeo.close() | ||
|
r47418 | if pipei and not pipei.closed: | ||
didsomething = True | ||||
|
r35951 | pipei.close() | ||
|
r47418 | if pipee and not pipee.closed: | ||
didsomething = True | ||||
|
r35951 | # Try to read from the err descriptor until EOF. | ||
try: | ||||
for l in pipee: | ||||
|
r43347 | ui.status(_(b'remote: '), l) | ||
|
r35951 | except (IOError, ValueError): | ||
pass | ||||
pipee.close() | ||||
|
r47418 | if didsomething and warn is not None: | ||
# Encourage explicit close of sshpeers. Closing via __del__ is | ||||
# not very predictable when exceptions are thrown, which has led | ||||
# to deadlocks due to a peer get gc'ed in a fork | ||||
# We add our own stack trace, because the stacktrace when called | ||||
# from __del__ is useless. | ||||
|
r47419 | ui.develwarn(b'missing close on SSH connection created at:\n%s' % warn) | ||
|
r47418 | |||
|
r43346 | |||
|
r35953 | def _makeconnection(ui, sshcmd, args, remotecmd, path, sshenv=None): | ||
"""Create an SSH connection to a server. | ||||
Returns a tuple of (process, stdin, stdout, stderr) for the | ||||
spawned process. | ||||
""" | ||||
|
r43347 | cmd = b'%s %s %s' % ( | ||
|
r35953 | sshcmd, | ||
args, | ||||
|
r43346 | procutil.shellquote( | ||
|
r43347 | b'%s -R %s serve --stdio' | ||
|
r43346 | % (_serverquote(remotecmd), _serverquote(path)) | ||
), | ||||
) | ||||
|
r35953 | |||
|
r43347 | ui.debug(b'running %s\n' % cmd) | ||
|
r35953 | |||
# no buffer allow the use of 'select' | ||||
# feel free to remove buffering and select usage when we ultimately | ||||
# move to threading. | ||||
|
r37138 | stdin, stdout, stderr, proc = procutil.popen4(cmd, bufsize=0, env=sshenv) | ||
|
r35953 | |||
return proc, stdin, stdout, stderr | ||||
|
r43346 | |||
|
r37411 | def _clientcapabilities(): | ||
"""Return list of capabilities of this client. | ||||
Returns a list of capabilities that are supported by this client. | ||||
""" | ||||
|
r43347 | protoparams = {b'partial-pull'} | ||
|
r43346 | comps = [ | ||
e.wireprotosupport().name | ||||
for e in util.compengines.supportedwireengines(util.CLIENTROLE) | ||||
] | ||||
|
r43347 | protoparams.add(b'comp=%s' % b','.join(comps)) | ||
|
r37411 | return protoparams | ||
|
r43346 | |||
|
r35956 | def _performhandshake(ui, stdin, stdout, stderr): | ||
def badresponse(): | ||||
|
r45387 | # Flush any output on stderr. In general, the stderr contains errors | ||
# from the remote (ssh errors, some hg errors), and status indications | ||||
# (like "adding changes"), with no current way to tell them apart. | ||||
# Here we failed so early that it's almost certainly only errors, so | ||||
# use warn=True so -q doesn't hide them. | ||||
_forwardoutput(ui, stderr, warn=True) | ||||
|
r36388 | |||
|
r43347 | msg = _(b'no suitable response from remote hg') | ||
hint = ui.config(b'ui', b'ssherrorhint') | ||||
|
r35956 | raise error.RepoError(msg, hint=hint) | ||
|
r35994 | # The handshake consists of sending wire protocol commands in reverse | ||
# order of protocol implementation and then sniffing for a response | ||||
# to one of them. | ||||
# | ||||
# Those commands (from oldest to newest) are: | ||||
|
r35957 | # | ||
|
r35994 | # ``between`` | ||
# Asks for the set of revisions between a pair of revisions. Command | ||||
# present in all Mercurial server implementations. | ||||
|
r35957 | # | ||
|
r35994 | # ``hello`` | ||
# Instructs the server to advertise its capabilities. Introduced in | ||||
# Mercurial 0.9.1. | ||||
# | ||||
# ``upgrade`` | ||||
# Requests upgrade from default transport protocol version 1 to | ||||
# a newer version. Introduced in Mercurial 4.6 as an experimental | ||||
# feature. | ||||
|
r35957 | # | ||
# The ``between`` command is issued with a request for the null | ||||
# range. If the remote is a Mercurial server, this request will | ||||
# generate a specific response: ``1\n\n``. This represents the | ||||
# wire protocol encoded value for ``\n``. We look for ``1\n\n`` | ||||
# in the output stream and know this is the response to ``between`` | ||||
# and we're at the end of our handshake reply. | ||||
# | ||||
# The response to the ``hello`` command will be a line with the | ||||
# length of the value returned by that command followed by that | ||||
# value. If the server doesn't support ``hello`` (which should be | ||||
# rare), that line will be ``0\n``. Otherwise, the value will contain | ||||
# RFC 822 like lines. Of these, the ``capabilities:`` line contains | ||||
# the capabilities of the server. | ||||
# | ||||
|
r35994 | # The ``upgrade`` command isn't really a command in the traditional | ||
# sense of version 1 of the transport because it isn't using the | ||||
# proper mechanism for formatting insteads: instead, it just encodes | ||||
# arguments on the line, delimited by spaces. | ||||
# | ||||
# The ``upgrade`` line looks like ``upgrade <token> <capabilities>``. | ||||
# If the server doesn't support protocol upgrades, it will reply to | ||||
# this line with ``0\n``. Otherwise, it emits an | ||||
# ``upgraded <token> <protocol>`` line to both stdout and stderr. | ||||
# Content immediately following this line describes additional | ||||
# protocol and server state. | ||||
# | ||||
|
r35957 | # In addition to the responses to our command requests, the server | ||
# may emit "banner" output on stdout. SSH servers are allowed to | ||||
# print messages to stdout on login. Issuing commands on connection | ||||
# allows us to flush this banner output from the server by scanning | ||||
# for output to our well-known ``between`` command. Of course, if | ||||
# the banner contains ``1\n\n``, this will throw off our detection. | ||||
|
r43347 | requestlog = ui.configbool(b'devel', b'debug.peer-request') | ||
|
r35956 | |||
|
r35994 | # Generate a random token to help identify responses to version 2 | ||
# upgrade request. | ||||
|
r36060 | token = pycompat.sysbytes(str(uuid.uuid4())) | ||
|
r35994 | upgradecaps = [ | ||
|
r43347 | (b'proto', wireprotoserver.SSHV2), | ||
|
r35994 | ] | ||
upgradecaps = util.urlreq.urlencode(upgradecaps) | ||||
|
r35956 | try: | ||
|
r43347 | pairsarg = b'%s-%s' % (b'0' * 40, b'0' * 40) | ||
|
r35956 | handshake = [ | ||
|
r43347 | b'hello\n', | ||
b'between\n', | ||||
b'pairs %d\n' % len(pairsarg), | ||||
|
r35956 | pairsarg, | ||
] | ||||
|
r35994 | # Request upgrade to version 2 if configured. | ||
|
r43347 | if ui.configbool(b'experimental', b'sshpeer.advertise-v2'): | ||
ui.debug(b'sending upgrade request: %s %s\n' % (token, upgradecaps)) | ||||
handshake.insert(0, b'upgrade %s %s\n' % (token, upgradecaps)) | ||||
|
r35994 | |||
|
r35956 | if requestlog: | ||
|
r43347 | ui.debug(b'devel-peer-request: hello+between\n') | ||
ui.debug(b'devel-peer-request: pairs: %d bytes\n' % len(pairsarg)) | ||||
ui.debug(b'sending hello command\n') | ||||
ui.debug(b'sending between command\n') | ||||
|
r35956 | |||
|
r43347 | stdin.write(b''.join(handshake)) | ||
|
r35956 | stdin.flush() | ||
except IOError: | ||||
badresponse() | ||||
|
r35994 | # Assume version 1 of wire protocol by default. | ||
|
r36553 | protoname = wireprototypes.SSHV1 | ||
|
r38494 | reupgraded = re.compile(b'^upgraded %s (.*)$' % stringutil.reescape(token)) | ||
|
r35994 | |||
|
r43347 | lines = [b'', b'dummy'] | ||
|
r35956 | max_noise = 500 | ||
while lines[-1] and max_noise: | ||||
try: | ||||
l = stdout.readline() | ||||
|
r45387 | _forwardoutput(ui, stderr, warn=True) | ||
|
r35994 | |||
# Look for reply to protocol upgrade request. It has a token | ||||
# in it, so there should be no false positives. | ||||
m = reupgraded.match(l) | ||||
if m: | ||||
protoname = m.group(1) | ||||
|
r43347 | ui.debug(b'protocol upgraded to %s\n' % protoname) | ||
|
r35994 | # If an upgrade was handled, the ``hello`` and ``between`` | ||
# requests are ignored. The next output belongs to the | ||||
# protocol, so stop scanning lines. | ||||
break | ||||
# Otherwise it could be a banner, ``0\n`` response if server | ||||
# doesn't support upgrade. | ||||
|
r43347 | if lines[-1] == b'1\n' and l == b'\n': | ||
|
r35956 | break | ||
if l: | ||||
|
r43347 | ui.debug(b'remote: ', l) | ||
|
r35956 | lines.append(l) | ||
max_noise -= 1 | ||||
except IOError: | ||||
badresponse() | ||||
else: | ||||
badresponse() | ||||
caps = set() | ||||
|
r35994 | # For version 1, we should see a ``capabilities`` line in response to the | ||
# ``hello`` command. | ||||
|
r36553 | if protoname == wireprototypes.SSHV1: | ||
|
r35994 | for l in reversed(lines): | ||
# Look for response to ``hello`` command. Scan from the back so | ||||
# we don't misinterpret banner output as the command reply. | ||||
|
r43347 | if l.startswith(b'capabilities:'): | ||
caps.update(l[:-1].split(b':')[1].split()) | ||||
|
r35994 | break | ||
elif protoname == wireprotoserver.SSHV2: | ||||
# We see a line with number of bytes to follow and then a value | ||||
# looking like ``capabilities: *``. | ||||
line = stdout.readline() | ||||
try: | ||||
valuelen = int(line) | ||||
except ValueError: | ||||
badresponse() | ||||
capsline = stdout.read(valuelen) | ||||
|
r43347 | if not capsline.startswith(b'capabilities: '): | ||
|
r35994 | badresponse() | ||
|
r43347 | ui.debug(b'remote: %s\n' % capsline) | ||
|
r36234 | |||
|
r43347 | caps.update(capsline.split(b':')[1].split()) | ||
|
r35994 | # Trailing newline. | ||
stdout.read(1) | ||||
# Error if we couldn't find capabilities, this means: | ||||
|
r35958 | # | ||
# 1. Remote isn't a Mercurial server | ||||
# 2. Remote is a <0.9.1 Mercurial server | ||||
# 3. Remote is a future Mercurial server that dropped ``hello`` | ||||
|
r35994 | # and other attempted handshake mechanisms. | ||
|
r35958 | if not caps: | ||
badresponse() | ||||
|
r36388 | # Flush any output on stderr before proceeding. | ||
|
r45387 | _forwardoutput(ui, stderr, warn=True) | ||
|
r36388 | |||
|
r35996 | return protoname, caps | ||
|
r35956 | |||
|
r43346 | |||
|
r37632 | class sshv1peer(wireprotov1peer.wirepeer): | ||
|
r43346 | def __init__( | ||
self, ui, url, proc, stdin, stdout, stderr, caps, autoreadstderr=True | ||||
): | ||||
|
r35954 | """Create a peer from an existing SSH connection. | ||
``proc`` is a handle on the underlying SSH process. | ||||
``stdin``, ``stdout``, and ``stderr`` are handles on the stdio | ||||
pipes for that process. | ||||
|
r35956 | ``caps`` is a set of capabilities supported by the remote. | ||
|
r36550 | ``autoreadstderr`` denotes whether to automatically read from | ||
stderr and to forward its output. | ||||
|
r35954 | """ | ||
self._url = url | ||||
|
r37337 | self.ui = ui | ||
|
r35953 | # self._subprocess is unused. Keeping a handle on the process | ||
# holds a reference and prevents it from being garbage collected. | ||||
|
r35954 | self._subprocess = proc | ||
|
r36388 | |||
# And we hook up our "doublepipe" wrapper to allow querying | ||||
# stderr any time we perform I/O. | ||||
|
r36550 | if autoreadstderr: | ||
stdout = doublepipe(ui, util.bufferedinputpipe(stdout), stderr) | ||||
stdin = doublepipe(ui, stdin, stderr) | ||||
|
r36388 | |||
|
r35954 | self._pipeo = stdin | ||
self._pipei = stdout | ||||
self._pipee = stderr | ||||
|
r35956 | self._caps = caps | ||
|
r36626 | self._autoreadstderr = autoreadstderr | ||
|
r47418 | self._initstack = b''.join(util.getstackframes(1)) | ||
|
r17192 | |||
|
r36385 | # Commands that have a "framed" response where the first line of the | ||
# response contains the length of that response. | ||||
_FRAMED_COMMANDS = { | ||||
|
r43347 | b'batch', | ||
|
r36385 | } | ||
|
r37336 | # Begin of ipeerconnection interface. | ||
|
r33803 | |||
|
r17192 | def url(self): | ||
return self._url | ||||
|
r33803 | def local(self): | ||
return None | ||||
def peer(self): | ||||
return self | ||||
def canpush(self): | ||||
return True | ||||
def close(self): | ||||
|
r47415 | self._cleanup() | ||
|
r33803 | |||
|
r37336 | # End of ipeerconnection interface. | ||
|
r33803 | |||
|
r37336 | # Begin of ipeercommands interface. | ||
|
r33803 | |||
def capabilities(self): | ||||
return self._caps | ||||
|
r37336 | # End of ipeercommands interface. | ||
|
r33803 | |||
|
r33763 | def _readerr(self): | ||
_forwardoutput(self.ui, self._pipee) | ||||
|
r17192 | |||
def _abort(self, exception): | ||||
|
r33763 | self._cleanup() | ||
|
r17192 | raise exception | ||
|
r47418 | def _cleanup(self, warn=None): | ||
_cleanuppipes(self.ui, self._pipei, self._pipeo, self._pipee, warn=warn) | ||||
|
r17192 | |||
|
r47418 | def __del__(self): | ||
self._cleanup(warn=self._initstack) | ||||
|
r17192 | |||
|
r36385 | def _sendrequest(self, cmd, args, framed=False): | ||
|
r43346 | if self.ui.debugflag and self.ui.configbool( | ||
|
r43347 | b'devel', b'debug.peer-request' | ||
|
r43346 | ): | ||
|
r35717 | dbg = self.ui.debug | ||
|
r43347 | line = b'devel-peer-request: %s\n' | ||
|
r35717 | dbg(line % cmd) | ||
for key, value in sorted(args.items()): | ||||
if not isinstance(value, dict): | ||||
|
r43347 | dbg(line % b' %s: %d bytes' % (key, len(value))) | ||
|
r35717 | else: | ||
for dk, dv in sorted(value.items()): | ||||
|
r43347 | dbg(line % b' %s-%s: %d' % (key, dk, len(dv))) | ||
self.ui.debug(b"sending %s command\n" % cmd) | ||||
self._pipeo.write(b"%s\n" % cmd) | ||||
|
r37803 | _func, names = wireprotov1server.commands[cmd] | ||
|
r17192 | keys = names.split() | ||
wireargs = {} | ||||
for k in keys: | ||||
|
r43347 | if k == b'*': | ||
wireargs[b'*'] = args | ||||
|
r17192 | break | ||
else: | ||||
wireargs[k] = args[k] | ||||
del args[k] | ||||
|
r43376 | for k, v in sorted(pycompat.iteritems(wireargs)): | ||
|
r43347 | self._pipeo.write(b"%s %d\n" % (k, len(v))) | ||
|
r17192 | if isinstance(v, dict): | ||
|
r43376 | for dk, dv in pycompat.iteritems(v): | ||
|
r43347 | self._pipeo.write(b"%s %d\n" % (dk, len(dv))) | ||
|
r33763 | self._pipeo.write(dv) | ||
|
r17192 | else: | ||
|
r33763 | self._pipeo.write(v) | ||
self._pipeo.flush() | ||||
|
r17192 | |||
|
r36385 | # We know exactly how many bytes are in the response. So return a proxy | ||
# around the raw output stream that allows reading exactly this many | ||||
# bytes. Callers then can read() without fear of overrunning the | ||||
# response. | ||||
if framed: | ||||
amount = self._getamount() | ||||
return util.cappedreader(self._pipei, amount) | ||||
|
r33763 | return self._pipei | ||
|
r17192 | |||
|
r36384 | def _callstream(self, cmd, **args): | ||
args = pycompat.byteskwargs(args) | ||||
|
r36385 | return self._sendrequest(cmd, args, framed=cmd in self._FRAMED_COMMANDS) | ||
|
r36384 | |||
|
r20905 | def _callcompressable(self, cmd, **args): | ||
|
r36384 | args = pycompat.byteskwargs(args) | ||
|
r36385 | return self._sendrequest(cmd, args, framed=cmd in self._FRAMED_COMMANDS) | ||
|
r20905 | |||
|
r17192 | def _call(self, cmd, **args): | ||
|
r36384 | args = pycompat.byteskwargs(args) | ||
|
r36385 | return self._sendrequest(cmd, args, framed=True).read() | ||
|
r17192 | |||
def _callpush(self, cmd, fp, **args): | ||||
|
r36390 | # The server responds with an empty frame if the client should | ||
# continue submitting the payload. | ||||
|
r17192 | r = self._call(cmd, **args) | ||
if r: | ||||
|
r43347 | return b'', r | ||
|
r36390 | |||
# The payload consists of frames with content followed by an empty | ||||
# frame. | ||||
|
r43347 | for d in iter(lambda: fp.read(4096), b''): | ||
|
r36383 | self._writeframed(d) | ||
|
r43347 | self._writeframed(b"", flush=True) | ||
|
r36390 | |||
# In case of success, there is an empty frame and a frame containing | ||||
# the integer result (as a string). | ||||
# In case of error, there is a non-empty frame containing the error. | ||||
|
r36383 | r = self._readframed() | ||
|
r17192 | if r: | ||
|
r43347 | return b'', r | ||
return self._readframed(), b'' | ||||
|
r17192 | |||
|
r21073 | def _calltwowaystream(self, cmd, fp, **args): | ||
|
r36390 | # The server responds with an empty frame if the client should | ||
# continue submitting the payload. | ||||
|
r21073 | r = self._call(cmd, **args) | ||
if r: | ||||
# XXX needs to be made better | ||||
|
r43347 | raise error.Abort(_(b'unexpected remote reply: %s') % r) | ||
|
r36390 | |||
# The payload consists of frames with content followed by an empty | ||||
# frame. | ||||
|
r43347 | for d in iter(lambda: fp.read(4096), b''): | ||
|
r36383 | self._writeframed(d) | ||
|
r43347 | self._writeframed(b"", flush=True) | ||
|
r36390 | |||
|
r33763 | return self._pipei | ||
|
r17192 | |||
|
r28438 | def _getamount(self): | ||
|
r33763 | l = self._pipei.readline() | ||
|
r43347 | if l == b'\n': | ||
|
r36626 | if self._autoreadstderr: | ||
self._readerr() | ||||
|
r43347 | msg = _(b'check previous remote output') | ||
|
r25243 | self._abort(error.OutOfBandError(hint=msg)) | ||
|
r36626 | if self._autoreadstderr: | ||
self._readerr() | ||||
|
r17192 | try: | ||
|
r28438 | return int(l) | ||
|
r17192 | except ValueError: | ||
|
r43347 | self._abort(error.ResponseError(_(b"unexpected response:"), l)) | ||
|
r28438 | |||
|
r36383 | def _readframed(self): | ||
|
r36646 | size = self._getamount() | ||
if not size: | ||||
return b'' | ||||
return self._pipei.read(size) | ||||
|
r17192 | |||
|
r36383 | def _writeframed(self, data, flush=False): | ||
|
r43347 | self._pipeo.write(b"%d\n" % len(data)) | ||
|
r17192 | if data: | ||
|
r33763 | self._pipeo.write(data) | ||
|
r17192 | if flush: | ||
|
r33763 | self._pipeo.flush() | ||
|
r36626 | if self._autoreadstderr: | ||
self._readerr() | ||||
|
r17192 | |||
|
r43346 | |||
|
r35996 | class sshv2peer(sshv1peer): | ||
"""A peer that speakers version 2 of the transport protocol.""" | ||||
|
r43346 | |||
|
r35996 | # Currently version 2 is identical to version 1 post handshake. | ||
# And handshake is performed before the peer is instantiated. So | ||||
# we need no custom code. | ||||
|
r43346 | |||
|
r36550 | def makepeer(ui, path, proc, stdin, stdout, stderr, autoreadstderr=True): | ||
|
r36505 | """Make a peer instance from existing pipes. | ||
``path`` and ``proc`` are stored on the eventual peer instance and may | ||||
not be used for anything meaningful. | ||||
``stdin``, ``stdout``, and ``stderr`` are the pipes connected to the | ||||
SSH server's stdio handles. | ||||
This function is factored out to allow creating peers that don't | ||||
actually spawn a new process. It is useful for starting SSH protocol | ||||
servers and clients via non-standard means, which can be useful for | ||||
testing. | ||||
""" | ||||
try: | ||||
protoname, caps = _performhandshake(ui, stdin, stdout, stderr) | ||||
except Exception: | ||||
|
r47418 | _cleanuppipes(ui, stdout, stdin, stderr, warn=None) | ||
|
r36505 | raise | ||
|
r36553 | if protoname == wireprototypes.SSHV1: | ||
|
r43346 | return sshv1peer( | ||
ui, | ||||
path, | ||||
proc, | ||||
stdin, | ||||
stdout, | ||||
stderr, | ||||
caps, | ||||
autoreadstderr=autoreadstderr, | ||||
) | ||||
|
r36553 | elif protoname == wireprototypes.SSHV2: | ||
|
r43346 | return sshv2peer( | ||
ui, | ||||
path, | ||||
proc, | ||||
stdin, | ||||
stdout, | ||||
stderr, | ||||
caps, | ||||
autoreadstderr=autoreadstderr, | ||||
) | ||||
|
r36505 | else: | ||
|
r47418 | _cleanuppipes(ui, stdout, stdin, stderr, warn=None) | ||
|
r43346 | raise error.RepoError( | ||
|
r43347 | _(b'unknown version of SSH protocol: %s') % protoname | ||
|
r43346 | ) | ||
|
r36505 | |||
|
r39585 | def instance(ui, path, create, intents=None, createopts=None): | ||
|
r35949 | """Create an SSH peer. | ||
|
r37632 | The returned object conforms to the ``wireprotov1peer.wirepeer`` interface. | ||
|
r35949 | """ | ||
r47669 | u = urlutil.url(path, parsequery=False, parsefragment=False) | |||
|
r43347 | if u.scheme != b'ssh' or not u.host or u.path is None: | ||
raise error.RepoError(_(b"couldn't parse location %s") % path) | ||||
|
r35949 | |||
r47669 | urlutil.checksafessh(path) | |||
|
r35949 | |||
if u.passwd is not None: | ||||
|
r43347 | raise error.RepoError(_(b'password in URL not supported')) | ||
|
r35949 | |||
|
r43347 | sshcmd = ui.config(b'ui', b'ssh') | ||
remotecmd = ui.config(b'ui', b'remotecmd') | ||||
sshaddenv = dict(ui.configitems(b'sshenv')) | ||||
|
r37138 | sshenv = procutil.shellenviron(sshaddenv) | ||
|
r43347 | remotepath = u.path or b'.' | ||
|
r35950 | |||
|
r37138 | args = procutil.sshargs(sshcmd, u.host, u.user, u.port) | ||
|
r35950 | |||
if create: | ||||
|
r39585 | # We /could/ do this, but only if the remote init command knows how to | ||
# handle them. We don't yet make any assumptions about that. And without | ||||
# querying the remote, there's no way of knowing if the remote even | ||||
# supports said requested feature. | ||||
if createopts: | ||||
|
r43346 | raise error.RepoError( | ||
|
r43347 | _( | ||
b'cannot create remote SSH repositories ' | ||||
b'with extra options' | ||||
) | ||||
|
r43346 | ) | ||
|
r39585 | |||
|
r43347 | cmd = b'%s %s %s' % ( | ||
|
r43346 | sshcmd, | ||
args, | ||||
procutil.shellquote( | ||||
|
r43347 | b'%s init %s' | ||
|
r43346 | % (_serverquote(remotecmd), _serverquote(remotepath)) | ||
), | ||||
) | ||||
|
r43347 | ui.debug(b'running %s\n' % cmd) | ||
res = ui.system(cmd, blockedtag=b'sshpeer', environ=sshenv) | ||||
|
r35950 | if res != 0: | ||
|
r43347 | raise error.RepoError(_(b'could not create remote repo')) | ||
|
r35950 | |||
|
r43346 | proc, stdin, stdout, stderr = _makeconnection( | ||
ui, sshcmd, args, remotecmd, remotepath, sshenv | ||||
) | ||||
|
r35953 | |||
|
r37411 | peer = makepeer(ui, path, proc, stdin, stdout, stderr) | ||
# Finally, if supported by the server, notify it about our own | ||||
# capabilities. | ||||
|
r43347 | if b'protocaps' in peer.capabilities(): | ||
|
r37411 | try: | ||
|
r43346 | peer._call( | ||
|
r43347 | b"protocaps", caps=b' '.join(sorted(_clientcapabilities())) | ||
|
r43346 | ) | ||
|
r37411 | except IOError: | ||
peer._cleanup() | ||||
|
r43347 | raise error.RepoError(_(b'capability exchange failed')) | ||
|
r37411 | |||
return peer | ||||