~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/transport/sftp.py

  • Committer: Aaron Bentley
  • Date: 2007-06-22 22:19:13 UTC
  • mto: (2520.5.2 bzr.mpbundle)
  • mto: This revision was merged to the branch mainline in revision 2631.
  • Revision ID: abentley@panoramicfeedback.com-20070622221913-mcjioqruw8rhgnd8
Improve locking in _BaseMergeDirective.from_object

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2005 Robey Pointer <robey@lag.net>
 
2
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
 
3
#
 
4
# This program is free software; you can redistribute it and/or modify
 
5
# it under the terms of the GNU General Public License as published by
 
6
# the Free Software Foundation; either version 2 of the License, or
 
7
# (at your option) any later version.
 
8
#
 
9
# This program is distributed in the hope that it will be useful,
 
10
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
11
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
12
# GNU General Public License for more details.
 
13
#
 
14
# You should have received a copy of the GNU General Public License
 
15
# along with this program; if not, write to the Free Software
 
16
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
17
 
 
18
"""Implementation of Transport over SFTP, using paramiko."""
 
19
 
 
20
# TODO: Remove the transport-based lock_read and lock_write methods.  They'll
 
21
# then raise TransportNotPossible, which will break remote access to any
 
22
# formats which rely on OS-level locks.  That should be fine as those formats
 
23
# are pretty old, but these combinations may have to be removed from the test
 
24
# suite.  Those formats all date back to 0.7; so we should be able to remove
 
25
# these methods when we officially drop support for those formats.
 
26
 
 
27
import errno
 
28
import os
 
29
import random
 
30
import select
 
31
import socket
 
32
import stat
 
33
import sys
 
34
import time
 
35
import urllib
 
36
import urlparse
 
37
import weakref
 
38
 
 
39
from bzrlib import (
 
40
    errors,
 
41
    urlutils,
 
42
    )
 
43
from bzrlib.errors import (FileExists,
 
44
                           NoSuchFile, PathNotChild,
 
45
                           TransportError,
 
46
                           LockError,
 
47
                           PathError,
 
48
                           ParamikoNotPresent,
 
49
                           )
 
50
from bzrlib.osutils import pathjoin, fancy_rename, getcwd
 
51
from bzrlib.trace import mutter, warning
 
52
from bzrlib.transport import (
 
53
    local,
 
54
    register_urlparse_netloc_protocol,
 
55
    Server,
 
56
    split_url,
 
57
    ssh,
 
58
    Transport,
 
59
    )
 
60
 
 
61
try:
 
62
    import paramiko
 
63
except ImportError, e:
 
64
    raise ParamikoNotPresent(e)
 
65
else:
 
66
    from paramiko.sftp import (SFTP_FLAG_WRITE, SFTP_FLAG_CREATE,
 
67
                               SFTP_FLAG_EXCL, SFTP_FLAG_TRUNC,
 
68
                               CMD_HANDLE, CMD_OPEN)
 
69
    from paramiko.sftp_attr import SFTPAttributes
 
70
    from paramiko.sftp_file import SFTPFile
 
71
 
 
72
 
 
73
register_urlparse_netloc_protocol('sftp')
 
74
 
 
75
 
 
76
# This is a weakref dictionary, so that we can reuse connections
 
77
# that are still active. Long term, it might be nice to have some
 
78
# sort of expiration policy, such as disconnect if inactive for
 
79
# X seconds. But that requires a lot more fanciness.
 
80
_connected_hosts = weakref.WeakValueDictionary()
 
81
 
 
82
 
 
83
_paramiko_version = getattr(paramiko, '__version_info__', (0, 0, 0))
 
84
# don't use prefetch unless paramiko version >= 1.5.5 (there were bugs earlier)
 
85
_default_do_prefetch = (_paramiko_version >= (1, 5, 5))
 
86
 
 
87
 
 
88
def clear_connection_cache():
 
89
    """Remove all hosts from the SFTP connection cache.
 
90
 
 
91
    Primarily useful for test cases wanting to force garbage collection.
 
92
    """
 
93
    _connected_hosts.clear()
 
94
 
 
95
 
 
96
class SFTPLock(object):
 
97
    """This fakes a lock in a remote location.
 
98
    
 
99
    A present lock is indicated just by the existence of a file.  This
 
100
    doesn't work well on all transports and they are only used in 
 
101
    deprecated storage formats.
 
102
    """
 
103
    
 
104
    __slots__ = ['path', 'lock_path', 'lock_file', 'transport']
 
105
 
 
106
    def __init__(self, path, transport):
 
107
        assert isinstance(transport, SFTPTransport)
 
108
 
 
109
        self.lock_file = None
 
110
        self.path = path
 
111
        self.lock_path = path + '.write-lock'
 
112
        self.transport = transport
 
113
        try:
 
114
            # RBC 20060103 FIXME should we be using private methods here ?
 
115
            abspath = transport._remote_path(self.lock_path)
 
116
            self.lock_file = transport._sftp_open_exclusive(abspath)
 
117
        except FileExists:
 
118
            raise LockError('File %r already locked' % (self.path,))
 
119
 
 
120
    def __del__(self):
 
121
        """Should this warn, or actually try to cleanup?"""
 
122
        if self.lock_file:
 
123
            warning("SFTPLock %r not explicitly unlocked" % (self.path,))
 
124
            self.unlock()
 
125
 
 
126
    def unlock(self):
 
127
        if not self.lock_file:
 
128
            return
 
129
        self.lock_file.close()
 
130
        self.lock_file = None
 
131
        try:
 
132
            self.transport.delete(self.lock_path)
 
133
        except (NoSuchFile,):
 
134
            # What specific errors should we catch here?
 
135
            pass
 
136
 
 
137
 
 
138
class SFTPUrlHandling(Transport):
 
139
    """Mix-in that does common handling of SSH/SFTP URLs."""
 
140
 
 
141
    def __init__(self, base):
 
142
        self._parse_url(base)
 
143
        base = self._unparse_url(self._path)
 
144
        if base[-1] != '/':
 
145
            base += '/'
 
146
        super(SFTPUrlHandling, self).__init__(base)
 
147
 
 
148
    def _parse_url(self, url):
 
149
        (self._scheme,
 
150
         self._username, self._password,
 
151
         self._host, self._port, self._path) = self._split_url(url)
 
152
 
 
153
    def _unparse_url(self, path):
 
154
        """Return a URL for a path relative to this transport.
 
155
        """
 
156
        path = urllib.quote(path)
 
157
        # handle homedir paths
 
158
        if not path.startswith('/'):
 
159
            path = "/~/" + path
 
160
        netloc = urllib.quote(self._host)
 
161
        if self._username is not None:
 
162
            netloc = '%s@%s' % (urllib.quote(self._username), netloc)
 
163
        if self._port is not None:
 
164
            netloc = '%s:%d' % (netloc, self._port)
 
165
        return urlparse.urlunparse((self._scheme, netloc, path, '', '', ''))
 
166
 
 
167
    def _split_url(self, url):
 
168
        (scheme, username, password, host, port, path) = split_url(url)
 
