~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/transport/sftp.py

  • Committer: Aaron Bentley
  • Date: 2006-06-21 14:30:57 UTC
  • mfrom: (1801.1.1 bzr.dev)
  • mto: This revision was merged to the branch mainline in revision 1803.
  • Revision ID: abentley@panoramicfeedback.com-20060621143057-776e4b8d707e430e
Install benchmarks. (Jelmer Vernooij)

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006, 2007, 2008, 2009 Canonical Ltd
2
 
#
 
1
# Copyright (C) 2005 Robey Pointer <robey@lag.net>
 
2
# Copyright (C) 2005, 2006 Canonical Ltd
 
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
5
6
# the Free Software Foundation; either version 2 of the License, or
6
7
# (at your option) any later version.
7
 
#
 
8
 
8
9
# This program is distributed in the hope that it will be useful,
9
10
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
11
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
12
# GNU General Public License for more details.
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
 
# TODO: Remove the transport-based lock_read and lock_write methods.  They'll
20
 
# then raise TransportNotPossible, which will break remote access to any
21
 
# formats which rely on OS-level locks.  That should be fine as those formats
22
 
# are pretty old, but these combinations may have to be removed from the test
23
 
# suite.  Those formats all date back to 0.7; so we should be able to remove
24
 
# these methods when we officially drop support for those formats.
25
 
 
26
 
import bisect
27
20
import errno
28
 
import itertools
 
21
import getpass
29
22
import os
30
23
import random
 
24
import re
31
25
import select
32
 
import socket
33
26
import stat
 
27
import subprocess
34
28
import sys
35
29
import time
36
30
import urllib
37
31
import urlparse
38
 
import warnings
 
32
import weakref
39
33
 
40
 
from bzrlib import (
41
 
    config,
42
 
    debug,
43
 
    errors,
44
 
    urlutils,
45
 
    )
46
 
from bzrlib.errors import (FileExists,
47
 
                           NoSuchFile, PathNotChild,
 
34
from bzrlib.config import config_dir, ensure_config_dir_exists
 
35
from bzrlib.errors import (ConnectionError,
 
36
                           FileExists, 
 
37
                           TransportNotPossible, NoSuchFile, PathNotChild,
48
38
                           TransportError,
49
 
                           LockError,
 
39
                           LockError, 
50
40
                           PathError,
51
41
                           ParamikoNotPresent,
52
42
                           )
53
 
from bzrlib.osutils import pathjoin, fancy_rename, getcwd
54
 
from bzrlib.symbol_versioning import (
55
 
        deprecated_function,
56
 
        )
57
 
from bzrlib.trace import mutter, warning
 
43
from bzrlib.osutils import pathjoin, fancy_rename
 
44
from bzrlib.trace import mutter, warning, error
58
45
from bzrlib.transport import (
59
 
    FileFileStream,
60
 
    _file_streams,
61
 
    local,
 
46
    register_urlparse_netloc_protocol,
62
47
    Server,
63
 
    ssh,
64
 
    ConnectedTransport,
 
48
    split_url,
 
49
    Transport,
65
50
    )
66
 
 
67
 
# Disable one particular warning that comes from paramiko in Python2.5; if
68
 
# this is emitted at the wrong time it tends to cause spurious test failures
69
 
# or at least noise in the test case::
70
 
#
71
 
# [1770/7639 in 86s, 1 known failures, 50 skipped, 2 missing features]
72
 
# test_permissions.TestSftpPermissions.test_new_files
73
 
# /var/lib/python-support/python2.5/paramiko/message.py:226: DeprecationWarning: integer argument expected, got float
74
 
#  self.packet.write(struct.pack('>I', n))
75
 
warnings.filterwarnings('ignore',
76
 
        'integer argument expected, got float',
77
 
        category=DeprecationWarning,
78
 
        module='paramiko.message')
 
51
import bzrlib.ui
 
52
import bzrlib.urlutils as urlutils
79
53
 
80
54
try:
81
55
    import paramiko
87
61
                               CMD_HANDLE, CMD_OPEN)
88
62
    from paramiko.sftp_attr import SFTPAttributes
89
63
    from paramiko.sftp_file import SFTPFile
90
 
 
91
 
 
92
 
_paramiko_version = getattr(paramiko, '__version_info__', (0, 0, 0))
93
 
# don't use prefetch unless paramiko version >= 1.5.5 (there were bugs earlier)
94
 
_default_do_prefetch = (_paramiko_version >= (1, 5, 5))
 
64
    from paramiko.sftp_client import SFTPClient
 
65
 
 
66
 
 
67
register_urlparse_netloc_protocol('sftp')
 
68
 
 
69
 
 
70
def _ignore_sigint():
 
71
    # TODO: This should possibly ignore SIGHUP as well, but bzr currently
 
72
    # doesn't handle it itself.
 
73
    # <https://launchpad.net/products/bzr/+bug/41433/+index>
 
74
    import signal
 
75
    signal.signal(signal.SIGINT, signal.SIG_IGN)
 
76
    
 
77
 
 
78
def os_specific_subprocess_params():
 
79
    """Get O/S specific subprocess parameters."""
 
80
    if sys.platform == 'win32':
 
81
        # setting the process group and closing fds is not supported on 
 
82
        # win32
 
83
        return {}
 
84
    else:
 
85
        # We close fds other than the pipes as the child process does not need 
 
86
        # them to be open.
 
87
        #
 
88
        # We also set the child process to ignore SIGINT.  Normally the signal
 
89
        # would be sent to every process in the foreground process group, but
 
90
        # this causes it to be seen only by bzr and not by ssh.  Python will
 
91
        # generate a KeyboardInterrupt in bzr, and we will then have a chance
 
92
        # to release locks or do other cleanup over ssh before the connection
 
93
        # goes away.  
 
94
        # <https://launchpad.net/products/bzr/+bug/5987>
 
95
        #
 
96
        # Running it in a separate process group is not good because then it
 
97
        # can't get non-echoed input of a password or passphrase.
 
98
        # <https://launchpad.net/products/bzr/+bug/40508>
 
99
        return {'preexec_fn': _ignore_sigint,
 
100
                'close_fds': True,
 
101
                }
 
102
 
 
103
 
 
104
# don't use prefetch unless paramiko version >= 1.5.2 (there were bugs earlier)
 
105
_default_do_prefetch = False
 
106
if getattr(paramiko, '__version_info__', (0, 0, 0)) >= (1, 5, 5):
 
107
    _default_do_prefetch = True
 
108
 
 
109
 
 
110
_ssh_vendor = None
 
111
def _get_ssh_vendor():
 
112
    """Find out what version of SSH is on the system."""
 
113
    global _ssh_vendor
 
114
    if _ssh_vendor is not None:
 
115
        return _ssh_vendor
 
116
 
 
117
    _ssh_vendor = 'none'
 
118
 
 
119
    if 'BZR_SSH' in os.environ:
 
120
        _ssh_vendor = os.environ['BZR_SSH']
 
121
        if _ssh_vendor == 'paramiko':
 
122
            _ssh_vendor = 'none'
 
123
        return _ssh_vendor
 
124
 
 
125
    try:
 
126
        p = subprocess.Popen(['ssh', '-V'],
 
127
                             stdin=subprocess.PIPE,
 
128
                             stdout=subprocess.PIPE,
 
129
                             stderr=subprocess.PIPE,
 
130
                             **os_specific_subprocess_params())
 
131
        returncode = p.returncode
 
132
        stdout, stderr = p.communicate()
 
133
    except OSError:
 
134
        returncode = -1
 
135
        stdout = stderr = ''
 
136
    if 'OpenSSH' in stderr:
 
137
        mutter('ssh implementation is OpenSSH')
 
138
        _ssh_vendor = 'openssh'
 
139
    elif 'SSH Secure Shell' in stderr:
 
140
        mutter('ssh implementation is SSH Corp.')
 
141
        _ssh_vendor = 'ssh'
 
142
 
 
143
    if _ssh_vendor != 'none':
 
144
        return _ssh_vendor
 
145
 
 
146
    # XXX: 20051123 jamesh
 
147
    # A check for putty's plink or lsh would go here.
 
148
 
 
149
    mutter('falling back to paramiko implementation')
 
150
    return _ssh_vendor
 
151
 
 
152
 
 
153
class SFTPSubprocess:
 
154
    """A socket-like object that talks to an ssh subprocess via pipes."""
 
155
    def __init__(self, hostname, vendor, port=None, user=None):
 
156
        assert vendor in ['openssh', 'ssh']
 
157
        if vendor == 'openssh':
 
158
            args = ['ssh',
 
159
                    '-oForwardX11=no', '-oForwardAgent=no',
 
160
                    '-oClearAllForwardings=yes', '-oProtocol=2',
 
161
                    '-oNoHostAuthenticationForLocalhost=yes']
 
162
            if port is not None:
 
163
                args.extend(['-p', str(port)])
 
164
            if user is not None:
 
165
                args.extend(['-l', user])
 
166
            args.extend(['-s', hostname, 'sftp'])
 
167
        elif vendor == 'ssh':
 
168
            args = ['ssh', '-x']
 
169
            if port is not None:
 
170
                args.extend(['-p', str(port)])
 
171
            if user is not None:
 
172
                args.extend(['-l', user])
 
