~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/transport/sftp.py

  • Committer: Alexander Belchenko
  • Date: 2006-06-29 08:41:31 UTC
  • mto: (1860.1.1 win32.installer)
  • mto: This revision was merged to the branch mainline in revision 1906.
  • Revision ID: bialix@ukr.net-20060629084131-3ea4d44e3204e36f
win32 installer for bzr.dev.0.9

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
26
import socket
33
27
import stat
 
28
import subprocess
34
29
import sys
35
30
import time
36
31
import urllib
37
32
import urlparse
38
 
import warnings
 
33
import weakref
39
34
 
40
 
from bzrlib import (
41
 
    config,
42
 
    debug,
43
 
    errors,
44
 
    urlutils,
45
 
    )
46
 
from bzrlib.errors import (FileExists,
47
 
                           NoSuchFile, PathNotChild,
 
35
from bzrlib.config import config_dir, ensure_config_dir_exists
 
36
from bzrlib.errors import (ConnectionError,
 
37
                           FileExists, 
 
38
                           TransportNotPossible, NoSuchFile, PathNotChild,
48
39
                           TransportError,
49
 
                           LockError,
 
40
                           LockError, 
50
41
                           PathError,
51
42
                           ParamikoNotPresent,
52
43
                           )
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
 
44
from bzrlib.osutils import pathjoin, fancy_rename
 
45
from bzrlib.trace import mutter, warning, error
58
46
from bzrlib.transport import (
59
 
    FileFileStream,
60
 
    _file_streams,
61
 
    local,
 
47
    register_urlparse_netloc_protocol,
62
48
    Server,
63
 
    ssh,
64
 
    ConnectedTransport,
 
49
    split_url,
 
50
    Transport,
65
51
    )
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')
 
52
import bzrlib.ui
 
53
import bzrlib.urlutils as urlutils
79
54
 
80
55
try:
81
56
    import paramiko
87
62
                               CMD_HANDLE, CMD_OPEN)
88
63
    from paramiko.sftp_attr import SFTPAttributes
89
64
    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))
 
65
    from paramiko.sftp_client import SFTPClient
 
66
 
 
67
 
 
68
register_urlparse_netloc_protocol('sftp')
 
69
 
 
70
 
 
71
def _ignore_sigint():
 
72
    # TODO: This should possibly ignore SIGHUP as well, but bzr currently
 
73
    # doesn't handle it itself.
 
74
    # <https://launchpad.net/products/bzr/+bug/41433/+index>
 
75
    import signal
 
76
    signal.signal(signal.SIGINT, signal.SIG_IGN)
 
77
    
 
78
 
 
79
def os_specific_subprocess_params():
 
80
    """Get O/S specific subprocess parameters."""
 
81
    if sys.platform == 'win32':
 
82
        # setting the process group and closing fds is not supported on 
 
83
        # win32
 
84
        return {}
 
85
    else:
 
86
        # We close fds other than the pipes as the child process does not need 
 
87
        # them to be open.
 
88
        #
 
89
        # We also set the child process to ignore SIGINT.  Normally the signal
 
90
        # would be sent to every process in the foreground process group, but
 
91
        # this causes it to be seen only by bzr and not by ssh.  Python will
 
92
        # generate a KeyboardInterrupt in bzr, and we will then have a chance
 
93
        # to release locks or do other cleanup over ssh before the connection
 
94
        # goes away.  
 
95
        # <https://launchpad.net/products/bzr/+bug/5987>
 
96
        #
 
97
        # Running it in a separate process group is not good because then it
 
98
        # can't get non-echoed input of a password or passphrase.
 
99
        # <https://launchpad.net/products/bzr/+bug/40508>
 
100
        return {'preexec_fn': _ignore_sigint,
 
101
                'close_fds': True,
 
102
                }
 
103
 
 
104
 
 
105
# don't use prefetch unless paramiko version >= 1.5.2 (there were bugs earlier)
 
106
_default_do_prefetch = False
 
107
if getattr(paramiko, '__version_info__', (0, 0, 0)) >= (1, 5, 5):
 
108
    _default_do_prefetch = True
 
109
 
 
110
 
 
111
_ssh_vendor = None
 
112
def _get_ssh_vendor():
 
113
    """Find out what version of SSH is on the system."""
 
114
    global _ssh_vendor
 
115
    if _ssh_vendor is not None:
 
116
        return _ssh_vendor
 
117
 
 
118
    _ssh_vendor = 'none'
 
119
 
 
120
    if 'BZR_SSH' in os.environ:
 
121
        _ssh_vendor = os.environ['BZR_SSH']
 
122
        if _ssh_vendor == 'paramiko':
 
123
            _ssh_vendor = 'none'
 
124
        return _ssh_vendor
 
125
 
 
126
    try:
 
127
        p = subprocess.Popen(['ssh', '-V'],
 
128
                             stdin=subprocess.PIPE,
 
129
                             stdout=subprocess.PIPE,
 
130
                             stderr=subprocess.PIPE,
 
131
                             **os_specific_subprocess_params())
 
132
        returncode = p.returncode
 
133
        stdout, stderr = p.communicate()
 
134
    except OSError:
 
135
        returncode = -1
 
136
        stdout = stderr = ''
 
137
    if 'OpenSSH' in stderr:
 