169
        ## assert scheme == 'sftp'
 
170
 
 
171
        # the initial slash should be removed from the path, and treated
 
172
        # as a homedir relative path (the path begins with a double slash
 
173
        # if it is absolute).
 
174
        # see draft-ietf-secsh-scp-sftp-ssh-uri-03.txt
 
175
        # RBC 20060118 we are not using this as its too user hostile. instead
 
176
        # we are following lftp and using /~/foo to mean '~/foo'.
 
177
        # handle homedir paths
 
178
        if path.startswith('/~/'):
 
179
            path = path[3:]
 
180
        elif path == '/~':
 
181
            path = ''
 
182
        return (scheme, username, password, host, port, path)
 
183
 
 
184
    def abspath(self, relpath):
 
185
        """Return the full url to the given relative path.
 
186
        
 
187
        @param relpath: the relative path or path components
 
188
        @type relpath: str or list
 
189
        """
 
190
        return self._unparse_url(self._remote_path(relpath))
 
191
    
 
192
    def _remote_path(self, relpath):
 
193
        """Return the path to be passed along the sftp protocol for relpath.
 
194
        
 
195
        :param relpath: is a urlencoded string.
 
196
        """
 
197
        return self._combine_paths(self._path, relpath)
 
198
 
 
199
 
 
200
class SFTPTransport(SFTPUrlHandling):
 
201
    """Transport implementation for SFTP access."""
 
202
 
 
203
    _do_prefetch = _default_do_prefetch
 
204
    # TODO: jam 20060717 Conceivably these could be configurable, either
 
205
    #       by auto-tuning at run-time, or by a configuration (per host??)
 
206
    #       but the performance curve is pretty flat, so just going with
 
207
    #       reasonable defaults.
 
208
    _max_readv_combine = 200
 
209
    # Having to round trip to the server means waiting for a response,
 
210
    # so it is better to download extra bytes.
 
211
    # 8KiB had good performance for both local and remote network operations
 
212
    _bytes_to_read_before_seek = 8192
 
213
 
 
214
    # The sftp spec says that implementations SHOULD allow reads
 
215
    # to be at least 32K. paramiko.readv() does an async request
 
216
    # for the chunks. So we need to keep it within a single request
 
217
    # size for paramiko <= 1.6.1. paramiko 1.6.2 will probably chop
 
218
    # up the request itself, rather than us having to worry about it
 
219
    _max_request_size = 32768
 
220
 
 
221
    def __init__(self, base, clone_from=None):
 
222
        super(SFTPTransport, self).__init__(base)
 
223
        if clone_from is None:
 
224
            self._sftp_connect()
 
225
        else:
 
226
            # use the same ssh connection, etc
 
227
            self._sftp = clone_from._sftp
 
228
        # super saves 'self.base'
 
229
    
 
230
    def should_cache(self):
 
231
        """
 
232
        Return True if the data pulled across should be cached locally.
 
233
        """
 
234
        return True
 
235
 
 
236
    def clone(self, offset=None):
 
237
        """
 
238
        Return a new SFTPTransport with root at self.base + offset.
 
239
        We share the same SFTP session between such transports, because it's
 
240
        fairly expensive to set them up.
 
241
        """
 
242
        if offset is None:
 
243
            return SFTPTransport(self.base, self)
 
244
        else:
 
245
            return SFTPTransport(self.abspath(offset), self)
 
246
 
 
247
    def _remote_path(self, relpath):
 
248
        """Return the path to be passed along the sftp protocol for relpath.
 
249
        
 
250
        relpath is a urlencoded string.
 
251
 
 
252
        :return: a path prefixed with / for regular abspath-based urls, or a
 
253
            path that does not begin with / for urls which begin with /~/.
 
254
        """
 
255
        # how does this work? 
 
256
        # it processes relpath with respect to 
 
257
        # our state:
 
258
        # firstly we create a path to evaluate: 
 
259
        # if relpath is an abspath or homedir path, its the entire thing
 
260
        # otherwise we join our base with relpath
 
261
        # then we eliminate all empty segments (double //'s) outside the first
 
262
        # two elements of the list. This avoids problems with trailing 
 
263
        # slashes, or other abnormalities.
 
264
        # finally we evaluate the entire path in a single pass
 
265
        # '.'s are stripped,
 
266
        # '..' result in popping the left most already 
 
267
        # processed path (which can never be empty because of the check for
 
268
        # abspath and homedir meaning that its not, or that we've used our
 
269
        # path. If the pop would pop the root, we ignore it.
 
270
 
 
271
        # Specific case examinations:
 
272
        # remove the special casefor ~: if the current root is ~/ popping of it
 
273
        # = / thus our seed for a ~ based path is ['', '~']
 
274
        # and if we end up with [''] then we had basically ('', '..') (which is
 
275
        # '/..' so we append '' if the length is one, and assert that the first
 
276
        # element is still ''. Lastly, if we end with ['', '~'] as a prefix for
 
277
        # the output, we've got a homedir path, so we strip that prefix before
 
278
        # '/' joining the resulting list.
 
279
        #
 
280
        # case one: '/' -> ['', ''] cannot shrink
 
281
        # case two: '/' + '../foo' -> ['', 'foo'] (take '', '', '..', 'foo')
 
282
        #           and pop the second '' for the '..', append 'foo'
 
283
        # case three: '/~/' -> ['', '~', ''] 
 
284
        # case four: '/~/' + '../foo' -> ['', '~', '', '..', 'foo'],
 
285
        #           and we want to get '/foo' - the empty path in the middle
 
286
        #           needs to be stripped, then normal path manipulation will 
 
287
        #           work.
 
288
        # case five: '/..' ['', '..'], we want ['', '']
 
289
        #            stripping '' outside the first two is ok
 
290
        #            ignore .. if its too high up
 
291
        #
 
292
        # lastly this code is possibly reusable by FTP, but not reusable by
 
293
        # local paths: ~ is resolvable correctly, nor by HTTP or the smart
 
294
        # server: ~ is resolved remotely.
 
295
        # 
 
296
        # however, a version of this that acts on self.base is possible to be
 
297
        # written which manipulates the URL in canonical form, and would be
 
298
        # reusable for all transports, if a flag for allowing ~/ at all was
 
299
        # provided.
 
300
        assert isinstance(relpath, basestring)
 
301
        relpath = urlutils.unescape(relpath)
 
302
 
 
303
        # case 1)
 
304
        if relpath.startswith('/'):
 
305
            # abspath - normal split is fine.
 
306
            current_path = relpath.split('/')
 
307
        elif relpath.startswith('~/'):
 
308
            # root is homedir based: normal split and prefix '' to remote the
 
309
            # special case
 
310
            current_path = [''].extend(relpath.split('/'))
 
311
        else:
 
312
            # root is from the current directory:
 
313
            if self._path.startswith('/'):
 
314
                # abspath, take the regular split
 
315
                current_path = []
 
316
            else:
 
317
                # homedir based, add the '', '~' not present in self._path
 
318
                current_path = ['', '~']
 
319
            # add our current dir
 