173
            args.extend(['-s', 'sftp', hostname])
 
174
 
 
175
        self.proc = subprocess.Popen(args,
 
176
                                     stdin=subprocess.PIPE,
 
177
                                     stdout=subprocess.PIPE,
 
178
                                     **os_specific_subprocess_params())
 
179
 
 
180
    def send(self, data):
 
181
        return os.write(self.proc.stdin.fileno(), data)
 
182
 
 
183
    def recv_ready(self):
 
184
        # TODO: jam 20051215 this function is necessary to support the
 
185
        # pipelined() function. In reality, it probably should use
 
186
        # poll() or select() to actually return if there is data
 
187
        # available, otherwise we probably don't get any benefit
 
188
        return True
 
189
 
 
190
    def recv(self, count):
 
191
        return os.read(self.proc.stdout.fileno(), count)
 
192
 
 
193
    def close(self):
 
194
        self.proc.stdin.close()
 
195
        self.proc.stdout.close()
 
196
        self.proc.wait()
 
197
 
 
198
 
 
199
class LoopbackSFTP(object):
 
200
    """Simple wrapper for a socket that pretends to be a paramiko Channel."""
 
201
 
 
202
    def __init__(self, sock):
 
203
        self.__socket = sock
 
204
 
 
205
    def send(self, data):
 
206
        return self.__socket.send(data)
 
207
 
 
208
    def recv(self, n):
 
209
        return self.__socket.recv(n)
 
210
 
 
211
    def recv_ready(self):
 
212
        return True
 
213
 
 
214
    def close(self):
 
215
        self.__socket.close()
 
216
 
 
217
 
 
218
SYSTEM_HOSTKEYS = {}
 
219
BZR_HOSTKEYS = {}
 
220
 
 
221
# This is a weakref dictionary, so that we can reuse connections
 
222
# that are still active. Long term, it might be nice to have some
 
223
# sort of expiration policy, such as disconnect if inactive for
 
224
# X seconds. But that requires a lot more fanciness.
 
225
_connected_hosts = weakref.WeakValueDictionary()
 
226
 
 
227
def clear_connection_cache():
 
228
    """Remove all hosts from the SFTP connection cache.
 
229
 
 
230
    Primarily useful for test cases wanting to force garbage collection.
 
231
    """
 
232
    _connected_hosts.clear()
 
233
 
 
234
 
 
235
def load_host_keys():
 
236
    """
 
237
    Load system host keys (probably doesn't work on windows) and any
 
238
    "discovered" keys from previous sessions.
 
239
    """
 
240
    global SYSTEM_HOSTKEYS, BZR_HOSTKEYS
 
241
    try:
 
242
        SYSTEM_HOSTKEYS = paramiko.util.load_host_keys(os.path.expanduser('~/.ssh/known_hosts'))
 
243
    except Exception, e:
 
244
        mutter('failed to load system host keys: ' + str(e))
 
245
    bzr_hostkey_path = pathjoin(config_dir(), 'ssh_host_keys')
 
246
    try:
 
247
        BZR_HOSTKEYS = paramiko.util.load_host_keys(bzr_hostkey_path)
 
248
    except Exception, e:
 
249
        mutter('failed to load bzr host keys: ' + str(e))
 
250
        save_host_keys()
 
251
 
 
252
 
 
253
def save_host_keys():
 
254
    """
 
255
    Save "discovered" host keys in $(config)/ssh_host_keys/.
 
256
    """
 
257
    global SYSTEM_HOSTKEYS, BZR_HOSTKEYS
 
258
    bzr_hostkey_path = pathjoin(config_dir(), 'ssh_host_keys')
 
259
    ensure_config_dir_exists()
 
260
 
 
261
    try:
 
262
        f = open(bzr_hostkey_path, 'w')
 
263
        f.write('# SSH host keys collected by bzr\n')
 
264
        for hostname, keys in BZR_HOSTKEYS.iteritems():
 
265
            for keytype, key in keys.iteritems():
 
266
                f.write('%s %s %s\n' % (hostname, keytype, key.get_base64()))
 
267
        f.close()
 
268
    except IOError, e:
 
269
        mutter('failed to save bzr host keys: ' + str(e))
95
270
 
96
271
 
97
272
class SFTPLock(object):
98
 
    """This fakes a lock in a remote location.
99
 
 
100
 
    A present lock is indicated just by the existence of a file.  This
101
 
    doesn't work well on all transports and they are only used in
102
 
    deprecated storage formats.
103
 
    """
104
 
 
 
273
    """This fakes a lock in a remote location."""
105
274
    __slots__ = ['path', 'lock_path', 'lock_file', 'transport']
106
 
 
107
275
    def __init__(self, path, transport):
 
276
        assert isinstance(transport, SFTPTransport)
 
277
 
108
278
        self.lock_file = None
109
279
        self.path = path
110
280
        self.lock_path = path + '.write-lock'
134
304
            pass
135
305
 
136
306
 
137
 
class _SFTPReadvHelper(object):
138
 
    """A class to help with managing the state of a readv request."""
139
 
 
140
 
    # See _get_requests for an explanation.
141
 
    _max_request_size = 32768
142
 
 
143
 
    def __init__(self, original_offsets, relpath, _report_activity):
144
 
        """Create a new readv helper.
145
 
 
146
 
        :param original_offsets: The original requests given by the caller of
147
 
            readv()
148
 
        :param relpath: The name of the file (if known)
149
 
        :param _report_activity: A Transport._report_activity bound method,
150
 
            to be called as data arrives.
151
 
        """
152
 
        self.original_offsets = list(original_offsets)
153
 
        self.relpath = relpath
154
 
        self._report_activity = _report_activity
155
 
 
156
 
    def _get_requests(self):
157
 
        """Break up the offsets into individual requests over sftp.
158
 
 
159
 
        The SFTP spec only requires implementers to support 32kB requests. We
160
 
        could try something larger (openssh supports 64kB), but then we have to
161
 
        handle requests that fail.
162
 
        So instead, we just break up our maximum chunks into 32kB chunks, and
163
 
        asyncronously requests them.
164
 
        Newer versions of paramiko would do the chunking for us, but we want to
165
 
        start processing results right away, so we do it ourselves.
166
 
        """
167
 
        # TODO: Because we issue async requests, we don't 'fudge' any extra
168
 
        #       data.  I'm not 100% sure that is the best choice.
169
 
 
170
 
        # The first thing we do, is to collapse the individual requests as much
171
 
        # as possible, so we don't issues requests <32kB
172
 
        sorted_offsets = sorted(self.original_offsets)
173
 
        coalesced = list(ConnectedTransport._coalesce_offsets(sorted_offsets,
174
 
                                                        limit=0, fudge_factor=0))
175
 
        requests = []
176
 
        for c_offset in coalesced:
177
 
            start = c_offset.start
178
 
            size = c_offset.length
179
 
 
180
 
            # Break this up into 32kB requests
181
 
            while size > 0:
182
 
                next_size = min(size, self._max_request_size)
183
 
                requests.append((start, next_size))
184
 
                size -= next_size
185
 
                start += next_size
186
 
        if 'sftp' in debug.debug_flags:
187
 
            mutter('SFTP.readv(%s) %s offsets => %s coalesced => %s requests',
188
 
                self.relpath, len(sorted_offsets), len(coalesced),
189
 
                len(requests))
190
 
        return requests
191
 
 
192
 
    def request_and_yield_offsets(self, fp):
193
 
        """Request the data from the remote machine, yielding the results.
194
 
 
195
 
        :param fp: A Paramiko SFTPFile object that supports readv.
196
 
        :return: Yield the data requested by the original readv caller, one by
197
 
            one.
198
 
        """
199
 
        requests = self._get_requests()
200
 
        offset_iter = iter(self.original_offsets)
201
 
        cur_offset, cur_size = offset_iter.next()
202
 
        # paramiko .readv() yields strings that are in the order of the requests
203
 
        # So we track the current request to know where the next data is
204
 
        # being returned from.
205
 
        input_start = None
206
 
        last_end = None
207
 
        buffered_data = []
208
 
        buffered_len = 0
209
 
 
210
 
        # This is used to buffer chunks which we couldn't process yet
211
 
        # It is (start, end, data) tuples.
212
 
        data_chunks = []
213
 
        # Create an 'unlimited' data stream, so we stop based on requests,
214
 
        # rather than just because the data stream ended. This lets us detect
215
 
        # short readv.
216
 
        data_stream = itertools.chain(fp.readv(requests),
217
 
                                      itertools.repeat(None))
218
 
        for (start, length), data in itertools.izip(requests, data_stream):
219
 
            if data is None:
220
 
                if cur_coalesced is not None:
221
 
                    raise errors.ShortReadvError(self.relpath,
222
 
                        start, length, len(data))
223
 
            if len(data) != length:
224
 
                raise errors.ShortReadvError(self.relpath,
225
 
                    start, length, len(data))
226
 
            self._report_activity(length, 'read')
227
 
            if last_end is None:
228
 
                # This is the first request, just buffer it
229
 
                buffered_data = [data]
230
 
                buffered_len = length
231
 
                input_start = start
232
 
            elif start == last_end:
233
 
                # The data we are reading fits neatly on the previous
234
 
                # buffer, so this is all part of a larger coalesced range.
235
 
                buffered_data.append(data)