138
        mutter('ssh implementation is OpenSSH')
 
139
        _ssh_vendor = 'openssh'
 
140
    elif 'SSH Secure Shell' in stderr:
 
141
        mutter('ssh implementation is SSH Corp.')
 
142
        _ssh_vendor = 'ssh'
 
143
 
 
144
    if _ssh_vendor != 'none':
 
145
        return _ssh_vendor
 
146
 
 
147
    # XXX: 20051123 jamesh
 
148
    # A check for putty's plink or lsh would go here.
 
149
 
 
150
    mutter('falling back to paramiko implementation')
 
151
    return _ssh_vendor
 
152
 
 
153
 
 
154
class SFTPSubprocess:
 
155
    """A socket-like object that talks to an ssh subprocess via pipes."""
 
156
    def __init__(self, hostname, vendor, port=None, user=None):
 
157
        assert vendor in ['openssh', 'ssh']
 
158
        if vendor == 'openssh':
 
159
            args = ['ssh',
 
160
                    '-oForwardX11=no', '-oForwardAgent=no',
 
161
                    '-oClearAllForwardings=yes', '-oProtocol=2',
 
162
                    '-oNoHostAuthenticationForLocalhost=yes']
 
163
            if port is not None:
 
164
                args.extend(['-p', str(port)])
 
165
            if user is not None:
 
166
                args.extend(['-l', user])
 
167
            args.extend(['-s', hostname, 'sftp'])
 
168
        elif vendor == 'ssh':
 
169
            args = ['ssh', '-x']
 
170
            if port is not None:
 
171
                args.extend(['-p', str(port)])
 
172
            if user is not None:
 
173
                args.extend(['-l', user])
 
174
            args.extend(['-s', 'sftp', hostname])
 
175
 
 
176
        self.proc = subprocess.Popen(args,
 
177
                                     stdin=subprocess.PIPE,
 
178
                                     stdout=subprocess.PIPE,
 
179
                                     **os_specific_subprocess_params())
 
180
 
 
181
    def send(self, data):
 
182
        return os.write(self.proc.stdin.fileno(), data)
 
183
 
 
184
    def recv_ready(self):
 
185
        # TODO: jam 20051215 this function is necessary to support the
 
186
        # pipelined() function. In reality, it probably should use
 
187
        # poll() or select() to actually return if there is data
 
188
        # available, otherwise we probably don't get any benefit
 
189
        return True
 
190
 
 
191
    def recv(self, count):
 
192
        return os.read(self.proc.stdout.fileno(), count)
 
193
 
 
194
    def close(self):
 
195
        self.proc.stdin.close()
 
196
        self.proc.stdout.close()
 
197
        self.proc.wait()
 
198
 
 
199
 
 
200
class LoopbackSFTP(object):
 
201
    """Simple wrapper for a socket that pretends to be a paramiko Channel."""
 
202
 
 
203
    def __init__(self, sock):
 
204
        self.__socket = sock
 
205
 
 
206
    def send(self, data):
 
207
        return self.__socket.send(data)
 
208
 
 
209
    def recv(self, n):
 
210
        return self.__socket.recv(n)
 
211
 
 
212
    def recv_ready(self):
 
213
        return True
 
214
 
 
215
    def close(self):
 
216
        self.__socket.close()
 
217
 
 
218
 
 
219
SYSTEM_HOSTKEYS = {}
 
220
BZR_HOSTKEYS = {}
 
221
 
 
222
# This is a weakref dictionary, so that we can reuse connections
 
223
# that are still active. Long term, it might be nice to have some
 
224
# sort of expiration policy, such as disconnect if inactive for
 
225
# X seconds. But that requires a lot more fanciness.
 
226
_connected_hosts = weakref.WeakValueDictionary()
 
227
 
 
228
def clear_connection_cache():
 
229
    """Remove all hosts from the SFTP connection cache.
 
230
 
 
231
    Primarily useful for test cases wanting to force garbage collection.
 
232
    """
 
233
    _connected_hosts.clear()
 
234
 
 
235
 
 
236
def load_host_keys():
 
237
    """
 
238
    Load system host keys (probably doesn't work on windows) and any
 
239
    "discovered" keys from previous sessions.
 
240
    """
 
241
    global SYSTEM_HOSTKEYS, BZR_HOSTKEYS
 
242
    try:
 
243
        SYSTEM_HOSTKEYS = paramiko.util.load_host_keys(os.path.expanduser('~/.ssh/known_hosts'))
 
244
    except Exception, e:
 
245
        mutter('failed to load system host keys: ' + str(e))
 
246
    bzr_hostkey_path = pathjoin(config_dir(), 'ssh_host_keys')
 
247
    try:
 
248
        BZR_HOSTKEYS = paramiko.util.load_host_keys(bzr_hostkey_path)
 
249
    except Exception, e:
 
250
        mutter('failed to load bzr host keys: ' + str(e))
 
251
        save_host_keys()
 
252
 
 
253
 
 
254
def save_host_keys():
 
255
    """
 
256
    Save "discovered" host keys in $(config)/ssh_host_keys/.
 
257
    """
 
258
    global SYSTEM_HOSTKEYS, BZR_HOSTKEYS
 
259
    bzr_hostkey_path = pathjoin(config_dir(), 'ssh_host_keys')
 
260
    ensure_config_dir_exists()
 
