~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-11 14:59:52 UTC
  • mto: (2520.5.2 bzr.mpbundle)
  • mto: This revision was merged to the branch mainline in revision 2631.
  • Revision ID: abentley@panoramicfeedback.com-20070611145952-cwt4480gphdhen6l
Get installation started

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2010 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
36
 
import warnings
 
35
import urllib
 
36
import urlparse
 
37
import weakref
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
52
51
from bzrlib.trace import mutter, warning
53
52
from bzrlib.transport import (
54
 
    FileFileStream,
55
 
    _file_streams,
 
53
    local,
 
54
    register_urlparse_netloc_protocol,
 
55
    Server,
 
56
    split_url,
56
57
    ssh,
57
 
    ConnectedTransport,
 
58
    Transport,
58
59
    )
59
60
 
60
 
# Disable one particular warning that comes from paramiko in Python2.5; if
61
 
# this is emitted at the wrong time it tends to cause spurious test failures
62
 
# or at least noise in the test case::
63
 
#
64
 
# [1770/7639 in 86s, 1 known failures, 50 skipped, 2 missing features]
65
 
# test_permissions.TestSftpPermissions.test_new_files
66
 
# /var/lib/python-support/python2.5/paramiko/message.py:226: DeprecationWarning: integer argument expected, got float
67
 
#  self.packet.write(struct.pack('>I', n))
68
 
warnings.filterwarnings('ignore',
69
 
        'integer argument expected, got float',
70
 
        category=DeprecationWarning,
71
 
        module='paramiko.message')
72
 
 
73
61
try:
74
62
    import paramiko
75
63
except ImportError, e:
77
65
else:
78
66
    from paramiko.sftp import (SFTP_FLAG_WRITE, SFTP_FLAG_CREATE,
79
67
                               SFTP_FLAG_EXCL, SFTP_FLAG_TRUNC,
80
 
                               SFTP_OK, CMD_HANDLE, CMD_OPEN)
 
68
                               CMD_HANDLE, CMD_OPEN)
81
69
    from paramiko.sftp_attr import SFTPAttributes
82
70
    from paramiko.sftp_file import SFTPFile
83
71
 
84
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
 
85
83
_paramiko_version = getattr(paramiko, '__version_info__', (0, 0, 0))
86
84
# don't use prefetch unless paramiko version >= 1.5.5 (there were bugs earlier)
87
85
_default_do_prefetch = (_paramiko_version >= (1, 5, 5))
88
86
 
89
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
 
90
96
class SFTPLock(object):
91
97
    """This fakes a lock in a remote location.
92
 
 
 
98
    
93
99
    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
 
100
    doesn't work well on all transports and they are only used in 
95
101
    deprecated storage formats.
96
102
    """
97
 
 
 
103
    
98
104
    __slots__ = ['path', 'lock_path', 'lock_file', 'transport']
99
105
 
100
106
    def __init__(self, path, transport):
 
107
        assert isinstance(transport, SFTPTransport)
 
108
 
101
109
        self.lock_file = None
102
110
        self.path = path
103
111
        self.lock_path = path + '.write-lock'
109
117
        except FileExists:
110
118
            raise LockError('File %r already locked' % (self.path,))
111
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
 
112
126
    def unlock(self):
113
127
        if not self.lock_file:
114
128
            return
121
135
            pass
122
136
 
123
137
 
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
 
class SFTPTransport(ConnectedTransport):
 
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):
315
201
    """Transport implementation for SFTP access."""
316
202
 
317
203
    _do_prefetch = _default_do_prefetch
332
218
    # up the request itself, rather than us having to worry about it
333
219
    _max_request_size = 32768
334
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
 
335
247
    def _remote_path(self, relpath):
336
248
        """Return the path to be passed along the sftp protocol for relpath.
337
 
 
338
 
        :param relpath: is a urlencoded string.
339
 
        """
340
 
        remote_path = self._parsed_url.clone(relpath).path
341
 
        # the initial slash should be removed from the path, and treated as a
342
 
        # homedir relative path (the path begins with a double slash if it is
