##// END OF EJS Templates
sshpeer: check for safe ssh url (SEC)...
Sean Farley -
r33725:d7a1c4c1 stable
parent child Browse files
Show More
@@ -1,369 +1,371 b''
1 1 # sshpeer.py - ssh repository proxy class for mercurial
2 2 #
3 3 # Copyright 2005, 2006 Matt Mackall <mpm@selenic.com>
4 4 #
5 5 # This software may be used and distributed according to the terms of the
6 6 # GNU General Public License version 2 or any later version.
7 7
8 8 from __future__ import absolute_import
9 9
10 10 import re
11 11
12 12 from .i18n import _
13 13 from . import (
14 14 error,
15 15 pycompat,
16 16 util,
17 17 wireproto,
18 18 )
19 19
20 20 class remotelock(object):
21 21 def __init__(self, repo):
22 22 self.repo = repo
23 23 def release(self):
24 24 self.repo.unlock()
25 25 self.repo = None
26 26 def __enter__(self):
27 27 return self
28 28 def __exit__(self, exc_type, exc_val, exc_tb):
29 29 if self.repo:
30 30 self.release()
31 31 def __del__(self):
32 32 if self.repo:
33 33 self.release()
34 34
35 35 def _serverquote(s):
36 36 if not s:
37 37 return s
38 38 '''quote a string for the remote shell ... which we assume is sh'''
39 39 if re.match('[a-zA-Z0-9@%_+=:,./-]*$', s):
40 40 return s
41 41 return "'%s'" % s.replace("'", "'\\''")
42 42
43 43 def _forwardoutput(ui, pipe):
44 44 """display all data currently available on pipe as remote output.
45 45
46 46 This is non blocking."""
47 47 s = util.readpipe(pipe)
48 48 if s:
49 49 for l in s.splitlines():
50 50 ui.status(_("remote: "), l, '\n')
51 51
52 52 class doublepipe(object):
53 53 """Operate a side-channel pipe in addition of a main one
54 54
55 55 The side-channel pipe contains server output to be forwarded to the user
56 56 input. The double pipe will behave as the "main" pipe, but will ensure the
57 57 content of the "side" pipe is properly processed while we wait for blocking
58 58 call on the "main" pipe.
59 59
60 60 If large amounts of data are read from "main", the forward will cease after
61 61 the first bytes start to appear. This simplifies the implementation
62 62 without affecting actual output of sshpeer too much as we rarely issue
63 63 large read for data not yet emitted by the server.
64 64
65 65 The main pipe is expected to be a 'bufferedinputpipe' from the util module
66 66 that handle all the os specific bits. This class lives in this module
67 67 because it focus on behavior specific to the ssh protocol."""
68 68
69 69 def __init__(self, ui, main, side):
70 70 self._ui = ui
71 71 self._main = main
72 72 self._side = side
73 73
74 74 def _wait(self):
75 75 """wait until some data are available on main or side
76 76
77 77 return a pair of boolean (ismainready, issideready)
78 78
79 79 (This will only wait for data if the setup is supported by `util.poll`)
80 80 """
81 81 if getattr(self._main, 'hasbuffer', False): # getattr for classic pipe
82 82 return (True, True) # main has data, assume side is worth poking at.
83 83 fds = [self._main.fileno(), self._side.fileno()]
84 84 try:
85 85 act = util.poll(fds)
86 86 except NotImplementedError:
87 87 # non supported yet case, assume all have data.
88 88 act = fds
89 89 return (self._main.fileno() in act, self._side.fileno() in act)
90 90
91 91 def write(self, data):
92 92 return self._call('write', data)
93 93
94 94 def read(self, size):
95 95 r = self._call('read', size)
96 96 if size != 0 and not r:
97 97 # We've observed a condition that indicates the
98 98 # stdout closed unexpectedly. Check stderr one
99 99 # more time and snag anything that's there before
100 100 # letting anyone know the main part of the pipe
101 101 # closed prematurely.
102 102 _forwardoutput(self._ui, self._side)
103 103 return r
104 104
105 105 def readline(self):
106 106 return self._call('readline')
107 107
108 108 def _call(self, methname, data=None):
109 109 """call <methname> on "main", forward output of "side" while blocking
110 110 """
111 111 # data can be '' or 0
112 112 if (data is not None and not data) or self._main.closed:
113 113 _forwardoutput(self._ui, self._side)
114 114 return ''
115 115 while True:
116 116 mainready, sideready = self._wait()
117 117 if sideready:
118 118 _forwardoutput(self._ui, self._side)
119 119 if mainready:
120 120 meth = getattr(self._main, methname)
121 121 if data is None:
122 122 return meth()
123 123 else:
124 124 return meth(data)
125 125
126 126 def close(self):
127 127 return self._main.close()
128 128
129 129 def flush(self):
130 130 return self._main.flush()
131 131
132 132 class sshpeer(wireproto.wirepeer):
133 133 def __init__(self, ui, path, create=False):
134 134 self._url = path
135 135 self.ui = ui
136 136 self.pipeo = self.pipei = self.pipee = None
137 137
138 138 u = util.url(path, parsequery=False, parsefragment=False)
139 139 if u.scheme != 'ssh' or not u.host or u.path is None:
140 140 self._abort(error.RepoError(_("couldn't parse location %s") % path))
141 141
142 util.checksafessh(path)
143
142 144 self.user = u.user
143 145 if u.passwd is not None:
144 146 self._abort(error.RepoError(_("password in URL not supported")))
145 147 self.host = u.host
146 148 self.port = u.port
147 149 self.path = u.path or "."
148 150
149 151 sshcmd = self.ui.config("ui", "ssh")
150 152 remotecmd = self.ui.config("ui", "remotecmd")
151 153
152 154 args = util.sshargs(sshcmd,
153 155 _serverquote(self.host),
154 156 _serverquote(self.user),
155 157 _serverquote(self.port))
156 158
157 159 if create:
158 160 cmd = '%s %s %s' % (sshcmd, args,
159 161 util.shellquote("%s init %s" %
160 162 (_serverquote(remotecmd), _serverquote(self.path))))
161 163 ui.debug('running %s\n' % cmd)
162 164 res = ui.system(cmd, blockedtag='sshpeer')
163 165 if res != 0:
164 166 self._abort(error.RepoError(_("could not create remote repo")))
165 167
166 168 self._validaterepo(sshcmd, args, remotecmd)
167 169
168 170 def url(self):
169 171 return self._url
170 172
171 173 def _validaterepo(self, sshcmd, args, remotecmd):
172 174 # cleanup up previous run
173 175 self.cleanup()
174 176
175 177 cmd = '%s %s %s' % (sshcmd, args,
176 178 util.shellquote("%s -R %s serve --stdio" %
177 179 (_serverquote(remotecmd), _serverquote(self.path))))
178 180 self.ui.debug('running %s\n' % cmd)
179 181 cmd = util.quotecommand(cmd)
180 182
181 183 # while self.subprocess isn't used, having it allows the subprocess to
182 184 # to clean up correctly later
183 185 #
184 186 # no buffer allow the use of 'select'
185 187 # feel free to remove buffering and select usage when we ultimately
186 188 # move to threading.
187 189 sub = util.popen4(cmd, bufsize=0)
188 190 self.pipeo, self.pipei, self.pipee, self.subprocess = sub
189 191
190 192 self.pipei = util.bufferedinputpipe(self.pipei)
191 193 self.pipei = doublepipe(self.ui, self.pipei, self.pipee)
192 194 self.pipeo = doublepipe(self.ui, self.pipeo, self.pipee)
193 195
194 196 # skip any noise generated by remote shell
195 197 self._callstream("hello")
196 198 r = self._callstream("between", pairs=("%s-%s" % ("0"*40, "0"*40)))
197 199 lines = ["", "dummy"]
198 200 max_noise = 500
199 201 while lines[-1] and max_noise:
200 202 l = r.readline()
201 203 self.readerr()
202 204 if lines[-1] == "1\n" and l == "\n":
203 205 break
204 206 if l:
205 207 self.ui.debug("remote: ", l)
206 208 lines.append(l)
207 209 max_noise -= 1
208 210 else:
209 211 self._abort(error.RepoError(_('no suitable response from '
210 212 'remote hg')))
211 213
212 214 self._caps = set()
213 215 for l in reversed(lines):
214 216 if l.startswith("capabilities:"):
215 217 self._caps.update(l[:-1].split(":")[1].split())
216 218 break
217 219
218 220 def _capabilities(self):
219 221 return self._caps
220 222
221 223 def readerr(self):
222 224 _forwardoutput(self.ui, self.pipee)
223 225
224 226 def _abort(self, exception):
225 227 self.cleanup()
226 228 raise exception
227 229
228 230 def cleanup(self):
229 231 if self.pipeo is None:
230 232 return
231 233 self.pipeo.close()
232 234 self.pipei.close()
233 235 try:
234 236 # read the error descriptor until EOF
235 237 for l in self.pipee:
236 238 self.ui.status(_("remote: "), l)
237 239 except (IOError, ValueError):
238 240 pass
239 241 self.pipee.close()
240 242
241 243 __del__ = cleanup
242 244
243 245 def _submitbatch(self, req):
244 246 rsp = self._callstream("batch", cmds=wireproto.encodebatchcmds(req))
245 247 available = self._getamount()
246 248 # TODO this response parsing is probably suboptimal for large
247 249 # batches with large responses.
248 250 toread = min(available, 1024)
249 251 work = rsp.read(toread)
250 252 available -= toread
251 253 chunk = work
252 254 while chunk:
253 255 while ';' in work:
254 256 one, work = work.split(';', 1)
255 257 yield wireproto.unescapearg(one)
256 258 toread = min(available, 1024)
257 259 chunk = rsp.read(toread)
258 260 available -= toread
259 261 work += chunk
260 262 yield wireproto.unescapearg(work)
261 263
262 264 def _callstream(self, cmd, **args):
263 265 args = pycompat.byteskwargs(args)
264 266 self.ui.debug("sending %s command\n" % cmd)
265 267 self.pipeo.write("%s\n" % cmd)
266 268 _func, names = wireproto.commands[cmd]
267 269 keys = names.split()
268 270 wireargs = {}
269 271 for k in keys:
270 272 if k == '*':
271 273 wireargs['*'] = args
272 274 break
273 275 else:
274 276 wireargs[k] = args[k]
275 277 del args[k]
276 278 for k, v in sorted(wireargs.iteritems()):
277 279 self.pipeo.write("%s %d\n" % (k, len(v)))
278 280 if isinstance(v, dict):
279 281 for dk, dv in v.iteritems():
280 282 self.pipeo.write("%s %d\n" % (dk, len(dv)))
281 283 self.pipeo.write(dv)
282 284 else:
283 285 self.pipeo.write(v)
284 286 self.pipeo.flush()
285 287
286 288 return self.pipei
287 289
288 290 def _callcompressable(self, cmd, **args):
289 291 return self._callstream(cmd, **args)
290 292
291 293 def _call(self, cmd, **args):
292 294 self._callstream(cmd, **args)
293 295 return self._recv()
294 296
295 297 def _callpush(self, cmd, fp, **args):
296 298 r = self._call(cmd, **args)
297 299 if r:
298 300 return '', r
299 301 for d in iter(lambda: fp.read(4096), ''):
300 302 self._send(d)
301 303 self._send("", flush=True)
302 304 r = self._recv()
303 305 if r:
304 306 return '', r
305 307 return self._recv(), ''
306 308
307 309 def _calltwowaystream(self, cmd, fp, **args):
308 310 r = self._call(cmd, **args)
309 311 if r:
310 312 # XXX needs to be made better
311 313 raise error.Abort(_('unexpected remote reply: %s') % r)
312 314 for d in iter(lambda: fp.read(4096), ''):
313 315 self._send(d)
314 316 self._send("", flush=True)
315 317 return self.pipei
316 318
317 319 def _getamount(self):
318 320 l = self.pipei.readline()
319 321 if l == '\n':
320 322 self.readerr()
321 323 msg = _('check previous remote output')
322 324 self._abort(error.OutOfBandError(hint=msg))
323 325 self.readerr()
324 326 try:
325 327 return int(l)
326 328 except ValueError:
327 329 self._abort(error.ResponseError(_("unexpected response:"), l))
328 330
329 331 def _recv(self):
330 332 return self.pipei.read(self._getamount())
331 333
332 334 def _send(self, data, flush=False):
333 335 self.pipeo.write("%d\n" % len(data))
334 336 if data:
335 337 self.pipeo.write(data)
336 338 if flush:
337 339 self.pipeo.flush()
338 340 self.readerr()
339 341
340 342 def lock(self):
341 343 self._call("lock")
342 344 return remotelock(self)
343 345
344 346 def unlock(self):
345 347 self._call("unlock")
346 348
347 349 def addchangegroup(self, cg, source, url, lock=None):
348 350 '''Send a changegroup to the remote server. Return an integer
349 351 similar to unbundle(). DEPRECATED, since it requires locking the
350 352 remote.'''
351 353 d = self._call("addchangegroup")
352 354 if d:
353 355 self._abort(error.RepoError(_("push refused: %s") % d))
354 356 for d in iter(lambda: cg.read(4096), ''):
355 357 self.pipeo.write(d)
356 358 self.readerr()
357 359
358 360 self.pipeo.flush()
359 361
360 362 self.readerr()
361 363 r = self._recv()
362 364 if not r:
363 365 return 1
364 366 try:
365 367 return int(r)
366 368 except ValueError:
367 369 self._abort(error.ResponseError(_("unexpected response:"), r))
368 370
369 371 instance = sshpeer
General Comments 0
You need to be logged in to leave comments. Login now