261
 
 
262
    try:
 
263
        f = open(bzr_hostkey_path, 'w')
 
264
        f.write('# SSH host keys collected by bzr\n')
 
265
        for hostname, keys in BZR_HOSTKEYS.iteritems():
 
266
            for keytype, key in keys.iteritems():
 
267
                f.write('%s %s %s\n' % (hostname, keytype, key.get_base64()))
 
268
        f.close()
 
269
    except IOError, e:
 
270
        mutter('failed to save bzr host keys: ' + str(e))
95
271
 
96
272
 
97
273
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
 
 
 
274
    """This fakes a lock in a remote location."""
105
275
    __slots__ = ['path', 'lock_path', 'lock_file', 'transport']
106
 
 
107
276
    def __init__(self, path, transport):
 
277
        assert isinstance(transport, SFTPTransport)
 
278
 
108
279
        self.lock_file = None
109
280
        self.path = path
110
281
        self.lock_path = path + '.write-lock'
134
305
            pass
135
306
 
136
307
 
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
 
 
 
308
class SFTPTransport (Transport):
 
309
    """
 
310
    Transport implementation for SFTP access.
 
311
    """
328
312
    _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
 
 
 
313
 
 
314
    def __init__(self, base, clone_from=None):
 
315
        assert base.startswith('sftp://')
 
316
        self._parse_url(base)
 
317
        base = self._unparse_url()
 
318
        if base[-1] != '/':
 
319
            base += '/'
 
320
        super(SFTPTransport, self).__init__(base)
 
321
        if clone_from is None:
 
322
            self._sftp_connect()
 
323
        else:
 
324
            # use the same ssh connection, etc
 
325
            self._sftp = clone_from._sftp
 
326
        # super saves 'self.base'
 
327
    
 
328
    def should_cache(self):
 
329
        """
 
330
        Return True if the data pulled across should be cached locally.
 
331
        """
 
332
        return True
 
333
 
 
334
    def clone(self, offset=None):
 
335
        """
 
336
        Return a new SFTPTransport with root at self.base + offset.
 
337
        We share the same SFTP session between such transports, because it's
 
338
        fairly expensive to set them up.
 
339
        """
 
340
        if offset is None:
 
341
            return SFTPTransport(self.base, self)
 
342
        else:
 
343
            return SFTPTransport(self.abspath(offset), self)
 
344
 
 
345
    def abspath(self, relpath):
 
346
        """
 
347
        Return the full url to the given relative path.
 
348
        
 
349
        @param relpath: the relative path or path components
 
350
        @type relpath: str or list
 
351
        """
 
352
        return self._unparse_url(self._remote_path(relpath))
 
353
    
350
354
    def _remote_path(self, relpath):
351
355
        """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
 
356
        
 
357
        relpath is a urlencoded string.
 
358
        """
 
359
        # FIXME: share the common code across transports
 
360
        assert isinstance(relpath, basestring)
 
361
        relpath = urlutils.unescape(relpath).split('/')
 
362
        basepath = self._path.split('/')
 
363
        if len(basepath) > 0 and basepath[-1] == '':
 
364
            basepath = basepath[:-1]
 
365
 
 
366
        for p in relpath:
 
367
            if p == '..':
 
368
                if len(basepath) == 0:
 
369
                    # In most filesystems, a request for the parent
 
370
                    # of root, just returns root.
 
371
                    continue
 
372
                basepath.pop()
 
373
            elif p == '.':
 
374
                continue # No-op
 
375
            else:
 
376
                basepath.append(p)
 
377
 
 
378
        path = '/'.join(basepath)
 
379
        return path
 
380
 
 
381
    def relpath(self, abspath):
 
382
        username, password, host, port, path = self._split_url(abspath)
 
383
        error = []
 
384
        if (username != self._username):
 
385
            error.append('username mismatch')
 
386
        if (host != self._host):
 
387
            error.append('host mismatch')
 
388
        if (port != self._port):
 
389
            error.append('port mismatch')
 
390
        if (not path.startswith(self._path)):
 
391
            error.append('path mismatch')
 
392
        if error:
 
393
            extra = ': ' + ', '.join(error)
 
394
            raise PathNotChild(abspath, self.base, extra=extra)
 
395
        pl = len(self._path)
 
396
        return path[pl:].strip('/')
402
397
 
403
398
    def has(self, relpath):
404
399
        """
405
400
        Does the target location exist?
406
401
        """
407
402
        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')
 
403
            self._sftp.stat(self._remote_path(relpath))
411
404
            return True
412
405
        except IOError:
413
406
            return False
414
407
 
415
408
    def get(self, relpath):
416
 
        """Get the file at the given relative path.
 
409
        """
 
410
        Get the file at the given relative path.
417
411
 
418
412
        :param relpath: The relative path to the file
419
413
        """
420
414
        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
415
            path = self._remote_path(relpath)
427
 
            f = self._get_sftp().file(path, mode='rb')
 
416
            f = self._sftp.file(path, mode='rb')
428
417
            if self._do_prefetch and (getattr(f, 'prefetch', None) is not None):
429
418
                f.prefetch()
430
419
            return f
431
420
        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
 
421
            self._translate_io_exception(e, path, ': error retrieving')
 
422
 
 
423
    def get_partial(self, relpath, start, length=None):
 
424
        """
 