320
            current_path.extend(self._path.split('/'))
 
321
            # add the users relpath
 
322
            current_path.extend(relpath.split('/'))
 
323
        # strip '' segments that are not in the first one - the leading /.
 
324
        to_process = current_path[:1]
 
325
        for segment in current_path[1:]:
 
326
            if segment != '':
 
327
                to_process.append(segment)
 
328
 
 
329
        # process '.' and '..' segments into output_path.
 
330
        output_path = []
 
331
        for segment in to_process:
 
332
            if segment == '..':
 
333
                # directory pop. Remove a directory 
 
334
                # as long as we are not at the root
 
335
                if len(output_path) > 1:
 
336
                    output_path.pop()
 
337
                # else: pass
 
338
                # cannot pop beyond the root, so do nothing
 
339
            elif segment == '.':
 
340
                continue # strip the '.' from the output.
 
341
            else:
 
342
                # this will append '' to output_path for the root elements,
 
343
                # which is appropriate: its why we strip '' in the first pass.
 
344
                output_path.append(segment)
 
345
 
 
346
        # check output special cases:
 
347
        if output_path == ['']:
 
348
            # [''] -> ['', '']
 
349
            output_path = ['', '']
 
350
        elif output_path[:2] == ['', '~']:
 
351
            # ['', '~', ...] -> ...
 
352
            output_path = output_path[2:]
 
353
        path = '/'.join(output_path)
 
354
        return path
 
355
 
 
356
    def relpath(self, abspath):
 
357
        scheme, username, password, host, port, path = self._split_url(abspath)
 
358
        error = []
 
359
        if (username != self._username):
 
360
            error.append('username mismatch')
 
361
        if (host != self._host):
 
362
            error.append('host mismatch')
 
363
        if (port != self._port):
 
364
            error.append('port mismatch')
 
365
        if (not path.startswith(self._path)):
 
366
            error.append('path mismatch')
 
367
        if error:
 
368
            extra = ': ' + ', '.join(error)
 
369
            raise PathNotChild(abspath, self.base, extra=extra)
 
370
        pl = len(self._path)
 
371
        return path[pl:].strip('/')
 
372
 
 
373
    def has(self, relpath):
 
374
        """
 
375
        Does the target location exist?
 
376
        """
 
377
        try:
 
378
            self._sftp.stat(self._remote_path(relpath))
 
379
            return True
 
380
        except IOError:
 
381
            return False
 
382
 
 
383
    def get(self, relpath):
 
384
        """
 
385
        Get the file at the given relative path.
 
386
 
 
387
        :param relpath: The relative path to the file
 
388
        """
 
389
        try:
 
390
            path = self._remote_path(relpath)
 
391
            f = self._sftp.file(path, mode='rb')
 
392
            if self._do_prefetch and (getattr(f, 'prefetch', None) is not None):
 
393
                f.prefetch()
 
394
            return f
 
395
        except (IOError, paramiko.SSHException), e:
 
396
            self._translate_io_exception(e, path, ': error retrieving')
 
397
 
 
398
    def readv(self, relpath, offsets):
 
399
        """See Transport.readv()"""
 
400
        # We overload the default readv() because we want to use a file
 
401
        # that does not have prefetch enabled.
 
402
        # Also, if we have a new paramiko, it implements an async readv()
 
403
        if not offsets:
 
404
            return
 
405
 
 
406
        try:
 
407
            path = self._remote_path(relpath)
 
408
            fp = self._sftp.file(path, mode='rb')
 
409
            readv = getattr(fp, 'readv', None)
 
410
            if readv:
 
411
                return self._sftp_readv(fp, offsets, relpath)
 
412
            mutter('seek and read %s offsets', len(offsets))
 
413
            return self._seek_and_read(fp, offsets, relpath)
 
414
        except (IOError, paramiko.SSHException), e:
 
415
            self._translate_io_exception(e, path, ': error retrieving')
 
416
 
 
417
    def _sftp_readv(self, fp, offsets, relpath='<unknown>'):
 
418
        """Use the readv() member of fp to do async readv.
 
419
 
 
420
        And then read them using paramiko.readv(). paramiko.readv()
 
421
        does not support ranges > 64K, so it caps the request size, and
 
422
        just reads until it gets all the stuff it wants
 
423
        """
 
424
        offsets = list(offsets)
 
425
        sorted_offsets = sorted(offsets)
 
426
 
 
427
        # The algorithm works as follows:
 
428
        # 1) Coalesce nearby reads into a single chunk
 
429
        #    This generates a list of combined regions, the total size
 
430
        #    and the size of the sub regions. This coalescing step is limited
 
431
        #    in the number of nearby chunks to combine, and is allowed to
 
432
        #    skip small breaks in the requests. Limiting it makes sure that
 
433
        #    we can start yielding some data earlier, and skipping means we
 
434
        #    make fewer requests. (Beneficial even when using async)
 
435
        # 2) Break up this combined regions into chunks that are smaller
 
436
        #    than 64KiB. Technically the limit is 65536, but we are a
 
437
        #    little bit conservative. This is because sftp has a maximum
 
438
        #    return chunk size of 64KiB (max size of an unsigned short)
 
439
        # 3) Issue a readv() to paramiko to create an async request for
 
440
        #    all of this data
 
441
        # 4) Read in the data as it comes back, until we've read one
 
442
        #    continuous section as determined in step 1
 
443
        # 5) Break up the full sections into hunks for the original requested
 
444
        #    offsets. And put them in a cache
 
445
        # 6) Check if the next request is in the cache, and if it is, remove
 
446
        #    it from the cache, and yield its data. Continue until no more
 
447
        #    entries are in the cache.
 
448
        # 7) loop back to step 4 until all data has been read
 
449
        #
 
450
        # TODO: jam 20060725 This could be optimized one step further, by
 
451
        #       attempting to yield whatever data we have read, even before
 
452
        #       the first coallesced section has been fully processed.
 
453
 
 
454
        # When coalescing for use with readv(), we don't really need to
 
455
        # use any fudge factor, because the requests are made asynchronously
 
456
        coalesced = list(self._coalesce_offsets(sorted_offsets,
 
457
                               limit=self._max_readv_combine,
 
458
                               fudge_factor=0,
 
459
                               ))
 
460
        requests = []
 
461
        for c_offset in coalesced:
 
462
            start = c_offset.start
 
463
            size = c_offset.length
 
464
 
 
465
            # We need to break this up into multiple requests
 
466
            while size > 0:
 
467
                next_size = min(size, self._max_request_size)
 
468
                requests.append((start, next_size))
 
469
                size -= next_size
 
470
                start += next_size
 
471
 
 
472
        mutter('SFTP.readv() %s offsets => %s coalesced => %s requests',
 
473
                len(offsets), len(coalesced), len(requests))
 
474
 
 
475
        # Queue the current read until we have read the full coalesced section
 
476
        cur_data = []
 
477
        cur_data_len = 0
 
478
        cur_coalesced_stack = iter(coalesced)
 
479
        cur_coalesced = cur_coalesced_stack.next()
 
480
 
 
481
        # Cache the results, but only until they have been fulfilled
 
