← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/transport/sftp.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2007-12-18 23:41:30 UTC
  • mfrom: (3099.3.7 graph_optimization)
  • Revision ID: pqm@pqm.ubuntu.com-20071218234130-061grgxsaf1g7bao
(jam) Implement ParentProviders.get_parent_map() and deprecate
        get_parents()

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/transport/sftp.py
1
 
# Copyright (C) 2005-2011, 2016 Canonical Ltd
 
1
# Copyright (C) 2005 Robey Pointer <robey@lag.net>
 
2
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
2
3
#
3
4
# This program is free software; you can redistribute it and/or modify
4
5
# it under the terms of the GNU General Public License as published by
12
13
#
13
14
# You should have received a copy of the GNU General Public License
14
15
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
17
 
17
18
"""Implementation of Transport over SFTP, using paramiko."""
18
19
 
19
 
from __future__ import absolute_import
20
 
 
21
20
# TODO: Remove the transport-based lock_read and lock_write methods.  They'll
22
21
# then raise TransportNotPossible, which will break remote access to any
23
22
# formats which rely on OS-level locks.  That should be fine as those formats
25
24
# suite.  Those formats all date back to 0.7; so we should be able to remove
26
25
# these methods when we officially drop support for those formats.
27
26
 
28
 
import bisect
29
27
import errno
30
 
import itertools
31
28
import os
32
29
import random
 
30
import select
 
31
import socket
33
32
import stat
34
33
import sys
35
34
import time
 
35
import urllib
 
36
import urlparse
36
37
import warnings
37
38
 
38
39
from bzrlib import (
39
 
    config,
40
 
    debug,
41
40
    errors,
42
41
    urlutils,
43
42
    )
44
43
from bzrlib.errors import (FileExists,
45
 
                           NoSuchFile,
 
44
                           NoSuchFile, PathNotChild,
46
45
                           TransportError,
47
46
                           LockError,
48
47
                           PathError,
49
48
                           ParamikoNotPresent,
50
49
                           )
51
 
from bzrlib.osutils import fancy_rename
 
50
from bzrlib.osutils import pathjoin, fancy_rename, getcwd
 
51
from bzrlib.symbol_versioning import (
 
52
        deprecated_function,
 
53
        zero_ninety,
 
54
        )
52
55
from bzrlib.trace import mutter, warning
53
56
from bzrlib.transport import (
54
57
    FileFileStream,
55
58
    _file_streams,
 
59
    local,
 
60
    Server,
56
61
    ssh,
57
62
    ConnectedTransport,
58
63
    )
77
82
else:
78
83
    from paramiko.sftp import (SFTP_FLAG_WRITE, SFTP_FLAG_CREATE,
79
84
                               SFTP_FLAG_EXCL, SFTP_FLAG_TRUNC,
80
 
                               SFTP_OK, CMD_HANDLE, CMD_OPEN)
 
85
                               CMD_HANDLE, CMD_OPEN)
81
86
    from paramiko.sftp_attr import SFTPAttributes
82
87
    from paramiko.sftp_file import SFTPFile
83
88
 
87
92
_default_do_prefetch = (_paramiko_version >= (1, 5, 5))
88
93
 
89
94
 
 
95
@deprecated_function(zero_ninety)
 
96
def clear_connection_cache():
 
97
    """Remove all hosts from the SFTP connection cache.
 
98
 
 
99
    Primarily useful for test cases wanting to force garbage collection.
 
100
    We don't have a global connection cache anymore.
 
101
    """
 
102
 
90
103
class SFTPLock(object):
91
104
    """This fakes a lock in a remote location.
92
 
 
 
105
    
93
106
    A present lock is indicated just by the existence of a file.  This
94
 
    doesn't work well on all transports and they are only used in
 
107
    doesn't work well on all transports and they are only used in 
95
108
    deprecated storage formats.
96
109
    """
97
 
 
 
110
    
98
111
    __slots__ = ['path', 'lock_path', 'lock_file', 'transport']
99
112
 
100
113
    def __init__(self, path, transport):
 
114
        assert isinstance(transport, SFTPTransport)
 
115
 
101
116
        self.lock_file = None
102
117
        self.path = path
103
118
        self.lock_path = path + '.write-lock'
109
124
        except FileExists:
110
125
            raise LockError('File %r already locked' % (self.path,))
111
126
 
 
127
    def __del__(self):
 
128
        """Should this warn, or actually try to cleanup?"""
 
129
        if self.lock_file:
 
130
            warning("SFTPLock %r not explicitly unlocked" % (self.path,))
 
131
            self.unlock()
 
132
 
112
133
    def unlock(self):
113
134
        if not self.lock_file:
114
135
            return
121
142
            pass
122
143
 
123
144
 
124
 
class _SFTPReadvHelper(object):
125
 
    """A class to help with managing the state of a readv request."""
126
 
 
127
 
    # See _get_requests for an explanation.
128
 
    _max_request_size = 32768
129
 
 
130
 
    def __init__(self, original_offsets, relpath, _report_activity):
131
 
        """Create a new readv helper.
132
 
 
133
 
        :param original_offsets: The original requests given by the caller of
134
 
            readv()
135
 
        :param relpath: The name of the file (if known)
136
 
        :param _report_activity: A Transport._report_activity bound method,
137
 
            to be called as data arrives.
138
 
        """
139
 
        self.original_offsets = list(original_offsets)
140
 
        self.relpath = relpath
141
 
        self._report_activity = _report_activity
142
 
 
143
 
    def _get_requests(self):
144
 
        """Break up the offsets into individual requests over sftp.
145
 
 
146
 
        The SFTP spec only requires implementers to support 32kB requests. We
147
 
        could try something larger (openssh supports 64kB), but then we have to
148
 
        handle requests that fail.
149
 
        So instead, we just break up our maximum chunks into 32kB chunks, and
150
 
        asyncronously requests them.
151
 
        Newer versions of paramiko would do the chunking for us, but we want to
152
 
        start processing results right away, so we do it ourselves.
153
 
        """
154
 
        # TODO: Because we issue async requests, we don't 'fudge' any extra
155
 
        #       data.  I'm not 100% sure that is the best choice.