425
        Get just part of a file.
 
426
 
 
427
        :param relpath: Path to the file, relative to base
 
428
        :param start: The starting position to read from
 
429
        :param length: The length to read. A length of None indicates
 
430
                       read to the end of the file.
 
431
        :return: A file-like object containing at least the specified bytes.
 
432
                 Some implementations may return objects which can be read
 
433
                 past this length, but this is not guaranteed.
 
434
        """
 
435
        # TODO: implement get_partial_multi to help with knit support
437
436
        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.
 
437
        f.seek(start)
 
438
        if self._do_prefetch and hasattr(f, 'prefetch'):
 
439
            f.prefetch()
 
440
        return f
 
441
 
 
442
    def put(self, relpath, f, mode=None):
 
443
        """
 
444
        Copy the file-like or string object into the location.
486
445
 
487
446
        :param relpath: Location to put the contents, relative to base.
488
 
        :param f:       File-like object.
 
447
        :param f:       File-like or string object.
489
448
        :param mode: The final mode for the file
490
449
        """
491
450
        final_path = self._remote_path(relpath)
492
 
        return self._put(final_path, f, mode=mode)
 
451
        self._put(final_path, f, mode=mode)
493
452
 
494
453
    def _put(self, abspath, f, mode=None):
495
454
        """Helper function so both put() and copy_abspaths can reuse the code"""
500
459
        try:
501
460
            try:
502
461
                fout.set_pipelined(True)
503
 
                length = self._pump(f, fout)
 
462
                self._pump(f, fout)
504
463
            except (IOError, paramiko.SSHException), e:
505
464
                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
465
            if mode is not None:
520
 
                self._get_sftp().chmod(tmp_abspath, mode)
 
466
                self._sftp.chmod(tmp_abspath, mode)
521
467
            fout.close()
522
468
            closed = True
523
469
            self._rename_and_overwrite(tmp_abspath, abspath)
524
 
            return length
525
470
        except Exception, e:
526
471
            # If we fail, try to clean up the temporary file
527
472
            # before we throw the exception
533
478
            try:
534
479
                if not closed:
535
480
                    fout.close()
536
 
                self._get_sftp().remove(tmp_abspath)
 
481
                self._sftp.remove(tmp_abspath)
537
482
            except:
538
483
                # raise the saved except
539
484
                raise e
540
485
            # raise the original with its traceback if we can.
541
486
            raise
542
487
 
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
488
    def iter_files_recursive(self):
621
489
        """Walk the relative paths of all files in this transport."""
622
 
        # progress is handled by list_dir
623
490
        queue = list(self.list_dir('.'))
624
491
        while queue:
625
 
            relpath = queue.pop(0)
 
492
            relpath = urllib.quote(queue.pop(0))
626
493
            st = self.stat(relpath)
627
494
            if stat.S_ISDIR(st.st_mode):
628
495
                for i, basename in enumerate(self.list_dir(relpath)):
630
497
            else:
631
498
                yield relpath
632
499
 
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
500
    def mkdir(self, relpath, mode=None):
663
501
        """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
502
        try:
680
 
            handle = self._get_sftp().file(abspath, mode='wb')
681
 
            handle.set_pipelined(True)
 
503
            path = self._remote_path(relpath)
 
504
            # In the paramiko documentation, it says that passing a mode flag 
 
505
            # will filtered against the server umask.
 
506
            # StubSFTPServer does not do this, which would be nice, because it is
 
507
            # what we really want :)
 
508
            # However, real servers do use umask, so we really should do it that way
 
509
            self._sftp.mkdir(path)
 
510
            if mode is not None:
 
511
                self._sftp.chmod(path, mode=mode)
682
512
        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)
 
513
            self._translate_io_exception(e, path, ': unable to mkdir',
 
514
                failure_exc=FileExists)
687
515
 