482
        data_map = {}
 
483
        # turn the list of offsets into a stack
 
484
        offset_stack = iter(offsets)
 
485
        cur_offset_and_size = offset_stack.next()
 
486
 
 
487
        for data in fp.readv(requests):
 
488
            cur_data += data
 
489
            cur_data_len += len(data)
 
490
 
 
491
            if cur_data_len < cur_coalesced.length:
 
492
                continue
 
493
            assert cur_data_len == cur_coalesced.length, \
 
494
                "Somehow we read too much: %s != %s" % (cur_data_len,
 
495
                                                        cur_coalesced.length)
 
496
            all_data = ''.join(cur_data)
 
497
            cur_data = []
 
498
            cur_data_len = 0
 
499
 
 
500
            for suboffset, subsize in cur_coalesced.ranges:
 
501
                key = (cur_coalesced.start+suboffset, subsize)
 
502
                data_map[key] = all_data[suboffset:suboffset+subsize]
 
503
 
 
504
            # Now that we've read some data, see if we can yield anything back
 
505
            while cur_offset_and_size in data_map:
 
506
                this_data = data_map.pop(cur_offset_and_size)
 
507
                yield cur_offset_and_size[0], this_data
 
508
                cur_offset_and_size = offset_stack.next()
 
509
 
 
510
            # We read a coalesced entry, so mark it as done
 
511
            cur_coalesced = None
 
512
            # Now that we've read all of the data for this coalesced section
 
513
            # on to the next
 
514
            cur_coalesced = cur_coalesced_stack.next()
 
515
 
 
516
        if cur_coalesced is not None:
 
517
            raise errors.ShortReadvError(relpath, cur_coalesced.start,
 
518
                cur_coalesced.length, len(data))
 
519
 
 
520
    def put_file(self, relpath, f, mode=None):
 
521
        """
 
522
        Copy the file-like object into the location.
 
523
 
 
524
        :param relpath: Location to put the contents, relative to base.
 
525
        :param f:       File-like object.
 
526
        :param mode: The final mode for the file
 
527
        """
 
528
        final_path = self._remote_path(relpath)
 
529
        self._put(final_path, f, mode=mode)
 
530
 
 
531
    def _put(self, abspath, f, mode=None):
 
532
        """Helper function so both put() and copy_abspaths can reuse the code"""
 
533
        tmp_abspath = '%s.tmp.%.9f.%d.%d' % (abspath, time.time(),
 
534
                        os.getpid(), random.randint(0,0x7FFFFFFF))
 
535
        fout = self._sftp_open_exclusive(tmp_abspath, mode=mode)
 
536
        closed = False
 
537
        try:
 
538
            try:
 
539
                fout.set_pipelined(True)
 
540
                self._pump(f, fout)
 
541
            except (IOError, paramiko.SSHException), e:
 
542
                self._translate_io_exception(e, tmp_abspath)
 
543
            # XXX: This doesn't truly help like we would like it to.
 
544
            #      The problem is that openssh strips sticky bits. So while we
 
545
            #      can properly set group write permission, we lose the group
 
546
            #      sticky bit. So it is probably best to stop chmodding, and
 
547
            #      just tell users that they need to set the umask correctly.
 
548
            #      The attr.st_mode = mode, in _sftp_open_exclusive
 
549
            #      will handle when the user wants the final mode to be more 
 
550
            #      restrictive. And then we avoid a round trip. Unless 
 
551
            #      paramiko decides to expose an async chmod()
 
552
 
 
553
            # This is designed to chmod() right before we close.
 
554
            # Because we set_pipelined() earlier, theoretically we might 
 
555
            # avoid the round trip for fout.close()
 
556
            if mode is not None:
 
557
                self._sftp.chmod(tmp_abspath, mode)
 
558
            fout.close()
 
559
            closed = True
 
560
            self._rename_and_overwrite(tmp_abspath, abspath)
 
561
        except Exception, e:
 
562
            # If we fail, try to clean up the temporary file
 
563
            # before we throw the exception
 
564
            # but don't let another exception mess things up
 
565
            # Write out the traceback, because otherwise
 
566
            # the catch and throw destroys it
 
567
            import traceback
 
568
            mutter(traceback.format_exc())
 
569
            try:
 
570
                if not closed:
 
571
                    fout.close()
 
572
                self._sftp.remove(tmp_abspath)
 
573
            except:
 
574
                # raise the saved except
 
575
                raise e
 
576
            # raise the original with its traceback if we can.
 
577
            raise
 
578
 
 
579
    def _put_non_atomic_helper(self, relpath, writer, mode=None,
 
580
                               create_parent_dir=False,
 
581
                               dir_mode=None):
 
582
        abspath = self._remote_path(relpath)
 
583
 
 
584
        # TODO: jam 20060816 paramiko doesn't publicly expose a way to
 
585
        #       set the file mode at create time. If it does, use it.
 
586
        #       But for now, we just chmod later anyway.
 
587
 
 
588
        def _open_and_write_file():
 
589
            """Try to open the target file, raise error on failure"""
 
590
            fout = None
 
591
            try:
 
592
                try:
 
593
                    fout = self._sftp.file(abspath, mode='wb')
 
594
                    fout.set_pipelined(True)
 
595
                    writer(fout)
 
596
                except (paramiko.SSHException, IOError), e:
 
597
                    self._translate_io_exception(e, abspath,
 
598
                                                 ': unable to open')
 
599
 
 
600
                # This is designed to chmod() right before we close.
 
601
                # Because we set_pipelined() earlier, theoretically we might 
 
602
                # avoid the round trip for fout.close()
 
603
                if mode is not None:
 
604
                    self._sftp.chmod(abspath, mode)
 
605
            finally:
 
606
                if fout is not None:
 
607
                    fout.close()
 
608
 
 
609
        if not create_parent_dir:
 
610
            _open_and_write_file()
 
611
            return
 
612
 
 
613
        # Try error handling to create the parent directory if we need to
 
614
        try:
 
615
            _open_and_write_file()
 
616
        except NoSuchFile:
 
617
            # Try to create the parent directory, and then go back to
 
618
            # writing the file
 
619
            parent_dir = os.path.dirname(abspath)
 
620
            self._mkdir(parent_dir, dir_mode)
 
621
            _open_and_write_file()
 
622
 
 
623
    def put_file_non_atomic(self, relpath, f, mode=None,
 
624
                            create_parent_dir=False,
 
625
                            dir_mode=None):
 
626
        """Copy the file-like object into the target location.
 
627
 
 
628
        This function is not strictly safe to use. It is only meant to
 
629
        be used when you already know that the target does not exist.
 
630
        It is not safe, because it will open and truncate the remote
 
631
        file. So there may be a time when the file has invalid contents.
 
632
 
 
633
        :param relpath: The remote location to put the contents.
 
634
        :param f:       File-like object.
 
635
        :param mode:    Possible access permissions for new file.
 
636
                        None means do not set remote permissions.
 
637
        :param create_parent_dir: If we cannot create the target file because
 
638
                        the parent directory does not exist, go ahead and
 
639
                        create it, and then try again.
 
640
        """
 