156
 
 
157
 
        # The first thing we do, is to collapse the individual requests as much
158
 
        # as possible, so we don't issues requests <32kB
159
 
        sorted_offsets = sorted(self.original_offsets)
160
 
        coalesced = list(ConnectedTransport._coalesce_offsets(sorted_offsets,
161
 
                                                        limit=0, fudge_factor=0))
162
 
        requests = []
163
 
        for c_offset in coalesced:
164
 
            start = c_offset.start
165
 
            size = c_offset.length
166
 
 
167
 
            # Break this up into 32kB requests
168
 
            while size > 0:
169
 
                next_size = min(size, self._max_request_size)
170
 
                requests.append((start, next_size))
171
 
                size -= next_size
172
 
                start += next_size
173
 
        if 'sftp' in debug.debug_flags:
174
 
            mutter('SFTP.readv(%s) %s offsets => %s coalesced => %s requests',
175
 
                self.relpath, len(sorted_offsets), len(coalesced),
176
 
                len(requests))
177
 
        return requests
178
 
 
179
 
    def request_and_yield_offsets(self, fp):
180
 
        """Request the data from the remote machine, yielding the results.
181
 
 
182
 
        :param fp: A Paramiko SFTPFile object that supports readv.
183
 
        :return: Yield the data requested by the original readv caller, one by
184
 
            one.
185
 
        """
186
 
        requests = self._get_requests()
187
 
        offset_iter = iter(self.original_offsets)
188
 
        cur_offset, cur_size = offset_iter.next()
189
 
        # paramiko .readv() yields strings that are in the order of the requests
190
 
        # So we track the current request to know where the next data is
191
 
        # being returned from.
192
 
        input_start = None
193
 
        last_end = None
194
 
        buffered_data = []
195
 
        buffered_len = 0
196
 
 
197
 
        # This is used to buffer chunks which we couldn't process yet
198
 
        # It is (start, end, data) tuples.
199
 
        data_chunks = []
200
 
        # Create an 'unlimited' data stream, so we stop based on requests,
201
 
        # rather than just because the data stream ended. This lets us detect
202
 
        # short readv.
203
 
        data_stream = itertools.chain(fp.readv(requests),
204
 
                                      itertools.repeat(None))
205
 
        for (start, length), data in itertools.izip(requests, data_stream):
206
 
            if data is None:
207
 
                if cur_coalesced is not None:
208
 
                    raise errors.ShortReadvError(self.relpath,
209
 
                        start, length, len(data))
210
 
            if len(data) != length:
211
 
                raise errors.ShortReadvError(self.relpath,
212
 
                    start, length, len(data))
213
 
            self._report_activity(length, 'read')
214
 
            if last_end is None:
215
 
                # This is the first request, just buffer it
216
 
                buffered_data = [data]
217
 
                buffered_len = length
218
 
                input_start = start
219
 
            elif start == last_end:
220
 
                # The data we are reading fits neatly on the previous
221
 
                # buffer, so this is all part of a larger coalesced range.
222
 
                buffered_data.append(data)
223
 
                buffered_len += length
224
 
            else:
225
 
                # We have an 'interrupt' in the data stream. So we know we are
226
 
                # at a request boundary.
227
 
                if buffered_len > 0:
228
 
                    # We haven't consumed the buffer so far, so put it into
229
 
                    # data_chunks, and continue.
230
 
                    buffered = ''.join(buffered_data)
231
 
                    data_chunks.append((input_start, buffered))
232
 
                input_start = start
233
 
                buffered_data = [data]
234
 
                buffered_len = length
235
 
            last_end = start + length
236
 
            if input_start == cur_offset and cur_size <= buffered_len:
237
 
                # Simplify the next steps a bit by transforming buffered_data
238
 
                # into a single string. We also have the nice property that
239
 
                # when there is only one string ''.join([x]) == x, so there is
240
 
                # no data copying.
241
 
                buffered = ''.join(buffered_data)
242
 
                # Clean out buffered data so that we keep memory
243
 
                # consumption low
244
 
                del buffered_data[:]
245
 
                buffered_offset = 0
246
 
                # TODO: We *could* also consider the case where cur_offset is in
247
 
                #       in the buffered range, even though it doesn't *start*
248
 
                #       the buffered range. But for packs we pretty much always
249
 
                #       read in order, so you won't get any extra data in the
250
 
                #       middle.
251
 
                while (input_start == cur_offset
252
 
                       and (buffered_offset + cur_size) <= buffered_len):
253
 
                    # We've buffered enough data to process this request, spit it
254
 
                    # out
255
 
                    cur_data = buffered[buffered_offset:buffered_offset + cur_size]
256
 
                    # move the direct pointer into our buffered data
257
 
                    buffered_offset += cur_size
258
 
                    # Move the start-of-buffer pointer
259
 
                    input_start += cur_size
260
 
                    # Yield the requested data
261
 
                    yield cur_offset, cur_data
262
 
                    cur_offset, cur_size = offset_iter.next()
263
 
                # at this point, we've consumed as much of buffered as we can,
264
 
                # so break off the portion that we consumed
265
 
                if buffered_offset == len(buffered_data):
266
 
                    # No tail to leave behind
267
 
                    buffered_data = []
268
 
                    buffered_len = 0
269
 
                else:
270
 
                    buffered = buffered[buffered_offset:]
271
 
                    buffered_data = [buffered]
272
 
                    buffered_len = len(buffered)
273
 
        # now that the data stream is done, close the handle
274
 
        fp.close()
275
 
        if buffered_len:
276
 
            buffered = ''.join(buffered_data)
277
 
            del buffered_data[:]
278
 
            data_chunks.append((input_start, buffered))
279
 
        if data_chunks:
280
 
            if 'sftp' in debug.debug_flags:
281
 
                mutter('SFTP readv left with %d out-of-order bytes',
282
 
                    sum(map(lambda x: len(x[1]), data_chunks)))
283
 
            # We've processed all the readv data, at this point, anything we
284
 
            # couldn't process is in data_chunks. This doesn't happen often, so
285
 
            # this code path isn't optimized
286
 
            # We use an interesting process for data_chunks