236
 
                buffered_len += length
237
 
            else:
238
 
                # We have an 'interrupt' in the data stream. So we know we are
239
 
                # at a request boundary.
240
 
                if buffered_len > 0:
241
 
                    # We haven't consumed the buffer so far, so put it into
242
 
                    # data_chunks, and continue.
243
 
                    buffered = ''.join(buffered_data)
244
 
                    data_chunks.append((input_start, buffered))
245
 
                input_start = start
246
 
                buffered_data = [data]
247
 
                buffered_len = length
248
 
            last_end = start + length
249
 
            if input_start == cur_offset and cur_size <= buffered_len:
250
 
                # Simplify the next steps a bit by transforming buffered_data
251
 
                # into a single string. We also have the nice property that
252
 
                # when there is only one string ''.join([x]) == x, so there is
253
 
                # no data copying.
254
 
                buffered = ''.join(buffered_data)
255
 
                # Clean out buffered data so that we keep memory
256
 
                # consumption low
257
 
                del buffered_data[:]
258
 
                buffered_offset = 0
259
 
                # TODO: We *could* also consider the case where cur_offset is in
260
 
                #       in the buffered range, even though it doesn't *start*
261
 
                #       the buffered range. But for packs we pretty much always
262
 
                #       read in order, so you won't get any extra data in the
263
 
                #       middle.
264
 
                while (input_start == cur_offset
265
 
                       and (buffered_offset + cur_size) <= buffered_len):
266
 
                    # We've buffered enough data to process this request, spit it
267
 
                    # out
268
 
                    cur_data = buffered[buffered_offset:buffered_offset + cur_size]
269
 
                    # move the direct pointer into our buffered data
270
 
                    buffered_offset += cur_size
271
 
                    # Move the start-of-buffer pointer
272
 
                    input_start += cur_size
273
 
                    # Yield the requested data
274
 
                    yield cur_offset, cur_data
275
 
                    cur_offset, cur_size = offset_iter.next()
276
 
                # at this point, we've consumed as much of buffered as we can,
277
 
                # so break off the portion that we consumed
278
 
                if buffered_offset == len(buffered_data):
279
 
                    # No tail to leave behind
280
 
                    buffered_data = []
281
 
                    buffered_len = 0
282
 
                else:
283
 
                    buffered = buffered[buffered_offset:]
284
 
                    buffered_data = [buffered]
285
 
                    buffered_len = len(buffered)
286
 
        if buffered_len:
287
 
            buffered = ''.join(buffered_data)
288
 
            del buffered_data[:]
289
 
            data_chunks.append((input_start, buffered))
290
 
        if data_chunks:
291
 
            if 'sftp' in debug.debug_flags:
292
 
                mutter('SFTP readv left with %d out-of-order bytes',
293
 
                    sum(map(lambda x: len(x[1]), data_chunks)))
294
 
            # We've processed all the readv data, at this point, anything we
295
 
            # couldn't process is in data_chunks. This doesn't happen often, so
296
 
            # this code path isn't optimized
297
 
            # We use an interesting process for data_chunks
298
 
            # Specifically if we have "bisect_left([(start, len, entries)],
299
 
            #                                       (qstart,)])
300
 
            # If start == qstart, then we get the specific node. Otherwise we
301
 
            # get the previous node
302
 
            while True:
303
 
                idx = bisect.bisect_left(data_chunks, (cur_offset,))
304
 
                if idx < len(data_chunks) and data_chunks[idx][0] == cur_offset:
305
 
                    # The data starts here
306
 
                    data = data_chunks[idx][1][:cur_size]
307
 
                elif idx > 0:
308
 
                    # The data is in a portion of a previous page
309
 
                    idx -= 1
310
 
                    sub_offset = cur_offset - data_chunks[idx][0]
311
 
                    data = data_chunks[idx][1]
312
 
                    data = data[sub_offset:sub_offset + cur_size]
313
 
                else:
314
 
                    # We are missing the page where the data should be found,
315
 
                    # something is wrong
316
 
                    data = ''
317
 
                if len(data) != cur_size:
318
 
                    raise AssertionError('We must have miscalulated.'
319
 
                        ' We expected %d bytes, but only found %d'
320
 
                        % (cur_size, len(data)))
321
 
                yield cur_offset, data
322
 
                cur_offset, cur_size = offset_iter.next()
323
 
 
324
 
 
325
 
class SFTPTransport(ConnectedTransport):
326
 
    """Transport implementation for SFTP access."""
327
 
 
 
307
class SFTPTransport (Transport):
 
308
    """
 
309
    Transport implementation for SFTP access.
 
310
    """
328
311
    _do_prefetch = _default_do_prefetch
329
 
    # TODO: jam 20060717 Conceivably these could be configurable, either
330
 
    #       by auto-tuning at run-time, or by a configuration (per host??)
331
 
    #       but the performance curve is pretty flat, so just going with
332
 
    #       reasonable defaults.
333
 
    _max_readv_combine = 200
334
 
    # Having to round trip to the server means waiting for a response,
335
 
    # so it is better to download extra bytes.
336
 
    # 8KiB had good performance for both local and remote network operations
337
 
    _bytes_to_read_before_seek = 8192
338
 
 
339
 
    # The sftp spec says that implementations SHOULD allow reads
340
 
    # to be at least 32K. paramiko.readv() does an async request
341
 
    # for the chunks. So we need to keep it within a single request
342
 
    # size for paramiko <= 1.6.1. paramiko 1.6.2 will probably chop
343
 
    # up the request itself, rather than us having to worry about it
344
 
    _max_request_size = 32768
345
 
 
346
 
    def __init__(self, base, _from_transport=None):
347
 
        super(SFTPTransport, self).__init__(base,
348
 
                                            _from_transport=_from_transport)
349
 
 
 
312
 
 
313
    def __init__(self, base, clone_from=None):
 
314
        assert base.startswith('sftp://')
 
315
        self._parse_url(base)
 
316
        base = self._unparse_url()
 
317
        if base[-1] != '/':
 
318
            base += '/'
 
319
        super(SFTPTransport, self).__init__(base)
 
320
        if clone_from is None:
 
321
            self._sftp_connect()
 
322
        else:
 
323
            # use the same ssh connection, etc
 
324
            self._sftp = clone_from._sftp
 
325
        # super saves 'self.base'
 
326
    
 
327
    def should_cache(self):
 
328
        """
 
329
        Return True if the data pulled across should be cached locally.
 
330
        """
 
331
        return True
 
332
 
 
333
    def clone(self, offset=None):
 
334
        """
 
335
        Return a new SFTPTransport with root at self.base + offset.
 
336
        We share the same SFTP session between such transports, because it's
 
337
        fairly expensive to set them up.
 
338
        """
 
339
        if offset is None:
 
340
            return SFTPTransport(self.base, self)
 
341
        else:
 
342
            return SFTPTransport(self.abspath(offset), self)
 
343
 
 
344
    def abspath(self, relpath):
 
345
        """
 
346
        Return the full url to the given relative path.
 
347
        
 
348
        @param relpath: the relative path or path components
 
349
        @type relpath: str or list
 
350
        """
 
351
        return self._unparse_url(self._remote_path(relpath))
 
352
    
350
353
    def _remote_path(self, relpath):
351
354
        """Return the path to be passed along the sftp protocol for relpath.
352
 
 
353
 
        :param relpath: is a urlencoded string.
354
 
        """
355
 
        relative = urlutils.unescape(relpath).encode('utf-8')
356
 
        remote_path = self._combine_paths(self._path, relative)
357
 
        # the initial slash should be removed from the path, and treated as a
358
 
        # homedir relative path (the path begins with a double slash if it is
359
 
        # absolute).  see draft-ietf-secsh-scp-sftp-ssh-uri-03.txt
360
 
        # RBC 20060118 we are not using this as its too user hostile. instead
361
 
        # we are following lftp and using /~/foo to mean '~/foo'
362
 
        # vila--20070602 and leave absolute paths begin with a single slash.
363
 
        if remote_path.startswith('/~/'):
364
 
            remote_path = remote_path[3:]
365
 
        elif remote_path == '/~':
366
 
            remote_path = ''
367
 
        return remote_path
368
 
 
369
 
    def _create_connection(self, credentials=None):
370
 
        """Create a new connection with the provided credentials.
371
 
 
372
 
        :param credentials: The credentials needed to establish the connection.
373
 
 
374
 
        :return: The created connection and its associated credentials.
375
 
 
376
 
        The credentials are only the password as it may have been entered
377
 
        interactively by the user and may be different from the one provided
378
 
        in base url at transport creation time.
379
 
        """
380
 
        if credentials is None:
381
 
            password = self._password
382
 
        else:
383
 
            password = credentials
384
 
 
385
 
        vendor = ssh._get_ssh_vendor()
386
 
        user = self._user
387
 
        if user is None:
388
 
            auth = config.AuthenticationConfig()
389
 
            user = auth.get_user('ssh', self._host, self._port)
390
 
        connection = vendor.connect_sftp(self._user, password,
391
 
                                         self._host, self._port)
392
 
        return connection, (user, password)
393
 
 
394
 
    def _get_sftp(self):
395
 
        """Ensures that a connection is established"""
396
 
        connection = self._get_connection()
397
 
        if connection is None:
398
 
            # First connection ever