343
 
        # absolute).  see draft-ietf-secsh-scp-sftp-ssh-uri-03.txt
344
 
        # RBC 20060118 we are not using this as its too user hostile. instead
345
 
        # we are following lftp and using /~/foo to mean '~/foo'
346
 
        # vila--20070602 and leave absolute paths begin with a single slash.
347
 
        if remote_path.startswith('/~/'):
348
 
            remote_path = remote_path[3:]
349
 
        elif remote_path == '/~':
350
 
            remote_path = ''
351
 
        return remote_path
352
 
 
353
 
    def _create_connection(self, credentials=None):
354
 
        """Create a new connection with the provided credentials.
355
 
 
356
 
        :param credentials: The credentials needed to establish the connection.
357
 
 
358
 
        :return: The created connection and its associated credentials.
359
 
 
360
 
        The credentials are only the password as it may have been entered
361
 
        interactively by the user and may be different from the one provided
362
 
        in base url at transport creation time.
363
 
        """
364
 
        if credentials is None:
365
 
            password = self._parsed_url.password
 
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('/'))
366
311
        else:
367
 
            password = credentials
368
 
 
369
 
        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()
383
 
 
384
 
    def _get_sftp(self):
385
 
        """Ensures that a connection is established"""
386
 
        connection = self._get_connection()
387
 
        if connection is None:
388
 
            # First connection ever
389
 
            connection, credentials = self._create_connection()
390
 
            self._set_connection(connection, credentials)
391
 
        return connection
 
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('/')
392
372
 
393
373
    def has(self, relpath):
394
374
        """
395
375
        Does the target location exist?
396
376
        """
397
377
        try:
398
 
            self._get_sftp().stat(self._remote_path(relpath))
399
 
            # stat result is about 20 bytes, let's say
400
 
            self._report_activity(20, 'read')
 
378
            self._sftp.stat(self._remote_path(relpath))
401
379
            return True
402
380
        except IOError:
403
381
            return False
404
382
 
405
383
    def get(self, relpath):
406
 
        """Get the file at the given relative path.
 
384
        """
 
385
        Get the file at the given relative path.
407
386
 
408
387
        :param relpath: The relative path to the file
409
388
        """
410
389
        try:
411
390
            path = self._remote_path(relpath)
412
 
            f = self._get_sftp().file(path, mode='rb')
 
391
            f = self._sftp.file(path, mode='rb')
413
392
            if self._do_prefetch and (getattr(f, 'prefetch', None) is not None):
414
393
                f.prefetch()
415
394
            return f
416
395
        except (IOError, paramiko.SSHException), e:
417
 
            self._translate_io_exception(e, path, ': error retrieving',
418
 
                failure_exc=errors.ReadError)
419
 
 
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
 
    def _readv(self, relpath, offsets):
 
396
            self._translate_io_exception(e, path, ': error retrieving')
 
397
 
 
398
    def readv(self, relpath, offsets):
431
399
        """See Transport.readv()"""
432
400
        # We overload the default readv() because we want to use a file
433
401
        # that does not have prefetch enabled.
437
405
 
438
406
        try:
439
407
            path = self._remote_path(relpath)
440
 
            fp = self._get_sftp().file(path, mode='rb')
 
408
            fp = self._sftp.file(path, mode='rb')
441
409
            readv = getattr(fp, 'readv', None)
442
410
            if readv:
443
411
                return self._sftp_readv(fp, offsets, relpath)
444
 
            if 'sftp' in debug.debug_flags:
445
 
                mutter('seek and read %s offsets', len(offsets))
 
412
            mutter('seek and read %s offsets', len(offsets))
446
413
            return self._seek_and_read(fp, offsets, relpath)
447
414
        except (IOError, paramiko.SSHException), e:
448
415
            self._translate_io_exception(e, path, ': error retrieving')
449
416
 
450
 
    def recommended_page_size(self):
451
 
        """See Transport.recommended_page_size().
452
 
 
453
 
        For SFTP we suggest a large page size to reduce the overhead
454
 
        introduced by latency.
