sshpeer.py
374 lines
| 11.8 KiB
| text/x-python
|
PythonLexer
/ mercurial / sshpeer.py
|
r17192 | # sshpeer.py - ssh repository proxy class for mercurial | ||
# | ||||
# Copyright 2005, 2006 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. | ||||
|
r25975 | from __future__ import absolute_import | ||
|
r17192 | import re | ||
|
r25975 | |||
from .i18n import _ | ||||
from . import ( | ||||
error, | ||||
|
r33100 | pycompat, | ||
|
r25975 | util, | ||
wireproto, | ||||
) | ||||
|
r17192 | |||
def _serverquote(s): | ||||
|
r35475 | """quote a string for the remote shell ... which we assume is sh""" | ||
|
r23671 | if not s: | ||
return s | ||||
|
r17192 | if re.match('[a-zA-Z0-9@%_+=:,./-]*$', s): | ||
return s | ||||
return "'%s'" % s.replace("'", "'\\''") | ||||
|
r25244 | def _forwardoutput(ui, pipe): | ||
"""display all data currently available on pipe as remote output. | ||||
This is non blocking.""" | ||||
s = util.readpipe(pipe) | ||||
if s: | ||||
for l in s.splitlines(): | ||||
ui.status(_("remote: "), l, '\n') | ||||
|
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`) | ||||
""" | ||||
|
r25457 | if getattr(self._main, 'hasbuffer', False): # getattr for classic pipe | ||
|
r25421 | return (True, True) # main has data, assume side is worth poking at. | ||
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): | ||
return self._call('write', data) | ||||
|
r25421 | def read(self, size): | ||
|
r32062 | r = self._call('read', size) | ||
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): | ||||
return self._call('readline') | ||||
|
r25455 | def _call(self, methname, data=None): | ||
|
r25421 | """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) | ||
return '' | ||||
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() | ||||
|
r25456 | def flush(self): | ||
return self._main.flush() | ||||
|
r17192 | class sshpeer(wireproto.wirepeer): | ||
def __init__(self, ui, path, create=False): | ||||
self._url = path | ||||
|
r33803 | self._ui = ui | ||
|
r33763 | self._pipeo = self._pipei = self._pipee = None | ||
|
r17192 | |||
u = util.url(path, parsequery=False, parsefragment=False) | ||||
if u.scheme != 'ssh' or not u.host or u.path is None: | ||||
self._abort(error.RepoError(_("couldn't parse location %s") % path)) | ||||
|
r33725 | util.checksafessh(path) | ||
|
r17192 | if u.passwd is not None: | ||
self._abort(error.RepoError(_("password in URL not supported"))) | ||||
|
r33763 | |||
self._user = u.user | ||||
self._host = u.host | ||||
self._port = u.port | ||||
self._path = u.path or '.' | ||||
|
r17192 | |||
|
r33499 | sshcmd = self.ui.config("ui", "ssh") | ||
remotecmd = self.ui.config("ui", "remotecmd") | ||||
|
r35452 | sshaddenv = dict(self.ui.configitems("sshenv")) | ||
sshenv = util.shellenviron(sshaddenv) | ||||
|
r17192 | |||
|
r33763 | args = util.sshargs(sshcmd, self._host, self._user, self._port) | ||
|
r17192 | |||
if create: | ||||
cmd = '%s %s %s' % (sshcmd, args, | ||||
util.shellquote("%s init %s" % | ||||
|
r33763 | (_serverquote(remotecmd), _serverquote(self._path)))) | ||
|
r20794 | ui.debug('running %s\n' % cmd) | ||
|
r35452 | res = ui.system(cmd, blockedtag='sshpeer', environ=sshenv) | ||
|
r17192 | if res != 0: | ||
self._abort(error.RepoError(_("could not create remote repo"))) | ||||
|
r35452 | self._validaterepo(sshcmd, args, remotecmd, sshenv) | ||
|
r17192 | |||
|
r33803 | # Begin of _basepeer interface. | ||
@util.propertycache | ||||
def ui(self): | ||||
return self._ui | ||||
|
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): | ||||
pass | ||||
# End of _basepeer interface. | ||||
# Begin of _basewirecommands interface. | ||||
def capabilities(self): | ||||
return self._caps | ||||
# End of _basewirecommands interface. | ||||
|
r35452 | def _validaterepo(self, sshcmd, args, remotecmd, sshenv=None): | ||
|
r17192 | # cleanup up previous run | ||
|
r33763 | self._cleanup() | ||
|
r17192 | |||
cmd = '%s %s %s' % (sshcmd, args, | ||||
util.shellquote("%s -R %s serve --stdio" % | ||||
|
r33763 | (_serverquote(remotecmd), _serverquote(self._path)))) | ||
|
r20794 | self.ui.debug('running %s\n' % cmd) | ||
|
r17192 | cmd = util.quotecommand(cmd) | ||
|
r18759 | |||
|
r33763 | # while self._subprocess isn't used, having it allows the subprocess to | ||
|
r18759 | # to clean up correctly later | ||
|
r25408 | # | ||
# no buffer allow the use of 'select' | ||||
# feel free to remove buffering and select usage when we ultimately | ||||
# move to threading. | ||||
|
r35452 | sub = util.popen4(cmd, bufsize=0, env=sshenv) | ||
|
r33763 | self._pipeo, self._pipei, self._pipee, self._subprocess = sub | ||
|
r17192 | |||
|
r33763 | self._pipei = util.bufferedinputpipe(self._pipei) | ||
self._pipei = doublepipe(self.ui, self._pipei, self._pipee) | ||||
self._pipeo = doublepipe(self.ui, self._pipeo, self._pipee) | ||||
|
r25407 | |||
|
r34194 | def badresponse(): | ||
|
r35107 | msg = _("no suitable response from remote hg") | ||
hint = self.ui.config("ui", "ssherrorhint") | ||||
self._abort(error.RepoError(msg, hint=hint)) | ||||
|
r34194 | |||
try: | ||||
# skip any noise generated by remote shell | ||||
self._callstream("hello") | ||||
r = self._callstream("between", pairs=("%s-%s" % ("0"*40, "0"*40))) | ||||
except IOError: | ||||
badresponse() | ||||
|
r17192 | lines = ["", "dummy"] | ||
max_noise = 500 | ||||
while lines[-1] and max_noise: | ||||
|
r34194 | try: | ||
l = r.readline() | ||||
|
r34107 | self._readerr() | ||
|
r34194 | if lines[-1] == "1\n" and l == "\n": | ||
break | ||||
if l: | ||||
self.ui.debug("remote: ", l) | ||||
lines.append(l) | ||||
max_noise -= 1 | ||||
except IOError: | ||||
badresponse() | ||||
|
r17192 | else: | ||
|
r34194 | badresponse() | ||
|
r17192 | |||
self._caps = set() | ||||
for l in reversed(lines): | ||||
if l.startswith("capabilities:"): | ||||
self._caps.update(l[:-1].split(":")[1].split()) | ||||
break | ||||
|
r33763 | def _readerr(self): | ||
_forwardoutput(self.ui, self._pipee) | ||||
|
r17192 | |||
def _abort(self, exception): | ||||
|
r33763 | self._cleanup() | ||
|
r17192 | raise exception | ||
|
r33763 | def _cleanup(self): | ||
if self._pipeo is None: | ||||
|
r17192 | return | ||
|
r33763 | self._pipeo.close() | ||
self._pipei.close() | ||||
|
r17192 | try: | ||
# read the error descriptor until EOF | ||||
|
r33763 | for l in self._pipee: | ||
|
r17192 | self.ui.status(_("remote: "), l) | ||
except (IOError, ValueError): | ||||
pass | ||||
|
r33763 | self._pipee.close() | ||
|
r17192 | |||
|
r33763 | __del__ = _cleanup | ||
|
r17192 | |||
|
r28438 | def _submitbatch(self, req): | ||
|
r29733 | rsp = self._callstream("batch", cmds=wireproto.encodebatchcmds(req)) | ||
|
r28438 | available = self._getamount() | ||
# TODO this response parsing is probably suboptimal for large | ||||
# batches with large responses. | ||||
toread = min(available, 1024) | ||||
work = rsp.read(toread) | ||||
available -= toread | ||||
chunk = work | ||||
while chunk: | ||||
while ';' in work: | ||||
one, work = work.split(';', 1) | ||||
yield wireproto.unescapearg(one) | ||||
toread = min(available, 1024) | ||||
chunk = rsp.read(toread) | ||||
available -= toread | ||||
work += chunk | ||||
yield wireproto.unescapearg(work) | ||||
|
r17192 | def _callstream(self, cmd, **args): | ||
|
r33100 | args = pycompat.byteskwargs(args) | ||
|
r35717 | if (self.ui.debugflag | ||
and self.ui.configbool('devel', 'debug.peer-request')): | ||||
dbg = self.ui.debug | ||||
line = 'devel-peer-request: %s\n' | ||||
dbg(line % cmd) | ||||
for key, value in sorted(args.items()): | ||||
if not isinstance(value, dict): | ||||
dbg(line % ' %s: %d bytes' % (key, len(value))) | ||||
else: | ||||
for dk, dv in sorted(value.items()): | ||||
dbg(line % ' %s-%s: %d' % (key, dk, len(dv))) | ||||
|
r17192 | self.ui.debug("sending %s command\n" % cmd) | ||
|
r33763 | self._pipeo.write("%s\n" % cmd) | ||
|
r17192 | _func, names = wireproto.commands[cmd] | ||
keys = names.split() | ||||
wireargs = {} | ||||
for k in keys: | ||||
if k == '*': | ||||
wireargs['*'] = args | ||||
break | ||||
else: | ||||
wireargs[k] = args[k] | ||||
del args[k] | ||||
for k, v in sorted(wireargs.iteritems()): | ||||
|
r33763 | self._pipeo.write("%s %d\n" % (k, len(v))) | ||
|
r17192 | if isinstance(v, dict): | ||
for dk, dv in v.iteritems(): | ||||
|
r33763 | self._pipeo.write("%s %d\n" % (dk, len(dv))) | ||
self._pipeo.write(dv) | ||||
|
r17192 | else: | ||
|
r33763 | self._pipeo.write(v) | ||
self._pipeo.flush() | ||||
|
r17192 | |||
|
r33763 | return self._pipei | ||
|
r17192 | |||
|
r20905 | def _callcompressable(self, cmd, **args): | ||
return self._callstream(cmd, **args) | ||||
|
r17192 | def _call(self, cmd, **args): | ||
self._callstream(cmd, **args) | ||||
return self._recv() | ||||
def _callpush(self, cmd, fp, **args): | ||||
r = self._call(cmd, **args) | ||||
if r: | ||||
return '', r | ||||
|
r29727 | for d in iter(lambda: fp.read(4096), ''): | ||
|
r17192 | self._send(d) | ||
self._send("", flush=True) | ||||
r = self._recv() | ||||
if r: | ||||
return '', r | ||||
return self._recv(), '' | ||||
|
r21073 | def _calltwowaystream(self, cmd, fp, **args): | ||
r = self._call(cmd, **args) | ||||
if r: | ||||
# XXX needs to be made better | ||||
|
r29389 | raise error.Abort(_('unexpected remote reply: %s') % r) | ||
|
r29727 | for d in iter(lambda: fp.read(4096), ''): | ||
|
r21073 | self._send(d) | ||
self._send("", flush=True) | ||||
|
r33763 | return self._pipei | ||
|
r17192 | |||
|
r28438 | def _getamount(self): | ||
|
r33763 | l = self._pipei.readline() | ||
|
r17192 | if l == '\n': | ||
|
r33763 | self._readerr() | ||
|
r25243 | msg = _('check previous remote output') | ||
self._abort(error.OutOfBandError(hint=msg)) | ||||
|
r33763 | self._readerr() | ||
|
r17192 | try: | ||
|
r28438 | return int(l) | ||
|
r17192 | except ValueError: | ||
self._abort(error.ResponseError(_("unexpected response:"), l)) | ||||
|
r28438 | |||
def _recv(self): | ||||
|
r33763 | return self._pipei.read(self._getamount()) | ||
|
r17192 | |||
def _send(self, data, flush=False): | ||||
|
r33763 | self._pipeo.write("%d\n" % len(data)) | ||
|
r17192 | if data: | ||
|
r33763 | self._pipeo.write(data) | ||
|
r17192 | if flush: | ||
|
r33763 | self._pipeo.flush() | ||
self._readerr() | ||||
|
r17192 | |||
|
r35946 | def instance(ui, path, create): | ||
return sshpeer(ui, path, create=create) | ||||