399
 
            connection, credentials = self._create_connection()
400
 
            self._set_connection(connection, credentials)
401
 
        return connection
 
355
        
 
356
        relpath is a urlencoded string.
 
357
        """
 
358
        # FIXME: share the common code across transports
 
359
        assert isinstance(relpath, basestring)
 
360
        relpath = urlutils.unescape(relpath).split('/')
 
361
        basepath = self._path.split('/')
 
362
        if len(basepath) > 0 and basepath[-1] == '':
 
363
            basepath = basepath[:-1]
 
364
 
 
365
        for p in relpath:
 
366
            if p == '..':
 
367
                if len(basepath) == 0:
 
368
                    # In most filesystems, a request for the parent
 
369
                    # of root, just returns root.
 
370
                    continue
 
371
                basepath.pop()
 
372
            elif p == '.':
 
373
                continue # No-op
 
374
            else:
 
375
                basepath.append(p)
 
376
 
 
377
        path = '/'.join(basepath)
 
378
        return path
 
379
 
 
380
    def relpath(self, abspath):
 
381
        username, password, host, port, path = self._split_url(abspath)
 
382
        error = []
 
383
        if (username != self._username):
 
384
            error.append('username mismatch')
 
385
        if (host != self._host):
 
386
            error.append('host mismatch')
 
387
        if (port != self._port):
 
388
            error.append('port mismatch')
 
389
        if (not path.startswith(self._path)):
 
390
            error.append('path mismatch')
 
391
        if error:
 
392
            extra = ': ' + ', '.join(error)
 
393
            raise PathNotChild(abspath, self.base, extra=extra)
 
394
        pl = len(self._path)
 
395
        return path[pl:].strip('/')
402
396
 
403
397
    def has(self, relpath):
404
398
        """
405
399
        Does the target location exist?
406
400
        """
407
401
        try:
408
 
            self._get_sftp().stat(self._remote_path(relpath))
409
 
            # stat result is about 20 bytes, let's say
410
 
            self._report_activity(20, 'read')
 
402
            self._sftp.stat(self._remote_path(relpath))
411
403
            return True
412
404
        except IOError:
413
405
            return False
414
406
 
415
407
    def get(self, relpath):
416
 
        """Get the file at the given relative path.
 
408
        """
 
409
        Get the file at the given relative path.
417
410
 
418
411
        :param relpath: The relative path to the file
419
412
        """
420
413
        try:
421
 
            # FIXME: by returning the file directly, we don't pass this
422
 
            # through to report_activity.  We could try wrapping the object
423
 
            # before it's returned.  For readv and get_bytes it's handled in
424
 
            # the higher-level function.
425
 
            # -- mbp 20090126
426
414
            path = self._remote_path(relpath)
427
 
            f = self._get_sftp().file(path, mode='rb')
 
415
            f = self._sftp.file(path, mode='rb')
428
416
            if self._do_prefetch and (getattr(f, 'prefetch', None) is not None):
429
417
                f.prefetch()
430
418
            return f
431
419
        except (IOError, paramiko.SSHException), e:
432
 
            self._translate_io_exception(e, path, ': error retrieving',
433
 
                failure_exc=errors.ReadError)
434
 
 
435
 
    def get_bytes(self, relpath):
436
 
        # reimplement this here so that we can report how many bytes came back
 
420
            self._translate_io_exception(e, path, ': error retrieving')
 
421
 
 
422
    def get_partial(self, relpath, start, length=None):
 
423
        """
 
424
        Get just part of a file.
 
425
 
 
426
        :param relpath: Path to the file, relative to base
 
427
        :param start: The starting position to read from
 
428
        :param length: The length to read. A length of None indicates
 
429
                       read to the end of the file.
 
430
        :return: A file-like object containing at least the specified bytes.
 
431
                 Some implementations may return objects which can be read
 
432
                 past this length, but this is not guaranteed.
 
433
        """
 
434
        # TODO: implement get_partial_multi to help with knit support
437
435
        f = self.get(relpath)
438
 
        try:
439
 
            bytes = f.read()
440
 
            self._report_activity(len(bytes), 'read')
441
 
            return bytes
442
 
        finally:
443
 
            f.close()
444
 
 
445
 
    def _readv(self, relpath, offsets):
446
 
        """See Transport.readv()"""
447
 
        # We overload the default readv() because we want to use a file
448
 
        # that does not have prefetch enabled.
449
 
        # Also, if we have a new paramiko, it implements an async readv()
450
 
        if not offsets:
451
 
            return
452
 
 
453
 
        try:
454
 
            path = self._remote_path(relpath)
455
 
            fp = self._get_sftp().file(path, mode='rb')
456
 
            readv = getattr(fp, 'readv', None)
457
 
            if readv:
458
 
                return self._sftp_readv(fp, offsets, relpath)
459
 
            if 'sftp' in debug.debug_flags:
460
 
                mutter('seek and read %s offsets', len(offsets))
461
 
            return self._seek_and_read(fp, offsets, relpath)
462
 
        except (IOError, paramiko.SSHException), e:
463
 
            self._translate_io_exception(e, path, ': error retrieving')
464
 
 
465
 
    def recommended_page_size(self):
466
 
        """See Transport.recommended_page_size().
467
 
 
468
 
        For SFTP we suggest a large page size to reduce the overhead
469
 
        introduced by latency.
470
 
        """
471
 
        return 64 * 1024
472
 
 
473
 
    def _sftp_readv(self, fp, offsets, relpath):
474
 
        """Use the readv() member of fp to do async readv.
475
 
 
476
 
        Then read them using paramiko.readv(). paramiko.readv()
477
 
        does not support ranges > 64K, so it caps the request size, and
478
 
        just reads until it gets all the stuff it wants.
479
 
        """
480
 
        helper = _SFTPReadvHelper(offsets, relpath, self._report_activity)
481
 
        return helper.request_and_yield_offsets(fp)
482
 
 
483
 
    def put_file(self, relpath, f, mode=None):
484
 
        """
485
 
        Copy the file-like object into the location.
 
436
        f.seek(start)
 
437
        if self._do_prefetch and hasattr(f, 'prefetch'):
 
438
            f.prefetch()
 
439
        return f
 
440
 
 
441
    def put(self, relpath, f, mode=None):
 
442
        """
 
443
        Copy the file-like or string object into the location.
486
444
 
487
445
        :param relpath: Location to put the contents, relative to base.
488
 
        :param f:       File-like object.
 
446
        :param f:       File-like or string object.
489
447
        :param mode: The final mode for the file
490
448
        """
491
449
        final_path = self._remote_path(relpath)
492
 
        return self._put(final_path, f, mode=mode)
 
450
        self._put(final_path, f, mode=mode)
493
451
 
494
452
    def _put(self, abspath, f, mode=None):
495
453
        """Helper function so both put() and copy_abspaths can reuse the code"""
500
458
        try:
501
459
            try:
502
460
                fout.set_pipelined(True)
503
 
                length = self._pump(f, fout)
 
461
                self._pump(f, fout)
504
462
            except (IOError, paramiko.SSHException), e:
505
463
                self._translate_io_exception(e, tmp_abspath)
506
 
            # XXX: This doesn't truly help like we would like it to.
507
 
            #      The problem is that openssh strips sticky bits. So while we
508
 
            #      can properly set group write permission, we lose the group
509
 
            #      sticky bit. So it is probably best to stop chmodding, and
510
 
            #      just tell users that they need to set the umask correctly.
511
 
            #      The attr.st_mode = mode, in _sftp_open_exclusive
512
 
            #      will handle when the user wants the final mode to be more
513
 
            #      restrictive. And then we avoid a round trip. Unless
514
 
            #      paramiko decides to expose an async chmod()
515
 
 
516
 
            # This is designed to chmod() right before we close.
517
 
            # Because we set_pipelined() earlier, theoretically we might
518
 
            # avoid the round trip for fout.close()
519
464
            if mode is not None:
520
 
                self._get_sftp().chmod(tmp_abspath, mode)
 
465
                self._sftp.chmod(tmp_abspath, mode)
521
466
            fout.close()
522
467
            closed = True
523
468
            self._rename_and_overwrite(tmp_abspath, abspath)
524
 
            return length
525
469
        except Exception, e:
526
470
            # If we fail, try to clean up the temporary file
527
471
            # before we throw the exception
533
477
            try:
534
478
                if not closed:
535
479
                    fout.close()
536
 
                self._get_sftp().remove(tmp_abspath)
 
480
                self._sftp.remove(tmp_abspath)
537
481
            except:
538
482
                # raise the saved except
539
483
                raise e
540
484
            # raise the original with its traceback if we can.
541
485
            raise
542
486
 
543
 
    def _put_non_atomic_helper(self, relpath, writer, mode=None,
544
 
                               create_parent_dir=False,
545
 
                               dir_mode=None):
546
 
        abspath = self._remote_path(relpath)
547
 
 
548
 
        # TODO: jam 20060816 paramiko doesn't publicly expose a way to
549
 
        #       set the file mode at create time. If it does, use it.
550
 
        #       But for now, we just chmod later anyway.
551
 
 
552
 
        def _open_and_write_file():
553
 
            """Try to open the target file, raise error on failure"""
554
 
            fout = None
555
 
            try:
556
 
                try:
557
 
                    fout = self._get_sftp().file(abspath, mode='wb')
558
 
                    fout.set_pipelined(True)
559
 
                    writer(fout)
560
 
                except (paramiko.SSHException, IOError), e:
561
 
                    self._translate_io_exception(e, abspath,
562
 
                                                 ': unable to open')
563
 
 
564
 
                # This is designed to chmod() right before we close.
565
 
                # Because we set_pipelined() earlier, theoretically we might
566
 
                # avoid the round trip for fout.close()
567
 
                if mode is not None:
568
 
                    self._get_sftp().chmod(abspath, mode)
569
 
            finally:
570
 
                if fout is not None:
571
 
                    fout.close()
572
 
 
573
 
        if not create_parent_dir:
574
 
            _open_and_write_file()
575
 
            return
576
 
 
577
 
        # Try error handling to create the parent directory if we need to
578
 
        try:
579
 
            _open_and_write_file()
580
 
        except NoSuchFile:
581
 
            # Try to create the parent directory, and then go back to
582
 
            # writing the file
583
 
            parent_dir = os.path.dirname(abspath)
584
 
            self._mkdir(parent_dir, dir_mode)
585
 
            _open_and_write_file()
586
 
 
587
 
    def put_file_non_atomic(self, relpath, f, mode=None,
588
 
                            create_parent_dir=False,
589
 
                            dir_mode=None):
590
 
        """Copy the file-like object into the target location.