455
 
        """
456
 
        return 64 * 1024
457
 
 
458
 
    def _sftp_readv(self, fp, offsets, relpath):
 
417
    def _sftp_readv(self, fp, offsets, relpath='<unknown>'):
459
418
        """Use the readv() member of fp to do async readv.
460
419
 
461
 
        Then read them using paramiko.readv(). paramiko.readv()
 
420
        And then read them using paramiko.readv(). paramiko.readv()
462
421
        does not support ranges > 64K, so it caps the request size, and
463
 
        just reads until it gets all the stuff it wants.
 
422
        just reads until it gets all the stuff it wants
464
423
        """
465
 
        helper = _SFTPReadvHelper(offsets, relpath, self._report_activity)
466
 
        return helper.request_and_yield_offsets(fp)
 
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))
467
519
 
468
520
    def put_file(self, relpath, f, mode=None):
469
521
        """
474
526
        :param mode: The final mode for the file
475
527
        """
476
528
        final_path = self._remote_path(relpath)
477
 
        return self._put(final_path, f, mode=mode)
 
529
        self._put(final_path, f, mode=mode)
478
530
 
479
531
    def _put(self, abspath, f, mode=None):
480
532
        """Helper function so both put() and copy_abspaths can reuse the code"""
485
537
        try:
486
538
            try:
487
539
                fout.set_pipelined(True)
488
 
                length = self._pump(f, fout)
 
540
                self._pump(f, fout)
489
541
            except (IOError, paramiko.SSHException), e:
490
542
                self._translate_io_exception(e, tmp_abspath)
491
543
            # XXX: This doesn't truly help like we would like it to.
494
546
            #      sticky bit. So it is probably best to stop chmodding, and
495
547
            #      just tell users that they need to set the umask correctly.
496
548
            #      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
 
549
            #      will handle when the user wants the final mode to be more 
 
550
            #      restrictive. And then we avoid a round trip. Unless 
499
551
            #      paramiko decides to expose an async chmod()
500
552
 
501
553
            # This is designed to chmod() right before we close.
502
 
            # Because we set_pipelined() earlier, theoretically we might
 
554
            # Because we set_pipelined() earlier, theoretically we might 
503
555
            # avoid the round trip for fout.close()
504
556
            if mode is not None:
505
 
                self._get_sftp().chmod(tmp_abspath, mode)
 
557
                self._sftp.chmod(tmp_abspath, mode)
506
558
            fout.close()
507
559
            closed = True
508
560
            self._rename_and_overwrite(tmp_abspath, abspath)
509
 
            return length
510
561
        except Exception, e:
511
562
            # If we fail, try to clean up the temporary file
512
563
            # before we throw the exception
518
569
            try:
519
570
                if not closed:
520
571
                    fout.close()
521
 
                self._get_sftp().remove(tmp_abspath)
 
572
                self._sftp.remove(tmp_abspath)
522
573
            except:
523
574
                # raise the saved except
524
575
                raise e
539
590
            fout = None
540
591
            try:
541
592
                try:
542
 
                    fout = self._get_sftp().file(abspath, mode='wb')
 
593
                    fout = self._sftp.file(abspath, mode='wb')
543
594
                    fout.set_pipelined(True)
544
595
                    writer(fout)
545
596
                except (paramiko.SSHException, IOError), e:
547
598
                                                 ': unable to open')
548
599
 
549
600
                # This is designed to chmod() right before we close.
550
 
                # Because we set_pipelined() earlier, theoretically we might
 
601
                # Because we set_pipelined() earlier, theoretically we might 
551
602
                # avoid the round trip for fout.close()
552
603
                if mode is not None:
553
 
                    self._get_sftp().chmod(abspath, mode)
 
604
                    self._sftp.chmod(abspath, mode)
554
605
            finally:
555
606
                if fout is not None:
556
607
                    fout.close()
604
655
 
605
656
    def iter_files_recursive(self):
606
657
        """Walk the relative paths of all files in this transport."""
607
 
        # progress is handled by list_dir
608
658
        queue = list(self.list_dir('.'))
609
659
        while queue:
610
660
            relpath = queue.pop(0)
621
671
        else:
622
672
            local_mode = mode
623
673
        try:
624
 
            self._report_activity(len(abspath), 'write')
625
 
            self._get_sftp().mkdir(abspath, local_mode)
626
 
            self._report_activity(1, 'read')
 
674
            self._sftp.mkdir(abspath, local_mode)
627
675
            if mode is not None:
628
 
                # chmod a dir through sftp will erase any sgid bit set
629
 
                # on the server side.  So, if the bit mode are already
630
 
                # set, avoid the chmod.  If the mode is not fine but
631
 
                # the sgid bit is set, report a warning to the user
632
 
                # with the umask fix.
633
 
                stat = self._get_sftp().lstat(abspath)
634
 
                mode = mode & 0777 # can't set special bits anyway
635
 
                if mode != stat.st_mode & 0777:
636
 
                    if stat.st_mode & 06000:
637
 
                        warning('About to chmod %s over sftp, which will result'
638
 
                                ' in its suid or sgid bits being cleared.  If'
639
 
                                ' you want to preserve those bits, change your '
640
 
                                ' environment on the server to use umask 0%03o.'
641
 
                                % (abspath, 0777 - mode))
642
 
                    self._get_sftp().chmod(abspath, mode=mode)
 
676
                self._sftp.chmod(abspath, mode=mode)
643
677
        except (paramiko.SSHException, IOError), e:
644
678
            self._translate_io_exception(e, abspath, ': unable to mkdir',
645
679
                failure_exc=FileExists)
648
682
        """Create a directory at the given path."""
649
683
        self._mkdir(self._remote_path(relpath), mode=mode)
650
684
 
651
 
    def open_write_stream(self, relpath, mode=None):
652
 
        """See Transport.open_write_stream."""
653
 
        # initialise the file to zero-length
654
 
        # this is three round trips, but we don't use this
655
 
        # api more than once per write_group at the moment so
656
 
        # it is a tolerable overhead. Better would be to truncate
657
 
        # the file after opening. RBC 20070805
658
 
        self.put_bytes_non_atomic(relpath, "", mode)
659
 
        abspath = self._remote_path(relpath)
660
 
        # TODO: jam 20060816 paramiko doesn't publicly expose a way to
661
 
        #       set the file mode at create time. If it does, use it.
662
 
        #       But for now, we just chmod later anyway.
663
 
        handle = None
664
 
        try:
665
 
            handle = self._get_sftp().file(abspath, mode='wb')
666
 
            handle.set_pipelined(True)
667
 
        except (paramiko.SSHException, IOError), e:
668
 
            self._translate_io_exception(e, abspath,
669
 
                                         ': unable to open')
670
 
        _file_streams[self.abspath(relpath)] = handle
671
 
        return FileFileStream(self, relpath, handle)
672
 
 
673
 
    def _translate_io_exception(self, e, path, more_info='',
 
685
    def _translate_io_exception(self, e, path, more_info='', 
674
686
                                failure_exc=PathError):
675
687
        """Translate a paramiko or IOError into a friendlier exception.