287
 
            # Specifically if we have "bisect_left([(start, len, entries)],
288
 
            #                                       (qstart,)])
289
 
            # If start == qstart, then we get the specific node. Otherwise we
290
 
            # get the previous node
291
 
            while True:
292
 
                idx = bisect.bisect_left(data_chunks, (cur_offset,))
293
 
                if idx < len(data_chunks) and data_chunks[idx][0] == cur_offset:
294
 
                    # The data starts here
295
 
                    data = data_chunks[idx][1][:cur_size]
296
 
                elif idx > 0:
297
 
                    # The data is in a portion of a previous page
298
 
                    idx -= 1
299
 
                    sub_offset = cur_offset - data_chunks[idx][0]
300
 
                    data = data_chunks[idx][1]
301
 
                    data = data[sub_offset:sub_offset + cur_size]
302
 
                else:
303
 
                    # We are missing the page where the data should be found,
304
 
                    # something is wrong
305
 
                    data = ''
306
 
                if len(data) != cur_size:
307
 
                    raise AssertionError('We must have miscalulated.'
308
 
                        ' We expected %d bytes, but only found %d'
309
 
                        % (cur_size, len(data)))
310
 
                yield cur_offset, data
311
 
                cur_offset, cur_size = offset_iter.next()
312
 
 
313
 
 
314
145
class SFTPTransport(ConnectedTransport):
315
146
    """Transport implementation for SFTP access."""
316
147
 
332
163
    # up the request itself, rather than us having to worry about it
333
164
    _max_request_size = 32768
334
165
 
 
166
    def __init__(self, base, _from_transport=None):
 
167
        assert base.startswith('sftp://')
 
168
        super(SFTPTransport, self).__init__(base,
 
169
                                            _from_transport=_from_transport)
 
170
 
335
171
    def _remote_path(self, relpath):
336
172
        """Return the path to be passed along the sftp protocol for relpath.
337
 
 
 
173
        
338
174
        :param relpath: is a urlencoded string.
339
175
        """
340
 
        remote_path = self._parsed_url.clone(relpath).path
 
176
        relative = urlutils.unescape(relpath).encode('utf-8')
 
177
        remote_path = self._combine_paths(self._path, relative)
341
178
        # the initial slash should be removed from the path, and treated as a
342
179
        # homedir relative path (the path begins with a double slash if it is
343
180
        # absolute).  see draft-ietf-secsh-scp-sftp-ssh-uri-03.txt
362
199
        in base url at transport creation time.
363
200
        """
364
201
        if credentials is None:
365
 
            password = self._parsed_url.password
 
202
            password = self._password
366
203
        else:
367
204
            password = credentials
368
205
 
369
206
        vendor = ssh._get_ssh_vendor()
370
 
        user = self._parsed_url.user
371
 
        if user is None:
372
 
            auth = config.AuthenticationConfig()
373
 
            user = auth.get_user('ssh', self._parsed_url.host,
374
 
                self._parsed_url.port)
375
 
        connection = vendor.connect_sftp(self._parsed_url.user, password,
376
 
            self._parsed_url.host, self._parsed_url.port)
377
 
        return connection, (user, password)
378
 
 
379
 
    def disconnect(self):
380
 
        connection = self._get_connection()
381
 
        if connection is not None:
382
 
            connection.close()
 
207
        connection = vendor.connect_sftp(self._user, password,
 
208
                                         self._host, self._port)
 
209
        return connection, password
383
210
 
384
211
    def _get_sftp(self):
385
212
        """Ensures that a connection is established"""
396
223
        """
397
224
        try:
398
225
            self._get_sftp().stat(self._remote_path(relpath))
399
 
            # stat result is about 20 bytes, let's say
400
 
            self._report_activity(20, 'read')
401
226
            return True
402
227
        except IOError:
403
228
            return False
404
229
 
405
230
    def get(self, relpath):
406
 
        """Get the file at the given relative path.
 
231
        """
 
232
        Get the file at the given relative path.
407
233
 
408
234
        :param relpath: The relative path to the file
409
235
        """
417
243
            self._translate_io_exception(e, path, ': error retrieving',
418
244
                failure_exc=errors.ReadError)
419
245
 
420
 
    def get_bytes(self, relpath):
421
 
        # reimplement this here so that we can report how many bytes came back
422
 
        f = self.get(relpath)
423
 
        try:
424
 
            bytes = f.read()
425
 
            self._report_activity(len(bytes), 'read')
426
 
            return bytes
427
 
        finally:
428
 
            f.close()
429
 
 
430
246
    def _readv(self, relpath, offsets):
431
247
        """See Transport.readv()"""
432
248
        # We overload the default readv() because we want to use a file
441
257
            readv = getattr(fp, 'readv', None)
442
258
            if readv:
443
259
                return self._sftp_readv(fp, offsets, relpath)
444
 
            if 'sftp' in debug.debug_flags:
445
 
                mutter('seek and read %s offsets', len(offsets))
 
260
            mutter('seek and read %s offsets', len(offsets))
446
261
            return self._seek_and_read(fp, offsets, relpath)
447
262
        except (IOError, paramiko.SSHException), e:
448
263
            self._translate_io_exception(e, path, ': error retrieving')
455
270
        """
456
271
        return 64 * 1024
457
272
 
458
 
    def _sftp_readv(self, fp, offsets, relpath):
 
273
    def _sftp_readv(self, fp, offsets, relpath='<unknown>'):
459
274
        """Use the readv() member of fp to do async readv.
460
275
 
461
 
        Then read them using paramiko.readv(). paramiko.readv()
 
276
        And then read them using paramiko.readv(). paramiko.readv()
462
277
        does not support ranges > 64K, so it caps the request size, and
463
 
        just reads until it gets all the stuff it wants.
 
278
        just reads until it gets all the stuff it wants
464
279
        """
465
 
        helper = _SFTPReadvHelper(offsets, relpath, self._report_activity)
466
 
        return helper.request_and_yield_offsets(fp)
 
280
        offsets = list(offsets)
 
281
        sorted_offsets = sorted(offsets)
 
282
 
 
283
        # The algorithm works as follows:
 