591
 
 
592
 
        This function is not strictly safe to use. It is only meant to
593
 
        be used when you already know that the target does not exist.
594
 
        It is not safe, because it will open and truncate the remote
595
 
        file. So there may be a time when the file has invalid contents.
596
 
 
597
 
        :param relpath: The remote location to put the contents.
598
 
        :param f:       File-like object.
599
 
        :param mode:    Possible access permissions for new file.
600
 
                        None means do not set remote permissions.
601
 
        :param create_parent_dir: If we cannot create the target file because
602
 
                        the parent directory does not exist, go ahead and
603
 
                        create it, and then try again.
604
 
        """
605
 
        def writer(fout):
606
 
            self._pump(f, fout)
607
 
        self._put_non_atomic_helper(relpath, writer, mode=mode,
608
 
                                    create_parent_dir=create_parent_dir,
609
 
                                    dir_mode=dir_mode)
610
 
 
611
 
    def put_bytes_non_atomic(self, relpath, bytes, mode=None,
612
 
                             create_parent_dir=False,
613
 
                             dir_mode=None):
614
 
        def writer(fout):
615
 
            fout.write(bytes)
616
 
        self._put_non_atomic_helper(relpath, writer, mode=mode,
617
 
                                    create_parent_dir=create_parent_dir,
618
 
                                    dir_mode=dir_mode)
619
 
 
620
487
    def iter_files_recursive(self):
621
488
        """Walk the relative paths of all files in this transport."""
622
 
        # progress is handled by list_dir
623
489
        queue = list(self.list_dir('.'))
624
490
        while queue:
625
 
            relpath = queue.pop(0)
 
491
            relpath = urllib.quote(queue.pop(0))
626
492
            st = self.stat(relpath)
627
493
            if stat.S_ISDIR(st.st_mode):
628
494
                for i, basename in enumerate(self.list_dir(relpath)):
630
496
            else:
631
497
                yield relpath
632
498
 
633
 
    def _mkdir(self, abspath, mode=None):
634
 
        if mode is None:
635
 
            local_mode = 0777
636
 
        else:
637
 
            local_mode = mode
638
 
        try:
639
 
            self._report_activity(len(abspath), 'write')
640
 
            self._get_sftp().mkdir(abspath, local_mode)
641
 
            self._report_activity(1, 'read')
642
 
            if mode is not None:
643
 
                # chmod a dir through sftp will erase any sgid bit set
644
 
                # on the server side.  So, if the bit mode are already
645
 
                # set, avoid the chmod.  If the mode is not fine but
646
 
                # the sgid bit is set, report a warning to the user
647
 
                # with the umask fix.
648
 
                stat = self._get_sftp().lstat(abspath)
649
 
                mode = mode & 0777 # can't set special bits anyway
650
 
                if mode != stat.st_mode & 0777:
651
 
                    if stat.st_mode & 06000:
652
 
                        warning('About to chmod %s over sftp, which will result'
653
 
                                ' in its suid or sgid bits being cleared.  If'
654
 
                                ' you want to preserve those bits, change your '
655
 
                                ' environment on the server to use umask 0%03o.'
656
 
                                % (abspath, 0777 - mode))
657
 
                    self._get_sftp().chmod(abspath, mode=mode)
658
 
        except (paramiko.SSHException, IOError), e:
659
 
            self._translate_io_exception(e, abspath, ': unable to mkdir',
660
 
                failure_exc=FileExists)
661
 
 
662
499
    def mkdir(self, relpath, mode=None):
663
500
        """Create a directory at the given path."""
664
 
        self._mkdir(self._remote_path(relpath), mode=mode)
665
 
 
666
 
    def open_write_stream(self, relpath, mode=None):
667
 
        """See Transport.open_write_stream."""
668
 
        # initialise the file to zero-length
669
 
        # this is three round trips, but we don't use this
670
 
        # api more than once per write_group at the moment so
671
 
        # it is a tolerable overhead. Better would be to truncate
672
 
        # the file after opening. RBC 20070805
673
 
        self.put_bytes_non_atomic(relpath, "", mode)
674
 
        abspath = self._remote_path(relpath)
675
 
        # TODO: jam 20060816 paramiko doesn't publicly expose a way to
676
 
        #       set the file mode at create time. If it does, use it.
677
 
        #       But for now, we just chmod later anyway.
678
 
        handle = None
679
501
        try:
680
 
            handle = self._get_sftp().file(abspath, mode='wb')
681
 
            handle.set_pipelined(True)
 
502
            path = self._remote_path(relpath)
 
503
            # In the paramiko documentation, it says that passing a mode flag 
 
504
            # will filtered against the server umask.
 
505
            # StubSFTPServer does not do this, which would be nice, because it is
 
506
            # what we really want :)
 
507
            # However, real servers do use umask, so we really should do it that way
 
508
            self._sftp.mkdir(path)
 
509
            if mode is not None:
 
510
                self._sftp.chmod(path, mode=mode)
682
511
        except (paramiko.SSHException, IOError), e:
683
 
            self._translate_io_exception(e, abspath,
684
 
                                         ': unable to open')
685
 
        _file_streams[self.abspath(relpath)] = handle
686
 
        return FileFileStream(self, relpath, handle)
 
512
            self._translate_io_exception(e, path, ': unable to mkdir',
 
513
                failure_exc=FileExists)
687
514
 
688
 
    def _translate_io_exception(self, e, path, more_info='',
 
515
    def _translate_io_exception(self, e, path, more_info='', 
689
516
                                failure_exc=PathError):
690
517
        """Translate a paramiko or IOError into a friendlier exception.
691
518
 
696
523
        :param failure_exc: Paramiko has the super fun ability to raise completely
697
524
                           opaque errors that just set "e.args = ('Failure',)" with
698
525
                           no more information.
699
 
                           If this parameter is set, it defines the exception
 
526
                           If this parameter is set, it defines the exception 
700
527
                           to raise in these cases.
701
528
        """
702
529
        # paramiko seems to generate detailless errors.
703
530
        self._translate_error(e, path, raise_generic=False)
704
 
        if getattr(e, 'args', None) is not None:
 
531
        if hasattr(e, 'args'):
705
532
            if (e.args == ('No such file or directory',) or
706
533
                e.args == ('No such file',)):
707
534
                raise NoSuchFile(path, str(e) + more_info)
708
 
            if (e.args == ('mkdir failed',) or
709
 
                e.args[0].startswith('syserr: File exists')):
 
535
            if (e.args == ('mkdir failed',)):
710
536
                raise FileExists(path, str(e) + more_info)
711
537
            # strange but true, for the paramiko server.
712
538
            if (e.args == ('Failure',)):
713
539
                raise failure_exc(path, str(e) + more_info)
714
 
            # Can be something like args = ('Directory not empty:
715
 
            # '/srv/bazaar.launchpad.net/blah...: '
716
 
            # [Errno 39] Directory not empty',)
717
 
            if (e.args[0].startswith('Directory not empty: ')
718
 
                or getattr(e, 'errno', None) == errno.ENOTEMPTY):
719
 
                raise errors.DirectoryNotEmpty(path, str(e))
720
540
            mutter('Raising exception with args %s', e.args)
721
 
        if getattr(e, 'errno', None) is not None:
 
541
        if hasattr(e, 'errno'):
722
542
            mutter('Raising exception with errno %s', e.errno)
723
543
        raise e
724
544
 
725
 
    def append_file(self, relpath, f, mode=None):
 
545
    def append(self, relpath, f, mode=None):
726
546
        """
727
547
        Append the text in the file-like object into the final
728
548
        location.