641
        def writer(fout):
 
642
            self._pump(f, fout)
 
643
        self._put_non_atomic_helper(relpath, writer, mode=mode,
 
644
                                    create_parent_dir=create_parent_dir,
 
645
                                    dir_mode=dir_mode)
 
646
 
 
647
    def put_bytes_non_atomic(self, relpath, bytes, mode=None,
 
648
                             create_parent_dir=False,
 
649
                             dir_mode=None):
 
650
        def writer(fout):
 
651
            fout.write(bytes)
 
652
        self._put_non_atomic_helper(relpath, writer, mode=mode,
 
653
                                    create_parent_dir=create_parent_dir,
 
654
                                    dir_mode=dir_mode)
 
655
 
 
656
    def iter_files_recursive(self):
 
657
        """Walk the relative paths of all files in this transport."""
 
658
        queue = list(self.list_dir('.'))
 
659
        while queue:
 
660
            relpath = queue.pop(0)
 
661
            st = self.stat(relpath)
 
662
            if stat.S_ISDIR(st.st_mode):
 
663
                for i, basename in enumerate(self.list_dir(relpath)):
 
664
                    queue.insert(i, relpath+'/'+basename)
 
665
            else:
 
666
                yield relpath
 
667
 
 
668
    def _mkdir(self, abspath, mode=None):
 
669
        if mode is None:
 
670
            local_mode = 0777
 
671
        else:
 
672
            local_mode = mode
 
673
        try:
 
674
            self._sftp.mkdir(abspath, local_mode)
 
675
            if mode is not None:
 
676
                self._sftp.chmod(abspath, mode=mode)
 
677
        except (paramiko.SSHException, IOError), e:
 
678
            self._translate_io_exception(e, abspath, ': unable to mkdir',
 
679
                failure_exc=FileExists)
 
680
 
 
681
    def mkdir(self, relpath, mode=None):
 
682
        """Create a directory at the given path."""
 
683
        self._mkdir(self._remote_path(relpath), mode=mode)
 
684
 
 
685
    def _translate_io_exception(self, e, path, more_info='', 
 
686
                                failure_exc=PathError):
 
687
        """Translate a paramiko or IOError into a friendlier exception.
 
688
 
 
689
        :param e: The original exception
 
690
        :param path: The path in question when the error is raised
 
691
        :param more_info: Extra information that can be included,
 
692
                          such as what was going on
 
693
        :param failure_exc: Paramiko has the super fun ability to raise completely
 
694
                           opaque errors that just set "e.args = ('Failure',)" with
 
695
                           no more information.
 
696
                           If this parameter is set, it defines the exception 
 
697
                           to raise in these cases.
 
698
        """
 
699
        # paramiko seems to generate detailless errors.
 
700
        self._translate_error(e, path, raise_generic=False)
 
701
        if getattr(e, 'args', None) is not None:
 
702
            if (e.args == ('No such file or directory',) or
 
703
                e.args == ('No such file',)):
 
704
                raise NoSuchFile(path, str(e) + more_info)
 
705
            if (e.args == ('mkdir failed',)):
 
706
                raise FileExists(path, str(e) + more_info)
 
707
            # strange but true, for the paramiko server.
 
708
            if (e.args == ('Failure',)):
 
709
                raise failure_exc(path, str(e) + more_info)
 
710
            mutter('Raising exception with args %s', e.args)
 
711
        if getattr(e, 'errno', None) is not None:
 
712
            mutter('Raising exception with errno %s', e.errno)
 
713
        raise e
 
714
 
 
715
    def append_file(self, relpath, f, mode=None):
 
716
        """
 
717
        Append the text in the file-like object into the final
 
718
        location.
 
719
        """
 
720
        try:
 
721
            path = self._remote_path(relpath)
 
722
            fout = self._sftp.file(path, 'ab')
 
723
            if mode is not None:
 
724
                self._sftp.chmod(path, mode)
 
725
            result = fout.tell()
 
726
            self._pump(f, fout)
 
727
            return result
 
728
        except (IOError, paramiko.SSHException), e:
 
729
            self._translate_io_exception(e, relpath, ': unable to append')
 
730
 
 
731
    def rename(self, rel_from, rel_to):
 
732
        """Rename without special overwriting"""
 
733
        try:
 
734
            self._sftp.rename(self._remote_path(rel_from),
 
735
                              self._remote_path(rel_to))
 
736
        except (IOError, paramiko.SSHException), e:
 
737
            self._translate_io_exception(e, rel_from,
 
738
                    ': unable to rename to %r' % (rel_to))
 
739
 
 
740
    def _rename_and_overwrite(self, abs_from, abs_to):
 
741
        """Do a fancy rename on the remote server.
 
742
        
 
743
        Using the implementation provided by osutils.
 
744
        """
 
745
        try:
 
746
            fancy_rename(abs_from, abs_to,
 
747
                    rename_func=self._sftp.rename,
 
748
                    unlink_func=self._sftp.remove)
 
749
        except (IOError, paramiko.SSHException), e:
 
750
            self._translate_io_exception(e, abs_from, ': unable to rename to %r' % (abs_to))
 
751
 
 
752
    def move(self, rel_from, rel_to):
 
753
        """Move the item at rel_from to the location at rel_to"""
 
754
        path_from = self._remote_path(rel_from)
 
755
        path_to = self._remote_path(rel_to)
 
756
        self._rename_and_overwrite(path_from, path_to)
 
757
 
 
758
    def delete(self, relpath):
 
759
        """Delete the item at relpath"""
 
760
        path = self._remote_path(relpath)
 
761
        try:
 
762
            self._sftp.remove(path)
 
763
        except (IOError, paramiko.SSHException), e:
 
764
            self._translate_io_exception(e, path, ': unable to delete')
 
765
            
 
766
    def listable(self):
 
767
        """Return True if this store supports listing."""
 
768
        return True
 
769
 
 
770
    def list_dir(self, relpath):
 
771
        """
 
772
        Return a list of all files at the given location.
 
773
        """
 
774
        # does anything actually use this?
 
775
        # -- Unknown
 
776
        # This is at least used by copy_tree for remote upgrades.
 
777
        # -- David Allouche 2006-08-11
 
778
        path = self._remote_path(relpath)
 
779
        try:
 
780
            entries = self._sftp.listdir(path)
 
781
        except (IOError, paramiko.SSHException), e:
 
782
            self._translate_io_exception(e, path, ': failed to list_dir')
 
783
        return [urlutils.escape(entry) for entry in entries]
 
784
 
 
785
    def rmdir(self, relpath):
 
786
        """See Transport.rmdir."""
 
787
        path = self._remote_path(relpath)
 
788
        try:
 
789
            return self._sftp.rmdir(path)
 
790
        except (IOError, paramiko.SSHException), e:
 
791
            self._translate_io_exception(e, path, ': failed to rmdir')
 
792
 
 
793
    def stat(self, relpath):
 
794
        """Return the stat information for a file."""
 
795
        path = self._remote_path(relpath)
 
796
        try:
 