284
        # 1) Coalesce nearby reads into a single chunk
 
285
        #    This generates a list of combined regions, the total size
 
286
        #    and the size of the sub regions. This coalescing step is limited
 
287
        #    in the number of nearby chunks to combine, and is allowed to
 
288
        #    skip small breaks in the requests. Limiting it makes sure that
 
289
        #    we can start yielding some data earlier, and skipping means we
 
290
        #    make fewer requests. (Beneficial even when using async)
 
291
        # 2) Break up this combined regions into chunks that are smaller
 
292
        #    than 64KiB. Technically the limit is 65536, but we are a
 
293
        #    little bit conservative. This is because sftp has a maximum
 
294
        #    return chunk size of 64KiB (max size of an unsigned short)
 
295
        # 3) Issue a readv() to paramiko to create an async request for
 
296
        #    all of this data
 
297
        # 4) Read in the data as it comes back, until we've read one
 
298
        #    continuous section as determined in step 1
 
299
        # 5) Break up the full sections into hunks for the original requested
 
300
        #    offsets. And put them in a cache
 
301
        # 6) Check if the next request is in the cache, and if it is, remove
 
302
        #    it from the cache, and yield its data. Continue until no more
 
303
        #    entries are in the cache.
 
304
        # 7) loop back to step 4 until all data has been read
 
305
        #
 
306
        # TODO: jam 20060725 This could be optimized one step further, by
 
307
        #       attempting to yield whatever data we have read, even before
 
308
        #       the first coallesced section has been fully processed.
 
309
 
 
310
        # When coalescing for use with readv(), we don't really need to
 
311
        # use any fudge factor, because the requests are made asynchronously
 
312
        coalesced = list(self._coalesce_offsets(sorted_offsets,
 
313
                               limit=self._max_readv_combine,
 
314
                               fudge_factor=0,
 
315
                               ))
 
316
        requests = []
 
317
        for c_offset in coalesced:
 
318
            start = c_offset.start
 
319
            size = c_offset.length
 
320
 
 
321
            # We need to break this up into multiple requests
 
322
            while size > 0:
 
323
                next_size = min(size, self._max_request_size)
 
324
                requests.append((start, next_size))
 
325
                size -= next_size
 
326
                start += next_size
 
327
 
 
328
        mutter('SFTP.readv() %s offsets => %s coalesced => %s requests',
 
329
                len(offsets), len(coalesced), len(requests))
 
330
 
 
331
        # Queue the current read until we have read the full coalesced section
 
332
        cur_data = []
 
333
        cur_data_len = 0
 
334
        cur_coalesced_stack = iter(coalesced)
 
335
        cur_coalesced = cur_coalesced_stack.next()
 
336
 
 
337
        # Cache the results, but only until they have been fulfilled
 
338
        data_map = {}
 
339
        # turn the list of offsets into a stack
 
340
        offset_stack = iter(offsets)
 
341
        cur_offset_and_size = offset_stack.next()
 
342
 
 
343
        for data in fp.readv(requests):
 
344
            cur_data += data
 
345
            cur_data_len += len(data)
 
346
 
 
347
            if cur_data_len < cur_coalesced.length:
 
348
                continue
 
349
            assert cur_data_len == cur_coalesced.length, \
 
350
                "Somehow we read too much: %s != %s" % (cur_data_len,
 
351
                                                        cur_coalesced.length)
 
352
            all_data = ''.join(cur_data)
 
353
            cur_data = []
 
354
            cur_data_len = 0
 
355
 
 
356
            for suboffset, subsize in cur_coalesced.ranges:
 
357
                key = (cur_coalesced.start+suboffset, subsize)
 
358
                data_map[key] = all_data[suboffset:suboffset+subsize]
 
359
 
 
360
            # Now that we've read some data, see if we can yield anything back
 
361
            while cur_offset_and_size in data_map:
 
362
                this_data = data_map.pop(cur_offset_and_size)
 
363
                yield cur_offset_and_size[0], this_data
 
364
                cur_offset_and_size = offset_stack.next()
 
365
 
 
366
            # We read a coalesced entry, so mark it as done
 
367
            cur_coalesced = None
 
368
            # Now that we've read all of the data for this coalesced section
 
369
            # on to the next
 
370
            cur_coalesced = cur_coalesced_stack.next()
 
371
 
 
372
        if cur_coalesced is not None:
 
373
            raise errors.ShortReadvError(relpath, cur_coalesced.start,
 
374
                cur_coalesced.length, len(data))
467
375
 
468
376
    def put_file(self, relpath, f, mode=None):