729
549
        """
730
550
        try:
731
551
            path = self._remote_path(relpath)
732
 
            fout = self._get_sftp().file(path, 'ab')
 
552
            fout = self._sftp.file(path, 'ab')
733
553
            if mode is not None:
734
 
                self._get_sftp().chmod(path, mode)
 
554
                self._sftp.chmod(path, mode)
735
555
            result = fout.tell()
736
556
            self._pump(f, fout)
737
557
            return result
741
561
    def rename(self, rel_from, rel_to):
742
562
        """Rename without special overwriting"""
743
563
        try:
744
 
            self._get_sftp().rename(self._remote_path(rel_from),
 
564
            self._sftp.rename(self._remote_path(rel_from),
745
565
                              self._remote_path(rel_to))
746
566
        except (IOError, paramiko.SSHException), e:
747
567
            self._translate_io_exception(e, rel_from,
749
569
 
750
570
    def _rename_and_overwrite(self, abs_from, abs_to):
751
571
        """Do a fancy rename on the remote server.
752
 
 
 
572
        
753
573
        Using the implementation provided by osutils.
754
574
        """
755
575
        try:
756
 
            sftp = self._get_sftp()
757
576
            fancy_rename(abs_from, abs_to,
758
 
                         rename_func=sftp.rename,
759
 
                         unlink_func=sftp.remove)
 
577
                    rename_func=self._sftp.rename,
 
578
                    unlink_func=self._sftp.remove)
760
579
        except (IOError, paramiko.SSHException), e:
761
 
            self._translate_io_exception(e, abs_from,
762
 
                                         ': unable to rename to %r' % (abs_to))
 
580
            self._translate_io_exception(e, abs_from, ': unable to rename to %r' % (abs_to))
763
581
 
764
582
    def move(self, rel_from, rel_to):
765
583
        """Move the item at rel_from to the location at rel_to"""
771
589
        """Delete the item at relpath"""
772
590
        path = self._remote_path(relpath)
773
591
        try:
774
 
            self._get_sftp().remove(path)
 
592
            self._sftp.remove(path)
775
593
        except (IOError, paramiko.SSHException), e:
776
594
            self._translate_io_exception(e, path, ': unable to delete')
777
 
 
778
 
    def external_url(self):
779
 
        """See bzrlib.transport.Transport.external_url."""
780
 
        # the external path for SFTP is the base
781
 
        return self.base
782
 
 
 
595
            
783
596
    def listable(self):
784
597
        """Return True if this store supports listing."""
785
598
        return True
789
602
        Return a list of all files at the given location.
790
603
        """
791
604
        # does anything actually use this?
792
 
        # -- Unknown
793
 
        # This is at least used by copy_tree for remote upgrades.
794
 
        # -- David Allouche 2006-08-11
795
605
        path = self._remote_path(relpath)
796
606
        try:
797
 
            entries = self._get_sftp().listdir(path)
798
 
            self._report_activity(sum(map(len, entries)), 'read')
 
607
            return self._sftp.listdir(path)
799
608
        except (IOError, paramiko.SSHException), e:
800
609
            self._translate_io_exception(e, path, ': failed to list_dir')
801
 
        return [urlutils.escape(entry) for entry in entries]
802
610
 
803
611
    def rmdir(self, relpath):
804
612
        """See Transport.rmdir."""
805
613
        path = self._remote_path(relpath)
806
614
        try:
807
 
            return self._get_sftp().rmdir(path)
 
615
            return self._sftp.rmdir(path)
808
616
        except (IOError, paramiko.SSHException), e:
809
617
            self._translate_io_exception(e, path, ': failed to rmdir')
810
618
 
812
620
        """Return the stat information for a file."""
813
621
        path = self._remote_path(relpath)
814
622
        try:
815
 
            return self._get_sftp().stat(path)
 
623
            return self._sftp.stat(path)
816
624
        except (IOError, paramiko.SSHException), e:
817
625
            self._translate_io_exception(e, path, ': unable to stat')
818
626
 
842
650
        # that we have taken the lock.
843
651
        return SFTPLock(relpath, self)
844
652
 
 
653
    def _unparse_url(self, path=None):
 
654
        if path is None:
 
655
            path = self._path
 
656
        path = urllib.quote(path)
 
657
        # handle homedir paths
 
658
        if not path.startswith('/'):
 
659
            path = "/~/" + path
 
660
        netloc = urllib.quote(self._host)
 
661
        if self._username is not None:
 
662
            netloc = '%s@%s' % (urllib.quote(self._username), netloc)
 
663
        if self._port is not None:
 
664
            netloc = '%s:%d' % (netloc, self._port)
 
665
        return urlparse.urlunparse(('sftp', netloc, path, '', '', ''))
 
666
 
 
667
    def _split_url(self, url):
 
668
        (scheme, username, password, host, port, path) = split_url(url)
 
669
        assert scheme == 'sftp'
 
670
 
 
671
        # the initial slash should be removed from the path, and treated
 
672
        # as a homedir relative path (the path begins with a double slash
 
673
        # if it is absolute).
 
674
        # see draft-ietf-secsh-scp-sftp-ssh-uri-03.txt
 
675
        # RBC 20060118 we are not using this as its too user hostile. instead
 
676
        # we are following lftp and using /~/foo to mean '~/foo'.
 
677
        # handle homedir paths
 
678
        if path.startswith('/~/'):
 
679
            path = path[3:]
 
680
        elif path == '/~':
 
681
            path = ''
 
682
        return (username, password, host, port, path)
 
683
 
 
684
    def _parse_url(self, url):
 
685
        (self._username, self._password,
 
686
         self._host, self._port, self._path) = self._split_url(url)
 
687
 
 
688
    def _sftp_connect(self):
 
689
        """Connect to the remote sftp server.
 
690
        After this, self._sftp should have a valid connection (or
 
691
        we raise an TransportError 'could not connect').
 
692
 
 
693
        TODO: Raise a more reasonable ConnectionFailed exception
 
694
        """
 
695
        global _connected_hosts
 
696
 
 
697
        idx = (self._host, self._port, self._username)
 
698
        try:
 
699
            self._sftp = _connected_hosts[idx]
 
700
            return
 
701
        except KeyError:
 
702
            pass
 
703
        
 
704
        vendor = _get_ssh_vendor()
 
705
        if vendor == 'loopback':
 
706
            sock = socket.socket()
 
707
            try:
 
708
                sock.connect((self._host, self._port))
 
709
            except socket.error, e:
 
710
                raise ConnectionError('Unable to connect to SSH host %s:%s: %s'
 
711
                                      % (self._host, self._port, e))
 
712
            self._sftp = SFTPClient(LoopbackSFTP(sock))
 
713
        elif vendor != 'none':
 
714
            sock = SFTPSubprocess(self._host, vendor, self._port,
 
715
                                  self._username)
 
716
            self._sftp = SFTPClient(sock)
 
717
        else:
 
718
            self._paramiko_connect()
 
719
 
 
720
        _connected_hosts[idx] = self._sftp
 
721
 
 
722
    def _paramiko_connect(self):
 
723
        global SYSTEM_HOSTKEYS, BZR_HOSTKEYS
 
724
        
 
725
        load_host_keys()
 
726
 
 
727
        try:
 
728
            t = paramiko.Transport((self._host, self._port or 22))
 
729
            t.set_log_channel('bzr.paramiko')
 
730
            t.start_client()
 
731
        except paramiko.SSHException, e:
 
732
            raise ConnectionError('Unable to reach SSH host %s:%s: %s' 
 
733
                                  % (self._host, self._port, e))
 
734
            
 
735
        server_key = t.get_remote_server_key()
 
736
        server_key_hex = paramiko.util.hexify(server_key.get_fingerprint())
 
737
        keytype = server_key.get_name()
 
738
        if SYSTEM_HOSTKEYS.has_key(self._host) and SYSTEM_HOSTKEYS[self._host].has_key(keytype):
 
739
            our_server_key = SYSTEM_HOSTKEYS[self._host][keytype]
 
740
            our_server_key_hex = paramiko.util.hexify(our_server_key.get_fingerprint())
 
741
        elif BZR_HOSTKEYS.has_key(self._host) and BZR_HOSTKEYS[self._host].has_key(keytype):
 
742
            our_server_key = BZR_HOSTKEYS[self._host][keytype]
 
743
            our_server_key_hex = paramiko.util.hexify(our_server_key.get_fingerprint())
 
744
        else:
 
745
            warning('Adding %s host key for %s: %s' % (keytype, self._host, server_key_hex))
 
746
            if not BZR_HOSTKEYS.has_key(self._host):
 
747
                BZR_HOSTKEYS[self._host] = {}
 
748
            BZR_HOSTKEYS[self._host][keytype] = server_key
 
749
            our_server_key = server_key
 
750
            our_server_key_hex = paramiko.util.hexify(our_server_key.get_fingerprint())
 
751
            save_host_keys()
 
752
        if server_key != our_server_key:
 
753
            filename1 = os.path.expanduser('~/.ssh/known_hosts')
 
754
            filename2 = pathjoin(config_dir(), 'ssh_host_keys')
 
755
            raise TransportError('Host keys for %s do not match!  %s != %s' % \
 
756
                (self._host, our_server_key_hex, server_key_hex),
 
757
                ['Try editing %s or %s' % (filename1, filename2)])
 
758
 
 
759
        self._sftp_auth(t)
 
760
        
 
761
        try:
 
762
            self._sftp = t.open_sftp_client()
 
763
        except paramiko.SSHException, e:
 
764
            raise ConnectionError('Unable to start sftp client %s:%d' %
 
765
                                  (self._host, self._port), e)
 