797
            return self._sftp.stat(path)
 
798
        except (IOError, paramiko.SSHException), e:
 
799
            self._translate_io_exception(e, path, ': unable to stat')
 
800
 
 
801
    def lock_read(self, relpath):
 
802
        """
 
803
        Lock the given file for shared (read) access.
 
804
        :return: A lock object, which has an unlock() member function
 
805
        """
 
806
        # FIXME: there should be something clever i can do here...
 
807
        class BogusLock(object):
 
808
            def __init__(self, path):
 
809
                self.path = path
 
810
            def unlock(self):
 
811
                pass
 
812
        return BogusLock(relpath)
 
813
 
 
814
    def lock_write(self, relpath):
 
815
        """
 
816
        Lock the given file for exclusive (write) access.
 
817
        WARNING: many transports do not support this, so trying avoid using it
 
818
 
 
819
        :return: A lock object, which has an unlock() member function
 
820
        """
 
821
        # This is a little bit bogus, but basically, we create a file
 
822
        # which should not already exist, and if it does, we assume
 
823
        # that there is a lock, and if it doesn't, the we assume
 
824
        # that we have taken the lock.
 
825
        return SFTPLock(relpath, self)
 
826
 
 
827
    def _sftp_connect(self):
 
828
        """Connect to the remote sftp server.
 
829
        After this, self._sftp should have a valid connection (or
 
830
        we raise an TransportError 'could not connect').
 
831
 
 
832
        TODO: Raise a more reasonable ConnectionFailed exception
 
833
        """
 
834
        self._sftp = _sftp_connect(self._host, self._port, self._username,
 
835
                self._password)
 
836
 
 
837
    def _sftp_open_exclusive(self, abspath, mode=None):
 
838
        """Open a remote path exclusively.
 
839
 
 
840
        SFTP supports O_EXCL (SFTP_FLAG_EXCL), which fails if
 
841
        the file already exists. However it does not expose this
 
842
        at the higher level of SFTPClient.open(), so we have to
 
843
        sneak away with it.
 
844
 
 
845
        WARNING: This breaks the SFTPClient abstraction, so it
 
846
        could easily break against an updated version of paramiko.
 
847
 
 
848
        :param abspath: The remote absolute path where the file should be opened
 
849
        :param mode: The mode permissions bits for the new file
 
850
        """
 
851
        # TODO: jam 20060816 Paramiko >= 1.6.2 (probably earlier) supports
 
852
        #       using the 'x' flag to indicate SFTP_FLAG_EXCL.
 
853
        #       However, there is no way to set the permission mode at open 
 
854
        #       time using the sftp_client.file() functionality.
 
855
        path = self._sftp._adjust_cwd(abspath)
 
856
        # mutter('sftp abspath %s => %s', abspath, path)
 
857
        attr = SFTPAttributes()
 
858
        if mode is not None:
 
859
            attr.st_mode = mode
 
860
        omode = (SFTP_FLAG_WRITE | SFTP_FLAG_CREATE 
 
861
                | SFTP_FLAG_TRUNC | SFTP_FLAG_EXCL)
 
862
        try:
 
863
            t, msg = self._sftp._request(CMD_OPEN, path, omode, attr)
 
864
            if t != CMD_HANDLE:
 
865
                raise TransportError('Expected an SFTP handle')
 
866
            handle = msg.get_string()
 
867
            return SFTPFile(self._sftp, handle, 'wb', -1)
 
868
        except (paramiko.SSHException, IOError), e:
 
869
            self._translate_io_exception(e, abspath, ': unable to open',
 
870
                failure_exc=FileExists)
 
871
 
 
872
    def _can_roundtrip_unix_modebits(self):
 
873
        if sys.platform == 'win32':
 
874
            # anyone else?
 
875
            return False
 
876
        else:
 
877
            return True
 
878
 
 
879
# ------------- server test implementation --------------
 
880
import threading
 
881
 
 
882
from bzrlib.tests.stub_sftp import StubServer, StubSFTPServer
 
883
 
 
884
STUB_SERVER_KEY = """
 
885
-----BEGIN RSA PRIVATE KEY-----
 
886
MIICWgIBAAKBgQDTj1bqB4WmayWNPB+8jVSYpZYk80Ujvj680pOTh2bORBjbIAyz
 
887
oWGW+GUjzKxTiiPvVmxFgx5wdsFvF03v34lEVVhMpouqPAYQ15N37K/ir5XY+9m/
 
888
d8ufMCkjeXsQkKqFbAlQcnWMCRnOoPHS3I4vi6hmnDDeeYTSRvfLbW0fhwIBIwKB
 
889
gBIiOqZYaoqbeD9OS9z2K9KR2atlTxGxOJPXiP4ESqP3NVScWNwyZ3NXHpyrJLa0
 
890
EbVtzsQhLn6rF+TzXnOlcipFvjsem3iYzCpuChfGQ6SovTcOjHV9z+hnpXvQ/fon
 
891
soVRZY65wKnF7IAoUwTmJS9opqgrN6kRgCd3DASAMd1bAkEA96SBVWFt/fJBNJ9H
 
892
tYnBKZGw0VeHOYmVYbvMSstssn8un+pQpUm9vlG/bp7Oxd/m+b9KWEh2xPfv6zqU
 
893
avNwHwJBANqzGZa/EpzF4J8pGti7oIAPUIDGMtfIcmqNXVMckrmzQ2vTfqtkEZsA
 
894
4rE1IERRyiJQx6EJsz21wJmGV9WJQ5kCQQDwkS0uXqVdFzgHO6S++tjmjYcxwr3g
 
895
H0CoFYSgbddOT6miqRskOQF3DZVkJT3kyuBgU2zKygz52ukQZMqxCb1fAkASvuTv
 
896
qfpH87Qq5kQhNKdbbwbmd2NxlNabazPijWuphGTdW0VfJdWfklyS2Kr+iqrs/5wV
 
897
HhathJt636Eg7oIjAkA8ht3MQ+XSl9yIJIS8gVpbPxSw5OMfw0PjVE7tBdQruiSc
 
898
nvuQES5C9BMHjF39LZiGH1iLQy7FgdHyoP+eodI7
 
899
-----END RSA PRIVATE KEY-----
 
900
"""
 
901
 
 
902
 
 
903
class SocketListener(threading.Thread):
 
904
 
 
905
    def __init__(self, callback):
 
906
        threading.Thread.__init__(self)
 
907
        self._callback = callback
 
908
        self._socket = socket.socket()
 
909
        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 
910
        self._socket.bind(('localhost', 0))
 
911
        self._socket.listen(1)
 
912
        self.port = self._socket.getsockname()[1]
 
913
        self._stop_event = threading.Event()
 
914
 
 
915
    def stop(self):
 
916
        # called from outside this thread
 
917
        self._stop_event.set()
 
918
        # use a timeout here, because if the test fails, the server thread may
 
919
        # never notice the stop_event.
 
920
        self.join(5.0)
 
921
        self._socket.close()
 
922
 
 
923
    def run(self):
 
924
        while True:
 
925
            readable, writable_unused, exception_unused = \
 