688
 
    def _translate_io_exception(self, e, path, more_info='',
 
516
    def _translate_io_exception(self, e, path, more_info='', 
689
517
                                failure_exc=PathError):
690
518
        """Translate a paramiko or IOError into a friendlier exception.
691
519
 
696
524
        :param failure_exc: Paramiko has the super fun ability to raise completely
697
525
                           opaque errors that just set "e.args = ('Failure',)" with
698
526
                           no more information.
699
 
                           If this parameter is set, it defines the exception
 
527
                           If this parameter is set, it defines the exception 
700
528
                           to raise in these cases.
701
529
        """
702
530
        # paramiko seems to generate detailless errors.
703
531
        self._translate_error(e, path, raise_generic=False)
704
 
        if getattr(e, 'args', None) is not None:
 
532
        if hasattr(e, 'args'):
705
533
            if (e.args == ('No such file or directory',) or
706
534
                e.args == ('No such file',)):
707
535
                raise NoSuchFile(path, str(e) + more_info)
708
 
            if (e.args == ('mkdir failed',) or
709
 
                e.args[0].startswith('syserr: File exists')):
 
536
            if (e.args == ('mkdir failed',)):
710
537
                raise FileExists(path, str(e) + more_info)
711
538
            # strange but true, for the paramiko server.
712
539
            if (e.args == ('Failure',)):
713
540
                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
541
            mutter('Raising exception with args %s', e.args)
721
 
        if getattr(e, 'errno', None) is not None:
 
542
        if hasattr(e, 'errno'):
722
543
            mutter('Raising exception with errno %s', e.errno)
723
544
        raise e
724
545
 
725
 
    def append_file(self, relpath, f, mode=None):
 
546
    def append(self, relpath, f, mode=None):
726
547
        """
727
548
        Append the text in the file-like object into the final
728
549
        location.
729
550
        """
730
551
        try:
731
552
            path = self._remote_path(relpath)
732
 
            fout = self._get_sftp().file(path, 'ab')
 
553
            fout = self._sftp.file(path, 'ab')
733
554
            if mode is not None:
734
 
                self._get_sftp().chmod(path, mode)
 
555
                self._sftp.chmod(path, mode)
735
556
            result = fout.tell()
736
557
            self._pump(f, fout)
737
558
            return result
741
562
    def rename(self, rel_from, rel_to):
742
563
        """Rename without special overwriting"""
743
564
        try:
744
 
            self._get_sftp().rename(self._remote_path(rel_from),
 
565
            self._sftp.rename(self._remote_path(rel_from),
745
566
                              self._remote_path(rel_to))
746
567
        except (IOError, paramiko.SSHException), e:
747
568
            self._translate_io_exception(e, rel_from,
749
570
 
750
571
    def _rename_and_overwrite(self, abs_from, abs_to):
751
572
        """Do a fancy rename on the remote server.
752
 
 
 
573
        
753
574
        Using the implementation provided by osutils.
754
575
        """
755
576
        try:
756
 
            sftp = self._get_sftp()
757
577
            fancy_rename(abs_from, abs_to,
758
 
                         rename_func=sftp.rename,
759
 
                         unlink_func=sftp.remove)
 
578
                    rename_func=self._sftp.rename,
 
579
                    unlink_func=self._sftp.remove)
760
580
        except (IOError, paramiko.SSHException), e:
761
 
            self._translate_io_exception(e, abs_from,
762
 
                                         ': unable to rename to %r' % (abs_to))
 
581
            self._translate_io_exception(e, abs_from, ': unable to rename to %r' % (abs_to))
763
582
 
764
583
    def move(self, rel_from, rel_to):
765
584
        """Move the item at rel_from to the location at rel_to"""
771
590
        """Delete the item at relpath"""
772
591
        path = self._remote_path(relpath)
773
592
        try:
774
 
            self._get_sftp().remove(path)
 
593
            self._sftp.remove(path)
775
594
        except (IOError, paramiko.SSHException), e:
776
595
            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
 
 
 
596
            
783
597
    def listable(self):
784
598
        """Return True if this store supports listing."""
785
599
        return True
789
603
        Return a list of all files at the given location.
790
604
        """
791
605
        # 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
606
        path = self._remote_path(relpath)
796
607
        try:
797
 
            entries = self._get_sftp().listdir(path)
798
 
            self._report_activity(sum(map(len, entries)), 'read')
 
608
            return self._sftp.listdir(path)
799
609
        except (IOError, paramiko.SSHException), e:
800
610
            self._translate_io_exception(e, path, ': failed to list_dir')
801
 
        return [urlutils.escape(entry) for entry in entries]
802
611
 
803
612
    def rmdir(self, relpath):
804
613
        """See Transport.rmdir."""
805
614
        path = self._remote_path(relpath)
806
615
        try:
807
 
            return self._get_sftp().rmdir(path)
 
616
            return self._sftp.rmdir(path)
808
617
        except (IOError, paramiko.SSHException), e:
809
618
            self._translate_io_exception(e, path, ': failed to rmdir')
810
619
 
812
621
        """Return the stat information for a file."""
813
622
        path = self._remote_path(relpath)
814
623
        try:
815
 
            return self._get_sftp().stat(path)
 
624
            return self._sftp.stat(path)
816
625
        except (IOError, paramiko.SSHException), e:
817
626
            self._translate_io_exception(e, path, ': unable to stat')
818
627
 
842
651
        # that we have taken the lock.
843
652
        return SFTPLock(relpath, self)
844
653
 
 
654
    def _unparse_url(self, path=None):
 
655
        if path is None:
 
656
            path = self._path
 
657
        path = urllib.quote(path)
 
658
        # handle homedir paths
 
659
        if not path.startswith('/'):
 
660
            path = "/~/" + path
 
661
        netloc = urllib.quote(self._host)
 
662
        if self._username is not None:
 
663
            netloc = '%s@%s' % (urllib.quote(self._username), netloc)
 
664
        if self._port is not None:
 
665
            netloc = '%s:%d' % (netloc, self._port)
 
666
        return urlparse.urlunparse(('sftp', netloc, path, '', '', ''))
 
667
 
 
668
    def _split_url(self, url):
 
669
        (scheme, username, password, host, port, path) = split_url(url)
 
670
        assert scheme == 'sftp'
 
671
 
 
672
        # the initial slash should be removed from the path, and treated
 
673
        # as a homedir relative path (the path begins with a double slash
 
674
        # if it is absolute).
 
675
        # see draft-ietf-secsh-scp-sftp-ssh-uri-03.txt
 
676
        # RBC 20060118 we are not using this as its too user hostile. instead
 
677
        # we are following lftp and using /~/foo to mean '~/foo'.
 
678
        # handle homedir paths
 
679
        if path.startswith('/~/'):
 
680
            path = path[3:]
 
681
        elif path == '/~':
 
682
            path = ''
 
683
        return (username, password, host, port, path)
 
684
 
 
685
    def _parse_url(self, url):
 
686
        (self._username, self._password,
 
687
         self._host, self._port, self._path) = self._split_url(url)
 
688
 
 
689
    def _sftp_connect(self):
 
690
        """Connect to the remote sftp server.
 
691
        After this, self._sftp should have a valid connection (or
 
692
        we raise an TransportError 'could not connect').
 
693
 
 
694
        TODO: Raise a more reasonable ConnectionFailed exception
 
695
        """
 
696
        global _connected_hosts
 
697
 
 
698
        idx = (self._host, self._port, self._username)
 
699
        try:
 
700
            self._sftp = _connected_hosts[idx]
 
701
            return
 
702
        except KeyError:
 
703
            pass
 
704
        
 
705
        vendor = _get_ssh_vendor()
 
706
        if vendor == 'loopback':
 
707
            sock = socket.socket()
 
708
            try:
 
709
                sock.connect((self._host, self._port))
 
710
            except socket.error, e:
 
711
                raise ConnectionError('Unable to connect to SSH host %s:%s: %s'
 
712
                                      % (self._host, self._port, e))
 
713
            self._sftp = SFTPClient(LoopbackSFTP(sock))
 
714
        elif vendor != 'none':
 
715
            sock = SFTPSubprocess(self._host, vendor, self._port,
 
716
                                  self._username)
 
717
            self._sftp = SFTPClient(sock)
 
718
        else:
 
719
            self._paramiko_connect()
 
720
 
 
721
        _connected_hosts[idx] = self._sftp
 
722
 
 
723
    def _paramiko_connect(self):
 
724
        global SYSTEM_HOSTKEYS, BZR_HOSTKEYS
 
725
        
 
726
        load_host_keys()
 
727
 
 
728
        try:
 
729
            t = paramiko.Transport((self._host, self._port or 22))
 
730
            t.set_log_channel('bzr.paramiko')
 
731
            t.start_client()
 
732
        except paramiko.SSHException, e:
 
733
            raise ConnectionError('Unable to reach SSH host %s:%s: %s' 
 
734
                                  % (self._host, self._port, e))
 
735
            
 
736
        server_key = t.get_remote_server_key()
 
737
        server_key_hex = paramiko.util.hexify(server_key.get_fingerprint())
 
738
        keytype = server_key.get_name()
 
739
        if SYSTEM_HOSTKEYS.has_key(self._host) and SYSTEM_HOSTKEYS[self._host].has_key(keytype):
 
740
            our_server_key = SYSTEM_HOSTKEYS[self._host][keytype]
 
741
            our_server_key_hex = paramiko.util.hexify(our_server_key.get_fingerprint())
 
742
        elif BZR_HOSTKEYS.has_key(self._host) and BZR_HOSTKEYS[self._host].has_key(keytype):
 
743
            our_server_key = BZR_HOSTKEYS[self._host][keytype]
 
744
            our_server_key_hex = paramiko.util.hexify(our_server_key.get_fingerprint())
 
745
        else:
 
746
            warning('Adding %s host key for %s: %s' % (keytype, self._host, server_key_hex))
 
747
            if not BZR_HOSTKEYS.has_key(self._host):
 
748
                BZR_HOSTKEYS[self._host] = {}
 
749
            BZR_HOSTKEYS[self._host][keytype] = server_key
 
750
            our_server_key = server_key
 
751
            our_server_key_hex = paramiko.util.hexify(our_server_key.get_fingerprint())
 
752
            save_host_keys()
 
753
        if server_key != our_server_key:
 
754
            filename1 = os.path.expanduser('~/.ssh/known_hosts')
 
755
            filename2 = pathjoin(config_dir(), 'ssh_host_keys')
 
756
            raise TransportError('Host keys for %s do not match!  %s != %s' % \
 
757
                (self._host, our_server_key_hex, server_key_hex),
 
758
                ['Try editing %s or %s' % (filename1, filename2)])
 
759
 
 
760
        self._sftp_auth(t)
 
761
        
 
762
        try:
 
763
            self._sftp = t.open_sftp_client()
 
764
        except paramiko.SSHException, e:
 
765
            raise ConnectionError('Unable to start sftp client %s:%d' %
 
766
                                  (self._host, self._port), e)
 
767
 
 
768
    def _sftp_auth(self, transport):
 
769
        # paramiko requires a username, but it might be none if nothing was supplied
 
770
        # use the local username, just in case.
 
771
        # We don't override self._username, because if we aren't using paramiko,
 
772
        # the username might be specified in ~/.ssh/config and we don't want to
 
773
        # force it to something else
 
774
        # Also, it would mess up the self.relpath() functionality
 
775
        username = self._username or getpass.getuser()
 
776
 
 
777
        # Paramiko tries to open a socket.AF_UNIX in order to connect
 
778
        # to ssh-agent. That attribute doesn't exist on win32 (it does in cygwin)
 
779
        # so we get an AttributeError exception. For now, just don't try to
 
780
        # connect to an agent if we are on win32
 
781
        if sys.platform != 'win32':
 
782
            agent = paramiko.Agent()
 
783
            for key in agent.get_keys():
 
784
                mutter('Trying SSH agent key %s' % paramiko.util.hexify(key.get_fingerprint()))
 
785
                try:
 
786
                    transport.auth_publickey(username, key)
 
787
                    return
 
788
                except paramiko.SSHException, e:
 
789
                    pass
 
790
        
 
791
        # okay, try finding id_rsa or id_dss?  (posix only)
 
792
        if self._try_pkey_auth(transport, paramiko.RSAKey, username, 'id_rsa'):
 
793
            return
 
794
        if self._try_pkey_auth(transport, paramiko.DSSKey, username, 'id_dsa'):
 
795
            return
 
796
 
 
797
        if self._password:
 
798
            try:
 
799
                transport.auth_password(username, self._password)
 
800
                return
 
801
            except paramiko.SSHException, e:
 
802
                pass
 
803
 
 
804
            # FIXME: Don't keep a password held in memory if you can help it
 
805
            #self._password = None
 
806
 
 
807
        # give up and ask for a password
 
808
        password = bzrlib.ui.ui_factory.get_password(
 
809
                prompt='SSH %(user)s@%(host)s password',
 
810
                user=username, host=self._host)
 
811
        try:
 
812
            transport.auth_password(username, password)
 
813
        except paramiko.SSHException, e:
 
814
            raise ConnectionError('Unable to authenticate to SSH host as %s@%s' %
 
815
                                  (username, self._host), e)
 
816
 
 
817
    def _try_pkey_auth(self, transport, pkey_class, username, filename):
 
818
        filename = os.path.expanduser('~/.ssh/' + filename)
 
819
        try:
 
820
            key = pkey_class.from_private_key_file(filename)
 
821
            transport.auth_publickey(username, key)
 
822
            return True
 
823
        except paramiko.PasswordRequiredException:
 
824
            password = bzrlib.ui.ui_factory.get_password(
 
825
                    prompt='SSH %(filename)s password',
 
826
                    filename=filename)
 
827
            try:
 
828
                key = pkey_class.from_private_key_file(filename, password)
 
829
                transport.auth_publickey(username, key)
 
830
                return True
 
831
            except paramiko.SSHException:
 
832
                mutter('SSH authentication via %s key failed.' % (os.path.basename(filename),))
 
833
        except paramiko.SSHException:
 
834
            mutter('SSH authentication via %s key failed.' % (os.path.basename(filename),))
 
835
        except IOError:
 
836
            pass
 
837
        return False
 
838
 
845
839
    def _sftp_open_exclusive(self, abspath, mode=None):
846
840
        """Open a remote path exclusively.
847
841
 
856
850
        :param abspath: The remote absolute path where the file should be opened
857
851
        :param mode: The mode permissions bits for the new file
858
852
        """
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)
 
853
        path = self._sftp._adjust_cwd(abspath)
865
854
        attr = SFTPAttributes()
866
855
        if mode is not None:
867
856
            attr.st_mode = mode
868
 
        omode = (SFTP_FLAG_WRITE | SFTP_FLAG_CREATE
 
857
        omode = (SFTP_FLAG_WRITE | SFTP_FLAG_CREATE 
869
858
                | SFTP_FLAG_TRUNC | SFTP_FLAG_EXCL)
870
859
        try:
871
 
            t, msg = self._get_sftp()._request(CMD_OPEN, path, omode, attr)
 
860
            t, msg = self._sftp._request(CMD_OPEN, path, omode, attr)
872
861
            if t != CMD_HANDLE:
873
862
                raise TransportError('Expected an SFTP handle')
874
863
            handle = msg.get_string()
875
 
            return SFTPFile(self._get_sftp(), handle, 'wb', -1)
 
864
            return SFTPFile(self._sftp, handle, 'wb', -1)
876
865
        except (paramiko.SSHException, IOError), e:
877
866
            self._translate_io_exception(e, abspath, ': unable to open',
878
867
                failure_exc=FileExists)
879
868
 
880
 
    def _can_roundtrip_unix_modebits(self):
881
 
        if sys.platform == 'win32':
882
 
            # anyone else?
883
 
            return False
884
 
        else:
885
 
            return True
886
869
 
887
870
# ------------- server test implementation --------------
 
871
import socket
888
872
import threading
889
873
 
890
874
from bzrlib.tests.stub_sftp import StubServer, StubSFTPServer
949
933
                # probably a failed test; unit test thread will log the
950
934
                # failure/error
951
935
                sys.excepthook(*sys.exc_info())
952
 
                warning('Exception from within unit test server thread: %r' %
 
936
                warning('Exception from within unit test server thread: %r' % 
953
937
                        x)
954
938
 
955
939
 
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
940
class SFTPServer(Server):
1032
941
    """Common code for SFTP server facilities."""
1033
942
 
1034
 
    def __init__(self, server_interface=StubServer):
 
943
    def __init__(self):
1035
944
        self._original_vendor = None
1036
945
        self._homedir = None
1037
946
        self._server_homedir = None
1038
947
        self._listener = None
1039
948
        self._root = None
1040
 
        self._vendor = ssh.ParamikoVendor()
1041
 
        self._server_interface = server_interface
 
949
        self._vendor = 'none'
1042
950
        # sftp server logs
1043
951
        self.logs = []
1044
 
        self.add_latency = 0
1045
952
 
1046
953
    def _get_sftp_url(self, path):
1047
954
        """Calculate an sftp url to this server for path."""
1051
958
        """StubServer uses this to log when a new server is created."""
1052
959
        self.logs.append(message)
1053
960
 
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
961
    def _run_server(self, s):
1065
962
        ssh_server = paramiko.Transport(s)
1066
 
        key_file = pathjoin(self._homedir, 'test_rsa.key')
 
963
        key_file = os.path.join(self._homedir, 'test_rsa.key')
1067
964
        f = open(key_file, 'w')
1068
965
        f.write(STUB_SERVER_KEY)
1069
966
        f.close()
1070
967
        host_key = paramiko.RSAKey.from_private_key_file(key_file)
1071
968
        ssh_server.add_server_key(host_key)
1072
 
        server = self._server_interface(self)
 
969
        server = StubServer(self)
1073
970
        ssh_server.set_subsystem_handler('sftp', paramiko.SFTPServer,
1074
971
                                         StubSFTPServer, root=self._root,
1075
972
                                         home=self._server_homedir)
1076
973
        event = threading.Event()
1077
974
        ssh_server.start_server(event, server)
1078
975
        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()
 
976
    
 
977
    def setUp(self):
 
978
        global _ssh_vendor
 
979
        self._original_vendor = _ssh_vendor
 
980
        _ssh_vendor = self._vendor
 
981
        self._homedir = os.getcwd()
1096
982
        if self._server_homedir is None:
1097
983
            self._server_homedir = self._homedir
1098
984
        self._root = '/'
1099
 
        if sys.platform == 'win32':
1100
 
            self._root = ''
1101
 
        self._listener = SocketListener(self._run_server_entry)
 
985
        # FIXME WINDOWS: _root should be _server_homedir[0]:/
 
986
        self._listener = SocketListener(self._run_server)
1102
987
        self._listener.setDaemon(True)
1103
988
        self._listener.start()
1104
989
 
1105
990
    def tearDown(self):
1106
991
        """See bzrlib.transport.Server.tearDown."""
 
992
        global _ssh_vendor
1107
993
        self._listener.stop()
1108
 
        ssh._ssh_vendor_manager._cached_ssh_vendor = self._original_vendor
 
994
        _ssh_vendor = self._original_vendor
1109
995
 
1110
996
    def get_bogus_url(self):
1111
997
        """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()
 
998
        # this is chosen to try to prevent trouble with proxies, wierd dns,
 
999
        # etc
 
1000
        return 'sftp://127.0.0.1:1/'
 
1001
 
1118
1002
 
1119
1003
 
1120
1004
class SFTPFullAbsoluteServer(SFTPServer):
1122
1006
 
1123
1007
    def get_url(self):
1124
1008
        """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))
 
1009
        return self._get_sftp_url(urlutils.escape(self._homedir[1:]))
1130
1010
 
1131
1011
 
1132
1012
class SFTPServerWithoutSSH(SFTPServer):
1134
1014
 
1135
1015
    def __init__(self):
1136
1016
        super(SFTPServerWithoutSSH, self).__init__()
1137
 
        self._vendor = ssh.LoopbackVendor()
 
1017
        self._vendor = 'loopback'
1138
1018
 
1139
1019
    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
1020
        class FakeChannel(object):
1144
1021
            def get_transport(self):
1145
1022
                return self
1152
1029
            def close(self):
1153
1030
                pass
1154
1031
 
1155
 
        server = paramiko.SFTPServer(
1156
 
            FakeChannel(), 'sftp', StubServer(self), StubSFTPServer,
1157
 
            root=self._root, home=self._server_homedir)
 
1032
        server = paramiko.SFTPServer(FakeChannel(), 'sftp', StubServer(self), StubSFTPServer,
 
1033
                                     root=self._root, home=self._server_homedir)
1158
1034
        try:
1159
 
            server.start_subsystem(
1160
 
                'sftp', None, ssh.SocketAsChannelAdapter(sock))
 
1035
            server.start_subsystem('sftp', None, sock)
1161
1036
        except socket.error, e:
1162
1037
            if (len(e.args) > 0) and (e.args[0] == errno.EPIPE):
1163
1038
                # it's okay for the client to disconnect abruptly
1166
1041
            else:
1167
1042
                raise
1168
1043
        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,))
 
1044
            import sys; sys.stderr.write('\nEXCEPTION %r\n\n' % e.__class__)
1176
1045
        server.finish_subsystem()
1177
1046
 
1178
1047
 
1181
1050
 
1182
1051
    def get_url(self):
1183
1052
        """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))
 
1053
        return self._get_sftp_url(urlutils.escape(self._homedir[1:]))
1189
1054
 
1190
1055
 
1191
1056
class SFTPHomeDirServer(SFTPServerWithoutSSH):
1197
1062
 
1198
1063
 
1199
1064
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):
 
1065
    """A test servere for sftp transports, using absolute urls to non-home."""
 
1066
 
 
1067
    def setUp(self):
1206
1068
        self._server_homedir = '/dev/noone/runs/tests/here'
1207
 
        super(SFTPSiblingAbsoluteServer, self).setUp(backing_server)
 
1069
        super(SFTPSiblingAbsoluteServer, self).setUp()
1208
1070
 
1209
1071
 
1210
1072
def get_test_permutations():