676
688
 
681
693
        :param failure_exc: Paramiko has the super fun ability to raise completely
682
694
                           opaque errors that just set "e.args = ('Failure',)" with
683
695
                           no more information.
684
 
                           If this parameter is set, it defines the exception
 
696
                           If this parameter is set, it defines the exception 
685
697
                           to raise in these cases.
686
698
        """
687
699
        # paramiko seems to generate detailless errors.
690
702
            if (e.args == ('No such file or directory',) or
691
703
                e.args == ('No such file',)):
692
704
                raise NoSuchFile(path, str(e) + more_info)
693
 
            if (e.args == ('mkdir failed',) or
694
 
                e.args[0].startswith('syserr: File exists')):
 
705
            if (e.args == ('mkdir failed',)):
695
706
                raise FileExists(path, str(e) + more_info)
696
707
            # strange but true, for the paramiko server.
697
708
            if (e.args == ('Failure',)):
698
709
                raise failure_exc(path, str(e) + more_info)
699
 
            # Can be something like args = ('Directory not empty:
700
 
            # '/srv/bazaar.launchpad.net/blah...: '
701
 
            # [Errno 39] Directory not empty',)
702
 
            if (e.args[0].startswith('Directory not empty: ')
703
 
                or getattr(e, 'errno', None) == errno.ENOTEMPTY):
704
 
                raise errors.DirectoryNotEmpty(path, str(e))
705
 
            if e.args == ('Operation unsupported',):
706
 
                raise errors.TransportNotPossible()
707
710
            mutter('Raising exception with args %s', e.args)
708
711
        if getattr(e, 'errno', None) is not None:
709
712
            mutter('Raising exception with errno %s', e.errno)
716
719
        """