926
                select.select([self._socket], [], [], 0.1)
 
927
            if self._stop_event.isSet():
 
928
                return
 
929
            if len(readable) == 0:
 
930
                continue
 
931
            try:
 
932
                s, addr_unused = self._socket.accept()
 
933
                # because the loopback socket is inline, and transports are
 
934
                # never explicitly closed, best to launch a new thread.
 
935
                threading.Thread(target=self._callback, args=(s,)).start()
 
936
            except socket.error, x:
 
937
                sys.excepthook(*sys.exc_info())
 
938
                warning('Socket error during accept() within unit test server'
 
939
                        ' thread: %r' % x)
 
940
            except Exception, x:
 
941
                # probably a failed test; unit test thread will log the
 
942
                # failure/error
 
943
                sys.excepthook(*sys.exc_info())
 
944
                warning('Exception from within unit test server thread: %r' % 
 
945
                        x)
 
946
 
 
947
 
 
948
class SocketDelay(object):
 
949
    """A socket decorator to make TCP appear slower.
 
950
 
 
951
    This changes recv, send, and sendall to add a fixed latency to each python
 
952
    call if a new roundtrip is detected. That is, when a recv is called and the
 
953
    flag new_roundtrip is set, latency is charged. Every send and send_all
 
954
    sets this flag.
 
955
 
 
956
    In addition every send, sendall and recv sleeps a bit per character send to
 
957
    simulate bandwidth.
 
958
 
 
959
    Not all methods are implemented, this is deliberate as this class is not a
 
960
    replacement for the builtin sockets layer. fileno is not implemented to
 
961
    prevent the proxy being bypassed. 
 
962
    """
 
963
 
 
964
    simulated_time = 0
 
965
    _proxied_arguments = dict.fromkeys([
 
966
        "close", "getpeername", "getsockname", "getsockopt", "gettimeout",
 
967
        "setblocking", "setsockopt", "settimeout", "shutdown"])
 
968
 
 
969
    def __init__(self, sock, latency, bandwidth=1.0, 
 
970
                 really_sleep=True):
 
971
        """ 
 
972
        :param bandwith: simulated bandwith (MegaBit)
 
973
        :param really_sleep: If set to false, the SocketDelay will just
 
974
        increase a counter, instead of calling time.sleep. This is useful for
 
975
        unittesting the SocketDelay.
 
976
        """
 
977
        self.sock = sock
 
978
        self.latency = latency
 
979
        self.really_sleep = really_sleep
 
980
        self.time_per_byte = 1 / (bandwidth / 8.0 * 1024 * 1024) 
 
981
        self.new_roundtrip = False
 
982
 
 
983
    def sleep(self, s):
 
984
        if self.really_sleep:
 
985
            time.sleep(s)
 
986
        else:
 
987
            SocketDelay.simulated_time += s
 
988
 
 
989
    def __getattr__(self, attr):
 
990
        if attr in SocketDelay._proxied_arguments:
 
991
            return getattr(self.sock, attr)
 
992
        raise AttributeError("'SocketDelay' object has no attribute %r" %
 
993
                             attr)
 
994
 
 
995
    def dup(self):
 
996
        return SocketDelay(self.sock.dup(), self.latency, self.time_per_byte,
 
997
                           self._sleep)
 
998
 
 
999
    def recv(self, *args):
 
1000
        data = self.sock.recv(*args)
 
1001
        if data and self.new_roundtrip:
 
1002
            self.new_roundtrip = False
 
1003
            self.sleep(self.latency)
 
1004
        self.sleep(len(data) * self.time_per_byte)
 
1005
        return data
 
1006
 
 
1007
    def sendall(self, data, flags=0):
 
1008
        if not self.new_roundtrip:
 
1009
            self.new_roundtrip = True
 
1010
            self.sleep(self.latency)
 
1011
        self.sleep(len(data) * self.time_per_byte)
 
1012
        return self.sock.sendall(data, flags)
 
1013
 
 
1014
    def send(self, data, flags=0):
 
1015
        if not self.new_roundtrip:
 
1016
            self.new_roundtrip = True
 
1017
            self.sleep(self.latency)
 
1018
        bytes_sent = self.sock.send(data, flags)
 
1019
        self.sleep(bytes_sent * self.time_per_byte)
 
1020
        return bytes_sent
 
1021
 
 
1022
 
 
1023
class SFTPServer(Server):
 
1024
    """Common code for SFTP server facilities."""
 
1025
 
 
1026
    def __init__(self, server_interface=StubServer):
 
1027
        self._original_vendor = None
 
1028
        self._homedir = None
 
1029
        self._server_homedir = None
 
1030
        self._listener = None
 
1031
        self._root = None
 
1032
        self._vendor = ssh.ParamikoVendor()
 
1033
        self._server_interface = server_interface
 
1034
        # sftp server logs
 
1035
        self.logs = []
 
1036
        self.add_latency = 0
 
1037
 
 
1038
    def _get_sftp_url(self, path):
 
1039
        """Calculate an sftp url to this server for path."""
 
1040
        return 'sftp://foo:bar@localhost:%d/%s' % (self._listener.port, path)
 
1041
 
 
1042
    def log(self, message):
 
1043
        """StubServer uses this to log when a new server is created."""
 
1044
        self.logs.append(message)
 
1045
 
 
1046
    def _run_server_entry(self, sock):
 
1047
        """Entry point for all implementations of _run_server.
 
1048
        
 
1049
        If self.add_latency is > 0.000001 then sock is given a latency adding
 
1050
        decorator.
 
1051
        """
 
1052
        if self.add_latency > 0.000001:
 
1053
            sock = SocketDelay(sock, self.add_latency)
 
1054
        return self._run_server(sock)
 
1055
 
 
1056
    def _run_server(self, s):
 
1057
        ssh_server = paramiko.Transport(s)
 
1058
        key_file = pathjoin(self._homedir, 'test_rsa.key')
 
1059
        f = open(key_file, 'w')
 
1060
        f.write(STUB_SERVER_KEY)
 
1061
        f.close()
 
1062
        host_key = paramiko.RSAKey.from_private_key_file(key_file)
 
1063
        ssh_server.add_server_key(host_key)
 
1064
        server = self._server_interface(self)
 
1065
        ssh_server.set_subsystem_handler('sftp', paramiko.SFTPServer,
 
1066
                                         StubSFTPServer, root=self._root,
 
1067
                                         home=self._server_homedir)
 
1068
        event = threading.Event()
 
1069
        ssh_server.start_server(event, server)
 
1070
        event.wait(5.0)
 
1071
    
 
1072
    def setUp(self, backing_server=None):
 
1073
        # XXX: TODO: make sftpserver back onto backing_server rather than local
 
1074
        # disk.
 
1075
        assert (backing_server is None or
 
1076
                isinstance(backing_server, local.LocalURLServer)), (
 
1077
            "backing_server should not be %r, because this can only serve the "
 
1078
            "local current working directory." % (backing_server,))
 
1079
        self._original_vendor = ssh._ssh_vendor_manager._cached_ssh_vendor
 