766
 
 
767
    def _sftp_auth(self, transport):
 
768
        # paramiko requires a username, but it might be none if nothing was supplied
 
769
        # use the local username, just in case.
 
770
        # We don't override self._username, because if we aren't using paramiko,
 
771
        # the username might be specified in ~/.ssh/config and we don't want to
 
772
        # force it to something else
 
773
        # Also, it would mess up the self.relpath() functionality
 
774
        username = self._username or getpass.getuser()
 
775
 
 
776
        # Paramiko tries to open a socket.AF_UNIX in order to connect
 
777
        # to ssh-agent. That attribute doesn't exist on win32 (it does in cygwin)
 
778
        # so we get an AttributeError exception. For now, just don't try to
 
779
        # connect to an agent if we are on win32
 
780
        if sys.platform != 'win32':
 
781
            agent = paramiko.Agent()
 
782
            for key in agent.get_keys():
 
783
                mutter('Trying SSH agent key %s' % paramiko.util.hexify(key.get_fingerprint()))
 
784
                try:
 
785
                    transport.auth_publickey(username, key)
 
786
                    return
 
787
                except paramiko.SSHException, e:
 
788
                    pass
 
789
        
 
790
        # okay, try finding id_rsa or id_dss?  (posix only)
 
791
        if self._try_pkey_auth(transport, paramiko.RSAKey, username, 'id_rsa'):
 
792
            return
 
793
        if self._try_pkey_auth(transport, paramiko.DSSKey, username, 'id_dsa'):
 
794
            return
 
795
 
 
796
        if self._password:
 
797
            try:
 
798
                transport.auth_password(username, self._password)
 
799
                return
 
800
            except paramiko.SSHException, e:
 
801
                pass
 
802
 
 
803
            # FIXME: Don't keep a password held in memory if you can help it
 
804
            #self._password = None
 
805
 
 
806
        # give up and ask for a password
 
807
        password = bzrlib.ui.ui_factory.get_password(
 
808
                prompt='SSH %(user)s@%(host)s password',
 
809
                user=username, host=self._host)
 
810
        try:
 
811
            transport.auth_password(username, password)
 
812
        except paramiko.SSHException, e:
 
813
            raise ConnectionError('Unable to authenticate to SSH host as %s@%s' %
 
814
                                  (username, self._host), e)
 
815
 
 
816
    def _try_pkey_auth(self, transport, pkey_class, username, filename):
 
817
        filename = os.path.expanduser('~/.ssh/' + filename)
 
818
        try:
 
819
            key = pkey_class.from_private_key_file(filename)
 
820
            transport.auth_publickey(username, key)
 
821
            return True
 
822
        except paramiko.PasswordRequiredException:
 
823
            password = bzrlib.ui.ui_factory.get_password(
 
824
                    prompt='SSH %(filename)s password',
 
825
                    filename=filename)
 
826
            try:
 
827
                key = pkey_class.from_private_key_file(filename, password)
 
828
                transport.auth_publickey(username, key)
 
829
                return True
 
830
            except paramiko.SSHException:
 
831
                mutter('SSH authentication via %s key failed.' % (os.path.basename(filename),))
 
832
        except paramiko.SSHException:
 
833
            mutter('SSH authentication via %s key failed.' % (os.path.basename(filename),))
 
834
        except IOError:
 
835
            pass
 
836
        return False
 
837
 
845
838
    def _sftp_open_exclusive(self, abspath, mode=None):
846
839
        """Open a remote path exclusively.
847
840
 
856
849
        :param abspath: The remote absolute path where the file should be opened
857
850
        :param mode: The mode permissions bits for the new file
858
851
        """
859
 
        # TODO: jam 20060816 Paramiko >= 1.6.2 (probably earlier) supports
860
 
        #       using the 'x' flag to indicate SFTP_FLAG_EXCL.
861
 
        #       However, there is no way to set the permission mode at open
862
 
        #       time using the sftp_client.file() functionality.
863
 
        path = self._get_sftp()._adjust_cwd(abspath)
864
 
        # mutter('sftp abspath %s => %s', abspath, path)
 
852
        path = self._sftp._adjust_cwd(abspath)
865
853
        attr = SFTPAttributes()
866
854
        if mode is not None:
867
855
            attr.st_mode = mode