469
377
        """
494
402
            #      sticky bit. So it is probably best to stop chmodding, and
495
403
            #      just tell users that they need to set the umask correctly.
496
404
            #      The attr.st_mode = mode, in _sftp_open_exclusive
497
 
            #      will handle when the user wants the final mode to be more
498
 
            #      restrictive. And then we avoid a round trip. Unless
 
405
            #      will handle when the user wants the final mode to be more 
 
406
            #      restrictive. And then we avoid a round trip. Unless 
499
407
            #      paramiko decides to expose an async chmod()
500
408
 
501
409
            # This is designed to chmod() right before we close.
502
 
            # Because we set_pipelined() earlier, theoretically we might
 
410
            # Because we set_pipelined() earlier, theoretically we might 
503
411
            # avoid the round trip for fout.close()
504
412
            if mode is not None:
505
413
                self._get_sftp().chmod(tmp_abspath, mode)
547
455
                                                 ': unable to open')
548
456
 
549
457
                # This is designed to chmod() right before we close.
550
 
                # Because we set_pipelined() earlier, theoretically we might
 
458
                # Because we set_pipelined() earlier, theoretically we might 
551
459
                # avoid the round trip for fout.close()
552
460
                if mode is not None:
553
461
                    self._get_sftp().chmod(abspath, mode)
593
501
                                    create_parent_dir=create_parent_dir,
594
502
                                    dir_mode=dir_mode)
595
503
 
596
 
    def put_bytes_non_atomic(self, relpath, raw_bytes, mode=None,
 
504
    def put_bytes_non_atomic(self, relpath, bytes, mode=None,
597
505
                             create_parent_dir=False,
598
506
                             dir_mode=None):
599
 
        if not isinstance(raw_bytes, str):
600
 
            raise TypeError(
601
 
                'raw_bytes must be a plain string, not %s' % type(raw_bytes))
602
 
 
603
507
        def writer(fout):
604
 
            fout.write(raw_bytes)
 
508
            fout.write(bytes)
605
509
        self._put_non_atomic_helper(relpath, writer, mode=mode,
606
510
                                    create_parent_dir=create_parent_dir,
607
511
                                    dir_mode=dir_mode)
608
512
 
609
513
    def iter_files_recursive(self):
610
514
        """Walk the relative paths of all files in this transport."""
611
 
        # progress is handled by list_dir
612
515
        queue = list(self.list_dir('.'))
613
516
        while queue:
614
517
            relpath = queue.pop(0)
625
528
        else:
626
529
            local_mode = mode
627
530
        try:
628
 
            self._report_activity(len(abspath), 'write')
629
531
            self._get_sftp().mkdir(abspath, local_mode)
630
 
            self._report_activity(1, 'read')
631
532
            if mode is not None:
632
 
                # chmod a dir through sftp will erase any sgid bit set
633
 
                # on the server side.  So, if the bit mode are already
634
 
                # set, avoid the chmod.  If the mode is not fine but
635
 
                # the sgid bit is set, report a warning to the user
636
 
                # with the umask fix.
637
 
                stat = self._get_sftp().lstat(abspath)
638
 
                mode = mode & 0777 # can't set special bits anyway
639
 
                if mode != stat.st_mode & 0777:
640
 
                    if stat.st_mode & 06000:
641
 
                        warning('About to chmod %s over sftp, which will result'
642
 
                                ' in its suid or sgid bits being cleared.  If'
643
 
                                ' you want to preserve those bits, change your '
644
 
                                ' environment on the server to use umask 0%03o.'
645
 
                                % (abspath, 0777 - mode))
646
 
                    self._get_sftp().chmod(abspath, mode=mode)
 
533
                self._get_sftp().chmod(abspath, mode=mode)
647
534
        except (paramiko.SSHException, IOError), e:
648
535
            self._translate_io_exception(e, abspath, ': unable to mkdir',
649
536
                failure_exc=FileExists)
655
542
    def open_write_stream(self, relpath, mode=None):
656
543
        """See Transport.open_write_stream."""
657
544
        # initialise the file to zero-length
658
 
        # this is three round trips, but we don't use this
659
 
        # api more than once per write_group at the moment so
 
545
        # this is three round trips, but we don't use this 
 
546
        # api more than once per write_group at the moment so 
660
547
        # it is a tolerable overhead. Better would be to truncate
661
548
        # the file after opening. RBC 20070805
662
549
        self.put_bytes_non_atomic(relpath, "", mode)
685
572
        :param failure_exc: Paramiko has the super fun ability to raise completely
686
573
                           opaque errors that just set "e.args = ('Failure',)" with
687
574
                           no more information.
688
 
                           If this parameter is set, it defines the exception
 
575
                           If this parameter is set, it defines the exception 
689
576
                           to raise in these cases.
690
577
        """
691
578
        # paramiko seems to generate detailless errors.
694
581
            if (e.args == ('No such file or directory',) or
695
582
                e.args == ('No such file',)):
696
583
                raise NoSuchFile(path, str(e) + more_info)
697
 
            if (e.args == ('mkdir failed',) or
698
 
                e.args[0].startswith('syserr: File exists')):
 
584
            if (e.args == ('mkdir failed',)):
699
585
                raise FileExists(path, str(e) + more_info)
700
586
            # strange but true, for the paramiko server.
701
587
            if (e.args == ('Failure',)):
702
588
                raise failure_exc(path, str(e) + more_info)
703
 
            # Can be something like args = ('Directory not empty:
704
 
            # '/srv/bazaar.launchpad.net/blah...: '
705
 
            # [Errno 39] Directory not empty',)
706
 
            if (e.args[0].startswith('Directory not empty: ')
707
 
                or getattr(e, 'errno', None) == errno.ENOTEMPTY):
708
 
                raise errors.DirectoryNotEmpty(path, str(e))
709
 
            if e.args == ('Operation unsupported',):
710
 
                raise errors.TransportNotPossible()
711
589
            mutter('Raising exception with args %s', e.args)
712
590
        if getattr(e, 'errno', None) is not None:
713
591
            mutter('Raising exception with errno %s', e.errno)
740
618
 
741
619
    def _rename_and_overwrite(self, abs_from, abs_to):
742
620
        """Do a fancy rename on the remote server.
743
 
 
 
621
        
744
622
        Using the implementation provided by osutils.
745
623
        """
746
624
        try:
765
643
            self._get_sftp().remove(path)
766
644
        except (IOError, paramiko.SSHException), e:
767
645
            self._translate_io_exception(e, path, ': unable to delete')
768
 
 
 
646
            
769
647
    def external_url(self):
770
648
        """See bzrlib.transport.Transport.external_url."""
771
649
        # the external path for SFTP is the base
786
664
        path = self._remote_path(relpath)
787
665
        try:
788
666
            entries = self._get_sftp().listdir(path)
789
 
            self._report_activity(sum(map(len, entries)), 'read')
790
667
        except (IOError, paramiko.SSHException), e:
791
668
            self._translate_io_exception(e, path, ': failed to list_dir')
792
669
        return [urlutils.escape(entry) for entry in entries]
803
680
        """Return the stat information for a file."""
804
681
        path = self._remote_path(relpath)
805
682
        try:
806
 
            return self._get_sftp().lstat(path)
 
683
            return self._get_sftp().stat(path)
807
684
        except (IOError, paramiko.SSHException), e:
808
685
            self._translate_io_exception(e, path, ': unable to stat')
809
686
 
810
 
    def readlink(self, relpath):
811
 
        """See Transport.readlink."""
812
 
        path = self._remote_path(relpath)
813
 
        try:
814
 
            return self._get_sftp().readlink(path)
815
 
        except (IOError, paramiko.SSHException), e:
816
 
            self._translate_io_exception(e, path, ': unable to readlink')
817
 
 
818
 
    def symlink(self, source, link_name):
819
 
        """See Transport.symlink."""
820
 
        try:
821
 
            conn = self._get_sftp()
822
 
            sftp_retval = conn.symlink(source, link_name)
823
 
            if SFTP_OK != sftp_retval:
824
 
                raise TransportError(
825
 
                    '%r: unable to create symlink to %r' % (link_name, source),
826
 
                    sftp_retval
827
 
                )
828
 
        except (IOError, paramiko.SSHException), e:
829
 
            self._translate_io_exception(e, link_name,
830
 
                                         ': unable to create symlink to %r' % (source))
831
 
 
832
687
    def lock_read(self, relpath):
833
688
        """
834
689
        Lock the given file for shared (read) access.
871
726
        """
872
727
        # TODO: jam 20060816 Paramiko >= 1.6.2 (probably earlier) supports
873
728
        #       using the 'x' flag to indicate SFTP_FLAG_EXCL.
874
 
        #       However, there is no way to set the permission mode at open
 
729
        #       However, there is no way to set the permission mode at open 
875
730
        #       time using the sftp_client.file() functionality.
876
731
        path = self._get_sftp()._adjust_cwd(abspath)
877
732
        # mutter('sftp abspath %s => %s', abspath, path)
878
733
        attr = SFTPAttributes()
879
734
        if mode is not None:
880
735
            attr.st_mode = mode