717
720
        try:
718
721
            path = self._remote_path(relpath)
719
 
            fout = self._get_sftp().file(path, 'ab')
 
722
            fout = self._sftp.file(path, 'ab')
720
723
            if mode is not None:
721
 
                self._get_sftp().chmod(path, mode)
 
724
                self._sftp.chmod(path, mode)
722
725
            result = fout.tell()
723
726
            self._pump(f, fout)
724
727
            return result
728
731
    def rename(self, rel_from, rel_to):
729
732
        """Rename without special overwriting"""
730
733
        try:
731
 
            self._get_sftp().rename(self._remote_path(rel_from),
 
734
            self._sftp.rename(self._remote_path(rel_from),
732
735
                              self._remote_path(rel_to))
733
736
        except (IOError, paramiko.SSHException), e:
734
737
            self._translate_io_exception(e, rel_from,
736
739
 
737
740
    def _rename_and_overwrite(self, abs_from, abs_to):
738
741
        """Do a fancy rename on the remote server.
739
 
 
 
742
        
740
743
        Using the implementation provided by osutils.
741
744
        """
742
745
        try:
743
 
            sftp = self._get_sftp()
744
746
            fancy_rename(abs_from, abs_to,
745
 
                         rename_func=sftp.rename,
746
 
                         unlink_func=sftp.remove)
 
747
                    rename_func=self._sftp.rename,
 
748
                    unlink_func=self._sftp.remove)
747
749
        except (IOError, paramiko.SSHException), e:
748
 
            self._translate_io_exception(e, abs_from,
749
 
                                         ': unable to rename to %r' % (abs_to))
 
750
            self._translate_io_exception(e, abs_from, ': unable to rename to %r' % (abs_to))
750
751
 
751
752
    def move(self, rel_from, rel_to):
752
753
        """Move the item at rel_from to the location at rel_to"""
758
759
        """Delete the item at relpath"""
759
760
        path = self._remote_path(relpath)
760
761
        try:
761
 
            self._get_sftp().remove(path)
 
762
            self._sftp.remove(path)
762
763
        except (IOError, paramiko.SSHException), e:
763
764
            self._translate_io_exception(e, path, ': unable to delete')
764
 
 
765
 
    def external_url(self):
766
 
        """See bzrlib.transport.Transport.external_url."""
767
 
        # the external path for SFTP is the base
768
 
        return self.base
769
 
 
 
765
            
770
766
    def listable(self):
771
767
        """Return True if this store supports listing."""
772
768
        return True
781
777
        # -- David Allouche 2006-08-11
782
778
        path = self._remote_path(relpath)
783
779
        try:
784
 
            entries = self._get_sftp().listdir(path)
785
 
            self._report_activity(sum(map(len, entries)), 'read')
 
780
            entries = self._sftp.listdir(path)
786
781
        except (IOError, paramiko.SSHException), e:
787
782
            self._translate_io_exception(e, path, ': failed to list_dir')
788
783
        return [urlutils.escape(entry) for entry in entries]
791
786
        """See Transport.rmdir."""
792
787
        path = self._remote_path(relpath)
793
788
        try:
794
 
            return self._get_sftp().rmdir(path)
 
789
            return self._sftp.rmdir(path)
795
790
        except (IOError, paramiko.SSHException), e:
796
791
            self._translate_io_exception(e, path, ': failed to rmdir')
797
792
 
799
794
        """Return the stat information for a file."""
800
795
        path = self._remote_path(relpath)
801
796
        try:
802
 
            return self._get_sftp().lstat(path)
 
797
            return self._sftp.stat(path)
803
798
        except (IOError, paramiko.SSHException), e:
804
799
            self._translate_io_exception(e, path, ': unable to stat')
805
800
 
806
 
    def readlink(self, relpath):
807
 
        """See Transport.readlink."""
808
 
        path = self._remote_path(relpath)
809
 
        try:
810
 
            return self._get_sftp().readlink(path)
811
 
        except (IOError, paramiko.SSHException), e:
812
 
            self._translate_io_exception(e, path, ': unable to readlink')
813
 
 
814
 
    def symlink(self, source, link_name):
815
 
        """See Transport.symlink."""
816
 
        try:
817
 
            conn = self._get_sftp()
818
 
            sftp_retval = conn.symlink(source, link_name)
819
 
            if SFTP_OK != sftp_retval:
820
 
                raise TransportError(
821
 
                    '%r: unable to create symlink to %r' % (link_name, source),
822
 
                    sftp_retval
823
 
                )
824
 
        except (IOError, paramiko.SSHException), e:
825
 
            self._translate_io_exception(e, link_name,
826
 
                                         ': unable to create symlink to %r' % (source))
827
 
 
828
801
    def lock_read(self, relpath):
829
802
        """
830
803
        Lock the given file for shared (read) access.
851
824
        # that we have taken the lock.
852
825
        return SFTPLock(relpath, self)
853
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
 
854
837
    def _sftp_open_exclusive(self, abspath, mode=None):
855
838
        """Open a remote path exclusively.
856
839
 
867
850
        """
868
851
        # TODO: jam 20060816 Paramiko >= 1.6.2 (probably earlier) supports
869
852
        #       using the 'x' flag to indicate SFTP_FLAG_EXCL.
870
 
        #       However, there is no way to set the permission mode at open
 
853
        #       However, there is no way to set the permission mode at open 
871
854
        #       time using the sftp_client.file() functionality.
872
 
        path = self._get_sftp()._adjust_cwd(abspath)
 
855
        path = self._sftp._adjust_cwd(abspath)
873
856
        # mutter('sftp abspath %s => %s', abspath, path)
874
857
        attr = SFTPAttributes()
875
858
        if mode is not None:
876
859
            attr.st_mode = mode
877
 
        omode = (SFTP_FLAG_WRITE | SFTP_FLAG_CREATE
 
860
        omode = (SFTP_FLAG_WRITE | SFTP_FLAG_CREATE 
878
861
                | SFTP_FLAG_TRUNC | SFTP_FLAG_EXCL)
879
862
        try:
880
 
            t, msg = self._get_sftp()._request(CMD_OPEN, path, omode, attr)
 
863
            t, msg = self._sftp._request(CMD_OPEN, path, omode, attr)
881
864
            if t != CMD_HANDLE:
882
865
                raise TransportError('Expected an SFTP handle')
883
866
            handle = msg.get_string()
884
 
            return SFTPFile(self._get_sftp(), handle, 'wb', -1)
 
867
            return SFTPFile(self._sftp, handle, 'wb', -1)
885
868
        except (paramiko.SSHException, IOError), e:
886
869
            self._translate_io_exception(e, abspath, ': unable to open',
887
870
                failure_exc=FileExists)
893
876
        else:
894
877
            return True
895
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
 
896
1223
 
897
1224
def get_test_permutations():
898
1225
    """Return the permutations to be used in testing."""
899
 
    from bzrlib.tests import stub_sftp
900
 
    return [(SFTPTransport, stub_sftp.SFTPAbsoluteServer),
901
 
            (SFTPTransport, stub_sftp.SFTPHomeDirServer),
902
 
            (SFTPTransport, stub_sftp.SFTPSiblingAbsoluteServer),
 
1226
    return [(SFTPTransport, SFTPAbsoluteServer),
 
1227
            (SFTPTransport, SFTPHomeDirServer),
 
1228
            (SFTPTransport, SFTPSiblingAbsoluteServer),
903
1229
            ]