868
 
        omode = (SFTP_FLAG_WRITE | SFTP_FLAG_CREATE
 
856
        omode = (SFTP_FLAG_WRITE | SFTP_FLAG_CREATE 
869
857
                | SFTP_FLAG_TRUNC | SFTP_FLAG_EXCL)
870
858
        try:
871
 
            t, msg = self._get_sftp()._request(CMD_OPEN, path, omode, attr)
 
859
            t, msg = self._sftp._request(CMD_OPEN, path, omode, attr)
872
860
            if t != CMD_HANDLE:
873
861
                raise TransportError('Expected an SFTP handle')
874
862
            handle = msg.get_string()
875
 
            return SFTPFile(self._get_sftp(), handle, 'wb', -1)
 
863
            return SFTPFile(self._sftp, handle, 'wb', -1)
876
864
        except (paramiko.SSHException, IOError), e:
877
865
            self._translate_io_exception(e, abspath, ': unable to open',
878
866
                failure_exc=FileExists)
879
867
 
880
 
    def _can_roundtrip_unix_modebits(self):
881
 
        if sys.platform == 'win32':
882
 
            # anyone else?
883
 
            return False
884
 
        else:
885
 
            return True
886
868
 
887
869
# ------------- server test implementation --------------
 
870
import socket
888
871
import threading
889
872
 
890
873
from bzrlib.tests.stub_sftp import StubServer, StubSFTPServer
949
932
                # probably a failed test; unit test thread will log the
950
933
                # failure/error
951
934
                sys.excepthook(*sys.exc_info())
952
 
                warning('Exception from within unit test server thread: %r' %
 
935
                warning('Exception from within unit test server thread: %r' % 
953
936
                        x)
954
937
 
955
938
 
956
 
class SocketDelay(object):
957
 
    """A socket decorator to make TCP appear slower.
958
 
 
959
 
    This changes recv, send, and sendall to add a fixed latency to each python
960
 
    call if a new roundtrip is detected. That is, when a recv is called and the
961
 
    flag new_roundtrip is set, latency is charged. Every send and send_all
962
 
    sets this flag.
963
 
 
964
 
    In addition every send, sendall and recv sleeps a bit per character send to
965
 
    simulate bandwidth.
966
 
 
967
 
    Not all methods are implemented, this is deliberate as this class is not a
968
 
    replacement for the builtin sockets layer. fileno is not implemented to
969
 
    prevent the proxy being bypassed.
970
 
    """
971
 
 
972
 
    simulated_time = 0
973
 
    _proxied_arguments = dict.fromkeys([
974
 
        "close", "getpeername", "getsockname", "getsockopt", "gettimeout",
975
 
        "setblocking", "setsockopt", "settimeout", "shutdown"])
976
 
 
977
 
    def __init__(self, sock, latency, bandwidth=1.0,
978
 
                 really_sleep=True):
979
 
        """
980
 
        :param bandwith: simulated bandwith (MegaBit)
981
 
        :param really_sleep: If set to false, the SocketDelay will just
982
 
        increase a counter, instead of calling time.sleep. This is useful for
983
 
        unittesting the SocketDelay.
984
 
        """
985
 
        self.sock = sock
986
 
        self.latency = latency
987
 
        self.really_sleep = really_sleep
988
 
        self.time_per_byte = 1 / (bandwidth / 8.0 * 1024 * 1024)
989
 
        self.new_roundtrip = False
990
 
 
991
 
    def sleep(self, s):
992
 
        if self.really_sleep:
993
 
            time.sleep(s)
994
 
        else:
995
 
            SocketDelay.simulated_time += s
996
 
 
997
 
    def __getattr__(self, attr):
998
 
        if attr in SocketDelay._proxied_arguments:
999
 
            return getattr(self.sock, attr)
1000
 
        raise AttributeError("'SocketDelay' object has no attribute %r" %
1001
 
                             attr)
1002
 
 
1003
 
    def dup(self):
1004
 
        return SocketDelay(self.sock.dup(), self.latency, self.time_per_byte,
1005
 
                           self._sleep)
1006
 
 
1007
 
    def recv(self, *args):
1008
 
        data = self.sock.recv(*args)
1009
 
        if data and self.new_roundtrip:
1010
 
            self.new_roundtrip = False
1011
 
            self.sleep(self.latency)
1012
 
        self.sleep(len(data) * self.time_per_byte)
1013
 
        return data
1014
 
 
1015
 
    def sendall(self, data, flags=0):
1016
 
        if not self.new_roundtrip:
1017
 
            self.new_roundtrip = True
1018
 
            self.sleep(self.latency)
1019
 
        self.sleep(len(data) * self.time_per_byte)
1020
 
        return self.sock.sendall(data, flags)
1021
 
 
1022
 
    def send(self, data, flags=0):
1023
 
        if not self.new_roundtrip:
1024
 
            self.new_roundtrip = True
1025
 
            self.sleep(self.latency)
1026
 
        bytes_sent = self.sock.send(data, flags)
1027
 
        self.sleep(bytes_sent * self.time_per_byte)
1028
 
        return bytes_sent
1029
 
 
1030
 
 
1031
939
class SFTPServer(Server):
1032
940
    """Common code for SFTP server facilities."""
1033
941
 
1034
 
    def __init__(self, server_interface=StubServer):
 
942
    def __init__(self):
1035
943
        self._original_vendor = None
1036
944
        self._homedir = None
1037
945
        self._server_homedir = None
1038
946
        self._listener = None
1039
947
        self._root = None
1040
 
        self._vendor = ssh.ParamikoVendor()
1041
 
        self._server_interface = server_interface
 
948
        self._vendor = 'none'
1042
949
        # sftp server logs
1043
950
        self.logs = []
1044
 
        self.add_latency = 0
1045
951
 
1046
952
    def _get_sftp_url(self, path):
1047
953
        """Calculate an sftp url to this server for path."""
1051
957
        """StubServer uses this to log when a new server is created."""
1052
958
        self.logs.append(message)
1053
959
 
1054
 
    def _run_server_entry(self, sock):
1055
 
        """Entry point for all implementations of _run_server.
1056
 
 
1057
 
        If self.add_latency is > 0.000001 then sock is given a latency adding
1058
 
        decorator.
1059
 
        """
1060
 
        if self.add_latency > 0.000001:
1061
 
            sock = SocketDelay(sock, self.add_latency)
1062
 
        return self._run_server(sock)
1063
 
 
1064
960
    def _run_server(self, s):
1065
961
        ssh_server = paramiko.Transport(s)
1066
 
        key_file = pathjoin(self._homedir, 'test_rsa.key')
 
962
        key_file = os.path.join(self._homedir, 'test_rsa.key')
1067
963
        f = open(key_file, 'w')
1068
964
        f.write(STUB_SERVER_KEY)
1069
965
        f.close()
1070
966
        host_key = paramiko.RSAKey.from_private_key_file(key_file)
1071
967
        ssh_server.add_server_key(host_key)
1072
 
        server = self._server_interface(self)
 
968
        server = StubServer(self)
1073
969
        ssh_server.set_subsystem_handler('sftp', paramiko.SFTPServer,
1074
970
                                         StubSFTPServer, root=self._root,
1075
971
                                         home=self._server_homedir)
1076
972
        event = threading.Event()
1077
973
        ssh_server.start_server(event, server)
1078
974
        event.wait(5.0)
1079
 
 
1080
 
    def setUp(self, backing_server=None):
1081
 
        # XXX: TODO: make sftpserver back onto backing_server rather than local
1082
 
        # disk.
1083
 
        if not (backing_server is None or
1084
 
                isinstance(backing_server, local.LocalURLServer)):
1085
 
            raise AssertionError(
1086
 
                "backing_server should not be %r, because this can only serve the "
1087
 
                "local current working directory." % (backing_server,))
1088
 
        self._original_vendor = ssh._ssh_vendor_manager._cached_ssh_vendor
1089
 
        ssh._ssh_vendor_manager._cached_ssh_vendor = self._vendor
1090
 
        if sys.platform == 'win32':
1091
 
            # Win32 needs to use the UNICODE api
1092
 
            self._homedir = getcwd()
1093
 
        else:
1094
 
            # But Linux SFTP servers should just deal in bytestreams
1095
 
            self._homedir = os.getcwd()
 
975
    
 
976
    def setUp(self):
 
977
        global _ssh_vendor
 
978
        self._original_vendor = _ssh_vendor
 
979
        _ssh_vendor = self._vendor
 
980
        self._homedir = os.getcwd()
1096
981
        if self._server_homedir is None:
1097
982
            self._server_homedir = self._homedir
1098
983
        self._root = '/'
1099
 
        if sys.platform == 'win32':
1100
 
            self._root = ''
1101
 
        self._listener = SocketListener(self._run_server_entry)
 
984
        # FIXME WINDOWS: _root should be _server_homedir[0]:/
 
985
        self._listener = SocketListener(self._run_server)
1102
986
        self._listener.setDaemon(True)
1103
987
        self._listener.start()
1104
988
 
1105
989
    def tearDown(self):
1106
990
        """See bzrlib.transport.Server.tearDown."""
 
991
        global _ssh_vendor
1107
992
        self._listener.stop()
1108
 
        ssh._ssh_vendor_manager._cached_ssh_vendor = self._original_vendor
 
993
        _ssh_vendor = self._original_vendor
1109
994
 
1110
995
    def get_bogus_url(self):
1111
996
        """See bzrlib.transport.Server.get_bogus_url."""
1112
 
        # this is chosen to try to prevent trouble with proxies, wierd dns, etc
1113
 
        # we bind a random socket, so that we get a guaranteed unused port
1114
 
        # we just never listen on that port
1115
 
        s = socket.socket()
1116
 
        s.bind(('localhost', 0))
1117
 
        return 'sftp://%s:%s/' % s.getsockname()
 
997
        # this is chosen to try to prevent trouble with proxies, wierd dns,
 
998
        # etc
 
999
        return 'sftp://127.0.0.1:1/'
 
1000
 
1118
1001
 
1119
1002
 
1120
1003
class SFTPFullAbsoluteServer(SFTPServer):
1122
1005
 
1123
1006
    def get_url(self):
1124
1007
        """See bzrlib.transport.Server.get_url."""
1125
 
        homedir = self._homedir
1126
 
        if sys.platform != 'win32':
1127
 
            # Remove the initial '/' on all platforms but win32
1128
 
            homedir = homedir[1:]
1129
 
        return self._get_sftp_url(urlutils.escape(homedir))
 
1008
        return self._get_sftp_url(urlutils.escape(self._homedir[1:]))
1130
1009
 
1131
1010
 
1132
1011
class SFTPServerWithoutSSH(SFTPServer):
1134
1013
 
1135
1014
    def __init__(self):
1136
1015
        super(SFTPServerWithoutSSH, self).__init__()
1137
 
        self._vendor = ssh.LoopbackVendor()
 
1016
        self._vendor = 'loopback'
1138
1017
 
1139
1018
    def _run_server(self, sock):
1140
 
        # Re-import these as locals, so that they're still accessible during
1141
 
        # interpreter shutdown (when all module globals get set to None, leading
1142
 
        # to confusing errors like "'NoneType' object has no attribute 'error'".
1143
1019
        class FakeChannel(object):
1144
1020
            def get_transport(self):
1145
1021
                return self
1152
1028
            def close(self):
1153
1029
                pass
1154
1030
 
1155
 
        server = paramiko.SFTPServer(
1156
 
            FakeChannel(), 'sftp', StubServer(self), StubSFTPServer,
1157
 
            root=self._root, home=self._server_homedir)
1158
 
        try:
1159
 
            server.start_subsystem(
1160
 
                'sftp', None, ssh.SocketAsChannelAdapter(sock))
1161
 
        except socket.error, e:
1162
 
            if (len(e.args) > 0) and (e.args[0] == errno.EPIPE):
1163
 
                # it's okay for the client to disconnect abruptly
1164
 
                # (bug in paramiko 1.6: it should absorb this exception)
1165
 
                pass
1166
 
            else:
1167
 
                raise
1168
 
        except Exception, e:
1169
 
            # This typically seems to happen during interpreter shutdown, so
1170
 
            # most of the useful ways to report this error are won't work.
1171
 
            # Writing the exception type, and then the text of the exception,
1172
 
            # seems to be the best we can do.
1173
 
            import sys
1174
 
            sys.stderr.write('\nEXCEPTION %r: ' % (e.__class__,))
1175
 
            sys.stderr.write('%s\n\n' % (e,))
 
1031
        server = paramiko.SFTPServer(FakeChannel(), 'sftp', StubServer(self), StubSFTPServer,
 
1032
                                     root=self._root, home=self._server_homedir)
 
1033
        server.start_subsystem('sftp', None, sock)
1176
1034
        server.finish_subsystem()
1177
1035
 
1178
1036
 
1181
1039
 
1182
1040
    def get_url(self):
1183
1041
        """See bzrlib.transport.Server.get_url."""
1184
 
        homedir = self._homedir
1185
 
        if sys.platform != 'win32':
1186
 
            # Remove the initial '/' on all platforms but win32
1187
 
            homedir = homedir[1:]
1188
 
        return self._get_sftp_url(urlutils.escape(homedir))
 
1042
        return self._get_sftp_url(urlutils.escape(self._homedir[1:]))
1189
1043
 
1190
1044
 
1191
1045
class SFTPHomeDirServer(SFTPServerWithoutSSH):
1197
1051
 
1198
1052
 
1199
1053
class SFTPSiblingAbsoluteServer(SFTPAbsoluteServer):
1200
 
    """A test server for sftp transports where only absolute paths will work.
1201
 
 
1202
 
    It does this by serving from a deeply-nested directory that doesn't exist.
1203
 
    """
1204
 
 
1205
 
    def setUp(self, backing_server=None):
 
1054
    """A test servere for sftp transports, using absolute urls to non-home."""
 
1055
 
 
1056
    def setUp(self):
1206
1057
        self._server_homedir = '/dev/noone/runs/tests/here'
1207
 
        super(SFTPSiblingAbsoluteServer, self).setUp(backing_server)
 
1058
        super(SFTPSiblingAbsoluteServer, self).setUp()
1208
1059
 
1209
1060
 
1210
1061
def get_test_permutations():