881
 
        omode = (SFTP_FLAG_WRITE | SFTP_FLAG_CREATE
 
736
        omode = (SFTP_FLAG_WRITE | SFTP_FLAG_CREATE 
882
737
                | SFTP_FLAG_TRUNC | SFTP_FLAG_EXCL)
883
738
        try:
884
739
            t, msg = self._get_sftp()._request(CMD_OPEN, path, omode, attr)
897
752
        else:
898
753
            return True
899
754
 
 
755
# ------------- server test implementation --------------
 
756
import threading
 
757
 
 
758
from bzrlib.tests.stub_sftp import StubServer, StubSFTPServer
 
759
 
 
760
STUB_SERVER_KEY = """
 
761
-----BEGIN RSA PRIVATE KEY-----
 
762
MIICWgIBAAKBgQDTj1bqB4WmayWNPB+8jVSYpZYk80Ujvj680pOTh2bORBjbIAyz
 
763
oWGW+GUjzKxTiiPvVmxFgx5wdsFvF03v34lEVVhMpouqPAYQ15N37K/ir5XY+9m/
 
764
d8ufMCkjeXsQkKqFbAlQcnWMCRnOoPHS3I4vi6hmnDDeeYTSRvfLbW0fhwIBIwKB
 
765
gBIiOqZYaoqbeD9OS9z2K9KR2atlTxGxOJPXiP4ESqP3NVScWNwyZ3NXHpyrJLa0
 
766
EbVtzsQhLn6rF+TzXnOlcipFvjsem3iYzCpuChfGQ6SovTcOjHV9z+hnpXvQ/fon
 
767
soVRZY65wKnF7IAoUwTmJS9opqgrN6kRgCd3DASAMd1bAkEA96SBVWFt/fJBNJ9H
 
768
tYnBKZGw0VeHOYmVYbvMSstssn8un+pQpUm9vlG/bp7Oxd/m+b9KWEh2xPfv6zqU
 
769
avNwHwJBANqzGZa/EpzF4J8pGti7oIAPUIDGMtfIcmqNXVMckrmzQ2vTfqtkEZsA
 
770
4rE1IERRyiJQx6EJsz21wJmGV9WJQ5kCQQDwkS0uXqVdFzgHO6S++tjmjYcxwr3g
 
771
H0CoFYSgbddOT6miqRskOQF3DZVkJT3kyuBgU2zKygz52ukQZMqxCb1fAkASvuTv
 
772
qfpH87Qq5kQhNKdbbwbmd2NxlNabazPijWuphGTdW0VfJdWfklyS2Kr+iqrs/5wV
 
773
HhathJt636Eg7oIjAkA8ht3MQ+XSl9yIJIS8gVpbPxSw5OMfw0PjVE7tBdQruiSc
 
774
nvuQES5C9BMHjF39LZiGH1iLQy7FgdHyoP+eodI7
 
775
-----END RSA PRIVATE KEY-----
 
776
"""
 
777
 
 
778
 
 
779
class SocketListener(threading.Thread):
 
780
 
 
781
    def __init__(self, callback):
 
782
        threading.Thread.__init__(self)
 
783
        self._callback = callback
 
784
        self._socket = socket.socket()
 
785
        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 
786
        self._socket.bind(('localhost', 0))
 
787
        self._socket.listen(1)
 
788
        self.port = self._socket.getsockname()[1]
 
789
        self._stop_event = threading.Event()
 
790
 
 
791
    def stop(self):
 
792
        # called from outside this thread
 
793
        self._stop_event.set()
 
794
        # use a timeout here, because if the test fails, the server thread may
 
795
        # never notice the stop_event.
 
796
        self.join(5.0)
 
797
        self._socket.close()
 
798
 
 
799
    def run(self):
 
800
        while True:
 
801
            readable, writable_unused, exception_unused = \
 
802
                select.select([self._socket], [], [], 0.1)
 
803
            if self._stop_event.isSet():
 
804
                return
 
805
            if len(readable) == 0:
 
806
                continue
 
807
            try:
 
808
                s, addr_unused = self._socket.accept()
 
809
                # because the loopback socket is inline, and transports are
 
810
                # never explicitly closed, best to launch a new thread.
 
811
                threading.Thread(target=self._callback, args=(s,)).start()
 
812
            except socket.error, x:
 
813
                sys.excepthook(*sys.exc_info())
 
814
                warning('Socket error during accept() within unit test server'
 
815
                        ' thread: %r' % x)
 
816
            except Exception, x:
 
817
                # probably a failed test; unit test thread will log the
 
818
                # failure/error
 
819
                sys.excepthook(*sys.exc_info())
 
820
                warning('Exception from within unit test server thread: %r' % 
 
821
                        x)
 
822
 
 
823
 
 
824
class SocketDelay(object):
 
825
    """A socket decorator to make TCP appear slower.
 
826
 
 
827
    This changes recv, send, and sendall to add a fixed latency to each python
 
828
    call if a new roundtrip is detected. That is, when a recv is called and the
 
829
    flag new_roundtrip is set, latency is charged. Every send and send_all
 
830
    sets this flag.
 
831
 
 
832
    In addition every send, sendall and recv sleeps a bit per character send to
 
833
    simulate bandwidth.
 
834
 
 
835
    Not all methods are implemented, this is deliberate as this class is not a
 
836
    replacement for the builtin sockets layer. fileno is not implemented to
 
837
    prevent the proxy being bypassed. 
 
838
    """
 
839
 
 
840
    simulated_time = 0
 
841
    _proxied_arguments = dict.fromkeys([
 
842
        "close", "getpeername", "getsockname", "getsockopt", "gettimeout",
 
843
        "setblocking", "setsockopt", "settimeout", "shutdown"])
 
844
 
 
845
    def __init__(self, sock, latency, bandwidth=1.0, 
 
846
                 really_sleep=True):
 
847
        """ 
 
848
        :param bandwith: simulated bandwith (MegaBit)
 
849
        :param really_sleep: If set to false, the SocketDelay will just
 
850
        increase a counter, instead of calling time.sleep. This is useful for
 
851
        unittesting the SocketDelay.
 
852
        """
 
853
        self.sock = sock
 
854
        self.latency = latency
 
855
        self.really_sleep = really_sleep
 
856
        self.time_per_byte = 1 / (bandwidth / 8.0 * 1024 * 1024) 
 
857
        self.new_roundtrip = False
 
858
 
 
859
    def sleep(self, s):
 
860
        if self.really_sleep:
 
861
            time.sleep(s)
 
862
        else:
 
863
            SocketDelay.simulated_time += s
 
864
 
 
865
    def __getattr__(self, attr):
 
866
        if attr in SocketDelay._proxied_arguments:
 
867
            return getattr(self.sock, attr)
 
868
        raise AttributeError("'SocketDelay' object has no attribute %r" %
 
869
                             attr)
 
870
 
 
871
    def dup(self):
 
872
        return SocketDelay(self.sock.dup(), self.latency, self.time_per_byte,
 
873
                           self._sleep)
 
874
 
 
875
    def recv(self, *args):
 
876
        data = self.sock.recv(*args)
 
877
        if data and self.new_roundtrip:
 
878
            self.new_roundtrip = False
 
879
            self.sleep(self.latency)
 
880
        self.sleep(len(data) * self.time_per_byte)
 
881
        return data
 
882
 
 
883
    def sendall(self, data, flags=0):
 
884
        if not self.new_roundtrip:
 
885
            self.new_roundtrip = True
 
886
            self.sleep(self.latency)
 
887
        self.sleep(len(data) * self.time_per_byte)
 
888
        return self.sock.sendall(data, flags)
 
889
 
 
890
    def send(self, data, flags=0):
 
891
        if not self.new_roundtrip:
 
892
            self.new_roundtrip = True
 
893
            self.sleep(self.latency)
 
894
        bytes_sent = self.sock.send(data, flags)
 
895
        self.sleep(bytes_sent * self.time_per_byte)
 
896
        return bytes_sent
 
897
 
 
898
 
 
899
class SFTPServer(Server):
 
900
    """Common code for SFTP server facilities."""
 
901
 
 
902
    def __init__(self, server_interface=StubServer):
 
903
        self._original_vendor = None
 
904
        self._homedir = None
 
905
        self._server_homedir = None
 
906
        self._listener = None
 
907
        self._root = None
 
908
        self._vendor = ssh.ParamikoVendor()
 
909
        self._server_interface = server_interface
 
910
        # sftp server logs
 
911
        self.logs = []
 
912
        self.add_latency = 0
 
913
 
 
914
    def _get_sftp_url(self, path):
 
915
        """Calculate an sftp url to this server for path."""
 
916
        return 'sftp://foo:bar@localhost:%d/%s' % (self._listener.port, path)
 
917
 
 
918
    def log(self, message):
 
919
        """StubServer uses this to log when a new server is created."""
 
920
        self.logs.append(message)
 
921
 
 
922
    def _run_server_entry(self, sock):
 
923
        """Entry point for all implementations of _run_server.
 
924
        
 
925
        If self.add_latency is > 0.000001 then sock is given a latency adding
 
926
        decorator.
 
927
        """
 
928
        if self.add_latency > 0.000001:
 
929
            sock = SocketDelay(sock, self.add_latency)
 
930
        return self._run_server(sock)
 
931
 
 
932
    def _run_server(self, s):
 
933
        ssh_server = paramiko.Transport(s)
 
934
        key_file = pathjoin(self._homedir, 'test_rsa.key')
 
935
        f = open(key_file, 'w')
 
936
        f.write(STUB_SERVER_KEY)
 
937
        f.close()
 
938
        host_key = paramiko.RSAKey.from_private_key_file(key_file)
 
939
        ssh_server.add_server_key(host_key)
 
940
        server = self._server_interface(self)
 
941
        ssh_server.set_subsystem_handler('sftp', paramiko.SFTPServer,
 
942
                                         StubSFTPServer, root=self._root,
 
943
                                         home=self._server_homedir)
 
944
        event = threading.Event()
 
945
        ssh_server.start_server(event, server)
 
946
        event.wait(5.0)
 
947
    
 
948
    def setUp(self, backing_server=None):
 
949
        # XXX: TODO: make sftpserver back onto backing_server rather than local
 
950
        # disk.
 
951
        assert (backing_server is None or
 
952
                isinstance(backing_server, local.LocalURLServer)), (
 
953
            "backing_server should not be %r, because this can only serve the "
 
954
            "local current working directory." % (backing_server,))
 
955
        self._original_vendor = ssh._ssh_vendor_manager._cached_ssh_vendor
 
956
        ssh._ssh_vendor_manager._cached_ssh_vendor = self._vendor
 
957
        if sys.platform == 'win32':
 
958
            # Win32 needs to use the UNICODE api
 
959
            self._homedir = getcwd()
 
960
        else:
 
961
            # But Linux SFTP servers should just deal in bytestreams
 
962
            self._homedir = os.getcwd()
 
963
        if self._server_homedir is None:
 
964
            self._server_homedir = self._homedir
 
965
        self._root = '/'
 
966
        if sys.platform == 'win32':
 
967
            self._root = ''
 
968
        self._listener = SocketListener(self._run_server_entry)
 
969
        self._listener.setDaemon(True)
 
970
        self._listener.start()
 
971
 
 
972
    def tearDown(self):
 
973
        """See bzrlib.transport.Server.tearDown."""
 
974
        self._listener.stop()
 
975
        ssh._ssh_vendor_manager._cached_ssh_vendor = self._original_vendor
 
976
 
 
977
    def get_bogus_url(self):
 
978
        """See bzrlib.transport.Server.get_bogus_url."""
 
979
        # this is chosen to try to prevent trouble with proxies, wierd dns, etc
 
980
        # we bind a random socket, so that we get a guaranteed unused port
 
981
        # we just never listen on that port
 
982
        s = socket.socket()
 
983
        s.bind(('localhost', 0))
 
984
        return 'sftp://%s:%s/' % s.getsockname()
 
985
 
 
986
 
 
987
class SFTPFullAbsoluteServer(SFTPServer):
 
988
    """A test server for sftp transports, using absolute urls and ssh."""
 
989
 
 
990
    def get_url(self):
 
991
        """See bzrlib.transport.Server.get_url."""
 
992
        homedir = self._homedir
 
993
        if sys.platform != 'win32':
 
994
            # Remove the initial '/' on all platforms but win32
 
995
            homedir = homedir[1:]
 
996
        return self._get_sftp_url(urlutils.escape(homedir))
 
997
 
 
998
 
 
999
class SFTPServerWithoutSSH(SFTPServer):
 
1000
    """An SFTP server that uses a simple TCP socket pair rather than SSH."""
 
1001
 
 
1002
    def __init__(self):
 
1003
        super(SFTPServerWithoutSSH, self).__init__()
 
1004
        self._vendor = ssh.LoopbackVendor()
 
1005
 
 
1006
    def _run_server(self, sock):
 
1007
        # Re-import these as locals, so that they're still accessible during
 
1008
        # interpreter shutdown (when all module globals get set to None, leading
 
1009
        # to confusing errors like "'NoneType' object has no attribute 'error'".
 
1010
        class FakeChannel(object):
 
1011
            def get_transport(self):
 
1012
                return self
 
1013
            def get_log_channel(self):
 
1014
                return 'paramiko'
 
1015
            def get_name(self):
 
1016
                return '1'
 
1017
            def get_hexdump(self):
 
1018
                return False
 
1019
            def close(self):
 
1020
                pass
 
1021
 
 
1022
        server = paramiko.SFTPServer(FakeChannel(), 'sftp', StubServer(self), StubSFTPServer,
 
1023
                                     root=self._root, home=self._server_homedir)
 
1024
        try:
 
1025
            server.start_subsystem('sftp', None, sock)
 
1026
        except socket.error, e:
 
1027
            if (len(e.args) > 0) and (e.args[0] == errno.EPIPE):
 
1028
                # it's okay for the client to disconnect abruptly
 
1029
                # (bug in paramiko 1.6: it should absorb this exception)
 
1030
                pass
 
1031
            else:
 
1032
                raise
 
1033
        except Exception, e:
 
1034
            # This typically seems to happen during interpreter shutdown, so
 
1035
            # most of the useful ways to report this error are won't work.
 
1036
            # Writing the exception type, and then the text of the exception,
 
1037
            # seems to be the best we can do.
 
1038
            import sys
 
1039
            sys.stderr.write('\nEXCEPTION %r: ' % (e.__class__,))
 
1040
            sys.stderr.write('%s\n\n' % (e,))
 
1041
        server.finish_subsystem()
 
1042
 
 
1043
 
 
1044
class SFTPAbsoluteServer(SFTPServerWithoutSSH):
 
1045
    """A test server for sftp transports, using absolute urls."""
 
1046
 
 
1047
    def get_url(self):
 
1048
        """See bzrlib.transport.Server.get_url."""
 
1049
        homedir = self._homedir
 
1050
        if sys.platform != 'win32':
 
1051
            # Remove the initial '/' on all platforms but win32
 
1052
            homedir = homedir[1:]
 
1053
        return self._get_sftp_url(urlutils.escape(homedir))
 
1054
 
 
1055
 
 
1056
class SFTPHomeDirServer(SFTPServerWithoutSSH):
 
1057
    """A test server for sftp transports, using homedir relative urls."""
 
1058
 
 
1059
    def get_url(self):
 
1060
        """See bzrlib.transport.Server.get_url."""
 
1061
        return self._get_sftp_url("~/")
 
1062
 
 
1063
 
 
1064
class SFTPSiblingAbsoluteServer(SFTPAbsoluteServer):
 
1065
    """A test server for sftp transports where only absolute paths will work.
 
1066
 
 
1067
    It does this by serving from a deeply-nested directory that doesn't exist.
 
1068
    """
 
1069
 
 
1070
    def setUp(self, backing_server=None):
 
1071
        self._server_homedir = '/dev/noone/runs/tests/here'
 
1072
        super(SFTPSiblingAbsoluteServer, self).setUp(backing_server)
 
1073
 
900
1074
 
901
1075
def get_test_permutations():
902
1076
    """Return the permutations to be used in testing."""
903
 
    from bzrlib.tests import stub_sftp
904
 
    return [(SFTPTransport, stub_sftp.SFTPAbsoluteServer),
905
 
            (SFTPTransport, stub_sftp.SFTPHomeDirServer),
906
 
            (SFTPTransport, stub_sftp.SFTPSiblingAbsoluteServer),
 
1077
    return [(SFTPTransport, SFTPAbsoluteServer),
 
1078
            (SFTPTransport, SFTPHomeDirServer),
 
1079
            (SFTPTransport, SFTPSiblingAbsoluteServer),
907
1080
            ]