1080
        ssh._ssh_vendor_manager._cached_ssh_vendor = self._vendor
 
1081
        if sys.platform == 'win32':
 
1082
            # Win32 needs to use the UNICODE api
 
1083
            self._homedir = getcwd()
 
1084
        else:
 
1085
            # But Linux SFTP servers should just deal in bytestreams
 
1086
            self._homedir = os.getcwd()
 
1087
        if self._server_homedir is None:
 
1088
            self._server_homedir = self._homedir
 
1089
        self._root = '/'
 
1090
        if sys.platform == 'win32':
 
1091
            self._root = ''
 
1092
        self._listener = SocketListener(self._run_server_entry)
 
1093
        self._listener.setDaemon(True)
 
1094
        self._listener.start()
 
1095
 
 
1096
    def tearDown(self):
 
1097
        """See bzrlib.transport.Server.tearDown."""
 
1098
        self._listener.stop()
 
1099
        ssh._ssh_vendor_manager._cached_ssh_vendor = self._original_vendor
 
1100
 
 
1101
    def get_bogus_url(self):
 
1102
        """See bzrlib.transport.Server.get_bogus_url."""
 
1103
        # this is chosen to try to prevent trouble with proxies, wierd dns, etc
 
1104
        # we bind a random socket, so that we get a guaranteed unused port
 
1105
        # we just never listen on that port
 
1106
        s = socket.socket()
 
1107
        s.bind(('localhost', 0))
 
1108
        return 'sftp://%s:%s/' % s.getsockname()
 
1109
 
 
1110
 
 
1111
class SFTPFullAbsoluteServer(SFTPServer):
 
1112
    """A test server for sftp transports, using absolute urls and ssh."""
 
1113
 
 
1114
    def get_url(self):
 
1115
        """See bzrlib.transport.Server.get_url."""
 
1116
        homedir = self._homedir
 
1117
        if sys.platform != 'win32':
 
1118
            # Remove the initial '/' on all platforms but win32
 
1119
            homedir = homedir[1:]
 
1120
        return self._get_sftp_url(urlutils.escape(homedir))
 
1121
 
 
1122
 
 
1123
class SFTPServerWithoutSSH(SFTPServer):
 
1124
    """An SFTP server that uses a simple TCP socket pair rather than SSH."""
 
1125
 
 
1126
    def __init__(self):
 
1127
        super(SFTPServerWithoutSSH, self).__init__()
 
1128
        self._vendor = ssh.LoopbackVendor()
 
1129
 
 
1130
    def _run_server(self, sock):
 
1131
        # Re-import these as locals, so that they're still accessible during
 
1132
        # interpreter shutdown (when all module globals get set to None, leading
 
1133
        # to confusing errors like "'NoneType' object has no attribute 'error'".
 
1134
        class FakeChannel(object):
 
1135
            def get_transport(self):
 
1136
                return self
 
1137
            def get_log_channel(self):
 
1138
                return 'paramiko'
 
1139
            def get_name(self):
 
1140
                return '1'
 
1141
            def get_hexdump(self):
 
1142
                return False
 
1143
            def close(self):
 
1144
                pass
 
1145
 
 
1146
        server = paramiko.SFTPServer(FakeChannel(), 'sftp', StubServer(self), StubSFTPServer,
 
1147
                                     root=self._root, home=self._server_homedir)
 
1148
        try:
 
1149
            server.start_subsystem('sftp', None, sock)
 
1150
        except socket.error, e:
 
1151
            if (len(e.args) > 0) and (e.args[0] == errno.EPIPE):
 
1152
                # it's okay for the client to disconnect abruptly
 
1153
                # (bug in paramiko 1.6: it should absorb this exception)
 
1154
                pass
 
1155
            else:
 
1156
                raise
 
1157
        except Exception, e:
 
1158
            # This typically seems to happen during interpreter shutdown, so
 
1159
            # most of the useful ways to report this error are won't work.
 
1160
            # Writing the exception type, and then the text of the exception,
 
1161
            # seems to be the best we can do.
 
1162
            import sys
 
1163
            sys.stderr.write('\nEXCEPTION %r: ' % (e.__class__,))
 
1164
            sys.stderr.write('%s\n\n' % (e,))
 
1165
        server.finish_subsystem()
 
1166
 
 
1167
 
 
1168
class SFTPAbsoluteServer(SFTPServerWithoutSSH):
 
1169
    """A test server for sftp transports, using absolute urls."""
 
1170
 
 
1171
    def get_url(self):
 
1172
        """See bzrlib.transport.Server.get_url."""
 
1173
        homedir = self._homedir
 
1174
        if sys.platform != 'win32':
 
1175
            # Remove the initial '/' on all platforms but win32
 
1176
            homedir = homedir[1:]
 
1177
        return self._get_sftp_url(urlutils.escape(homedir))
 
1178
 
 
1179
 
 
1180
class SFTPHomeDirServer(SFTPServerWithoutSSH):
 
1181
    """A test server for sftp transports, using homedir relative urls."""
 
1182
 
 
1183
    def get_url(self):
 
1184
        """See bzrlib.transport.Server.get_url."""
 
1185
        return self._get_sftp_url("~/")
 
1186
 
 
1187
 
 
1188
class SFTPSiblingAbsoluteServer(SFTPAbsoluteServer):
 
1189
    """A test server for sftp transports where only absolute paths will work.
 
1190
 
 
1191
    It does this by serving from a deeply-nested directory that doesn't exist.
 
1192
    """
 
1193
 
 
1194
    def setUp(self, backing_server=None):
 
1195
        self._server_homedir = '/dev/noone/runs/tests/here'
 
1196
        super(SFTPSiblingAbsoluteServer, self).setUp(backing_server)
 
1197
 
 
1198
 
 
1199
def _sftp_connect(host, port, username, password):
 
1200
    """Connect to the remote sftp server.
 
1201
 
 
1202
    :raises: a TransportError 'could not connect'.
 
1203
 
 
1204
    :returns: an paramiko.sftp_client.SFTPClient
 
1205
 
 
1206
    TODO: Raise a more reasonable ConnectionFailed exception
 
1207
    """
 
1208
    idx = (host, port, username)
 
1209
    try:
 
1210
        return _connected_hosts[idx]
 
1211
    except KeyError:
 
1212
        pass
 
1213
    
 
1214
    sftp = _sftp_connect_uncached(host, port, username, password)
 
1215
    _connected_hosts[idx] = sftp
 
1216
    return sftp
 
1217
 
 
1218
def _sftp_connect_uncached(host, port, username, password):
 
1219
    vendor = ssh._get_ssh_vendor()
 
1220
    sftp = vendor.connect_sftp(username, password, host, port)
 
1221
    return sftp
 
1222
 
 
1223
 
 
1224
def get_test_permutations():
 
1225
    """Return the permutations to be used in testing."""
 
1226
    return [(SFTPTransport, SFTPAbsoluteServer),
 
1227
            (SFTPTransport, SFTPHomeDirServer),
 
1228
            (SFTPTransport, SFTPSiblingAbsoluteServer),
 
1229
            ]