~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/transport/sftp.py

  • Committer: Aaron Bentley
  • Date: 2007-12-12 15:17:13 UTC
  • mto: This revision was merged to the branch mainline in revision 3113.
  • Revision ID: abentley@panoramicfeedback.com-20071212151713-ox5n8rlx8m3nsspy
Add support for reconfiguring repositories into branches or trees

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005 Robey Pointer <robey@lag.net>, Canonical Ltd
2
 
 
 
1
# Copyright (C) 2005 Robey Pointer <robey@lag.net>
 
2
# Copyright (C) 2005, 2006, 2007 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
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
 
 
20
# TODO: Remove the transport-based lock_read and lock_write methods.  They'll
 
21
# then raise TransportNotPossible, which will break remote access to any
 
22
# formats which rely on OS-level locks.  That should be fine as those formats
 
23
# are pretty old, but these combinations may have to be removed from the test
 
24
# suite.  Those formats all date back to 0.7; so we should be able to remove
 
25
# these methods when we officially drop support for those formats.
 
26
 
19
27
import errno
20
 
import getpass
21
28
import os
22
 
import re
 
29
import random
 
30
import select
 
31
import socket
23
32
import stat
24
33
import sys
 
34
import time
25
35
import urllib
26
36
import urlparse
27
 
import time
28
 
import random
29
 
import subprocess
30
 
import weakref
 
37
import warnings
31
38
 
32
 
from bzrlib.errors import (FileExists, 
33
 
                           TransportNotPossible, NoSuchFile, PathNotChild,
 
39
from bzrlib import (
 
40
    errors,
 
41
    urlutils,
 
42
    )
 
43
from bzrlib.errors import (FileExists,
 
44
                           NoSuchFile, PathNotChild,
34
45
                           TransportError,
35
 
                           LockError)
36
 
from bzrlib.config import config_dir, ensure_config_dir_exists
37
 
from bzrlib.trace import mutter, warning, error
38
 
from bzrlib.transport import Transport, register_transport
39
 
from bzrlib.osutils import pathjoin, fancy_rename
40
 
import bzrlib.ui
 
46
                           LockError,
 
47
                           PathError,
 
48
                           ParamikoNotPresent,
 
49
                           )
 
50
from bzrlib.osutils import pathjoin, fancy_rename, getcwd
 
51
from bzrlib.symbol_versioning import (
 
52
        deprecated_function,
 
53
        zero_ninety,
 
54
        )
 
55
from bzrlib.trace import mutter, warning
 
56
from bzrlib.transport import (
 
57
    FileFileStream,
 
58
    _file_streams,
 
59
    local,
 
60
    Server,
 
61
    ssh,
 
62
    ConnectedTransport,
 
63
    )
 
64
 
 
65
# Disable one particular warning that comes from paramiko in Python2.5; if
 
66
# this is emitted at the wrong time it tends to cause spurious test failures
 
67
# or at least noise in the test case::
 
68
#
 
69
# [1770/7639 in 86s, 1 known failures, 50 skipped, 2 missing features]
 
70
# test_permissions.TestSftpPermissions.test_new_files
 
71
# /var/lib/python-support/python2.5/paramiko/message.py:226: DeprecationWarning: integer argument expected, got float
 
72
#  self.packet.write(struct.pack('>I', n))
 
73
warnings.filterwarnings('ignore',
 
74
        'integer argument expected, got float',
 
75
        category=DeprecationWarning,
 
76
        module='paramiko.message')
41
77
 
42
78
try:
43
79
    import paramiko
44
 
except ImportError:
45
 
    error('The SFTP transport requires paramiko.')
46
 
    raise
 
80
except ImportError, e:
 
81
    raise ParamikoNotPresent(e)
47
82
else:
48
83
    from paramiko.sftp import (SFTP_FLAG_WRITE, SFTP_FLAG_CREATE,
49
84
                               SFTP_FLAG_EXCL, SFTP_FLAG_TRUNC,
50
85
                               CMD_HANDLE, CMD_OPEN)
51
86
    from paramiko.sftp_attr import SFTPAttributes
52
87
    from paramiko.sftp_file import SFTPFile
53
 
    from paramiko.sftp_client import SFTPClient
54
 
 
55
 
if 'sftp' not in urlparse.uses_netloc: urlparse.uses_netloc.append('sftp')
56
 
 
57
 
 
58
 
_close_fds = True
59
 
if sys.platform == 'win32':
60
 
    # close_fds not supported on win32
61
 
    _close_fds = False
62
 
 
63
 
_ssh_vendor = None
64
 
def _get_ssh_vendor():
65
 
    """Find out what version of SSH is on the system."""
66
 
    global _ssh_vendor
67
 
    if _ssh_vendor is not None:
68
 
        return _ssh_vendor
69
 
 
70
 
    _ssh_vendor = 'none'
71
 
 
72
 
    try:
73
 
        p = subprocess.Popen(['ssh', '-V'],
74
 
                             close_fds=_close_fds,
75
 
                             stdin=subprocess.PIPE,
76
 
                             stdout=subprocess.PIPE,
77
 
                             stderr=subprocess.PIPE)
78
 
        returncode = p.returncode
79
 
        stdout, stderr = p.communicate()
80
 
    except OSError:
81
 
        returncode = -1
82
 
        stdout = stderr = ''
83
 
    if 'OpenSSH' in stderr:
84
 
        mutter('ssh implementation is OpenSSH')
85
 
        _ssh_vendor = 'openssh'
86
 
    elif 'SSH Secure Shell' in stderr:
87
 
        mutter('ssh implementation is SSH Corp.')
88
 
        _ssh_vendor = 'ssh'
89
 
 
90
 
    if _ssh_vendor != 'none':
91
 
        return _ssh_vendor
92
 
 
93
 
    # XXX: 20051123 jamesh
94
 
    # A check for putty's plink or lsh would go here.
95
 
 
96
 
    mutter('falling back to paramiko implementation')
97
 
    return _ssh_vendor
98
 
 
99
 
 
100
 
class SFTPSubprocess:
101
 
    """A socket-like object that talks to an ssh subprocess via pipes."""
102
 
    def __init__(self, hostname, port=None, user=None):
103
 
        vendor = _get_ssh_vendor()
104
 
        assert vendor in ['openssh', 'ssh']
105
 
        if vendor == 'openssh':
106
 
            args = ['ssh',
107
 
                    '-oForwardX11=no', '-oForwardAgent=no',
108
 
                    '-oClearAllForwardings=yes', '-oProtocol=2',
109
 
                    '-oNoHostAuthenticationForLocalhost=yes']
110
 
            if port is not None:
111
 
                args.extend(['-p', str(port)])
112
 
            if user is not None:
113
 
                args.extend(['-l', user])
114
 
            args.extend(['-s', hostname, 'sftp'])
115
 
        elif vendor == 'ssh':
116
 
            args = ['ssh', '-x']
117
 
            if port is not None:
118
 
                args.extend(['-p', str(port)])
119
 
            if user is not None:
120
 
                args.extend(['-l', user])
121
 
            args.extend(['-s', 'sftp', hostname])
122
 
 
123
 
        self.proc = subprocess.Popen(args, close_fds=_close_fds,
124
 
                                     stdin=subprocess.PIPE,
125
 
                                     stdout=subprocess.PIPE)
126
 
 
127
 
    def send(self, data):
128
 
        return os.write(self.proc.stdin.fileno(), data)
129
 
 
130
 
    def recv_ready(self):
131
 
        # TODO: jam 20051215 this function is necessary to support the
132
 
        # pipelined() function. In reality, it probably should use
133
 
        # poll() or select() to actually return if there is data
134
 
        # available, otherwise we probably don't get any benefit
135
 
        return True
136
 
 
137
 
    def recv(self, count):
138
 
        return os.read(self.proc.stdout.fileno(), count)
139
 
 
140
 
    def close(self):
141
 
        self.proc.stdin.close()
142
 
        self.proc.stdout.close()
143
 
        self.proc.wait()
144
 
 
145
 
 
146
 
SYSTEM_HOSTKEYS = {}
147
 
BZR_HOSTKEYS = {}
148
 
 
149
 
# This is a weakref dictionary, so that we can reuse connections
150
 
# that are still active. Long term, it might be nice to have some
151
 
# sort of expiration policy, such as disconnect if inactive for
152
 
# X seconds. But that requires a lot more fanciness.
153
 
_connected_hosts = weakref.WeakValueDictionary()
154
 
 
155
 
def load_host_keys():
156
 
    """
157
 
    Load system host keys (probably doesn't work on windows) and any
158
 
    "discovered" keys from previous sessions.
159
 
    """
160
 
    global SYSTEM_HOSTKEYS, BZR_HOSTKEYS
161
 
    try:
162
 
        SYSTEM_HOSTKEYS = paramiko.util.load_host_keys(os.path.expanduser('~/.ssh/known_hosts'))
163
 
    except Exception, e:
164
 
        mutter('failed to load system host keys: ' + str(e))
165
 
    bzr_hostkey_path = pathjoin(config_dir(), 'ssh_host_keys')
166
 
    try:
167
 
        BZR_HOSTKEYS = paramiko.util.load_host_keys(bzr_hostkey_path)
168
 
    except Exception, e:
169
 
        mutter('failed to load bzr host keys: ' + str(e))
170
 
        save_host_keys()
171
 
 
172
 
def save_host_keys():
173
 
    """
174
 
    Save "discovered" host keys in $(config)/ssh_host_keys/.
175
 
    """
176
 
    global SYSTEM_HOSTKEYS, BZR_HOSTKEYS
177
 
    bzr_hostkey_path = pathjoin(config_dir(), 'ssh_host_keys')
178
 
    ensure_config_dir_exists()
179
 
 
180
 
    try:
181
 
        f = open(bzr_hostkey_path, 'w')
182
 
        f.write('# SSH host keys collected by bzr\n')
183
 
        for hostname, keys in BZR_HOSTKEYS.iteritems():
184
 
            for keytype, key in keys.iteritems():
185
 
                f.write('%s %s %s\n' % (hostname, keytype, key.get_base64()))
186
 
        f.close()
187
 
    except IOError, e:
188
 
        mutter('failed to save bzr host keys: ' + str(e))
189
 
 
 
88
 
 
89
 
 
90
_paramiko_version = getattr(paramiko, '__version_info__', (0, 0, 0))
 
91
# don't use prefetch unless paramiko version >= 1.5.5 (there were bugs earlier)
 
92
_default_do_prefetch = (_paramiko_version >= (1, 5, 5))
 
93
 
 
94
 
 
95
@deprecated_function(zero_ninety)
 
96
def clear_connection_cache():
 
97
    """Remove all hosts from the SFTP connection cache.
 
98
 
 
99
    Primarily useful for test cases wanting to force garbage collection.
 
100
    We don't have a global connection cache anymore.
 
101
    """
190
102
 
191
103
class SFTPLock(object):
192
 
    """This fakes a lock in a remote location."""
 
104
    """This fakes a lock in a remote location.
 
105
    
 
106
    A present lock is indicated just by the existence of a file.  This
 
107
    doesn't work well on all transports and they are only used in 
 
108
    deprecated storage formats.
 
109
    """
 
110
    
193
111
    __slots__ = ['path', 'lock_path', 'lock_file', 'transport']
 
112
 
194
113
    def __init__(self, path, transport):
195
114
        assert isinstance(transport, SFTPTransport)
196
115
 
199
118
        self.lock_path = path + '.write-lock'
200
119
        self.transport = transport
201
120
        try:
202
 
            abspath = transport._abspath(self.lock_path)
 
121
            # RBC 20060103 FIXME should we be using private methods here ?
 
122
            abspath = transport._remote_path(self.lock_path)
203
123
            self.lock_file = transport._sftp_open_exclusive(abspath)
204
124
        except FileExists:
205
125
            raise LockError('File %r already locked' % (self.path,))
207
127
    def __del__(self):
208
128
        """Should this warn, or actually try to cleanup?"""
209
129
        if self.lock_file:
210
 
            warn("SFTPLock %r not explicitly unlocked" % (self.path,))
 
130
            warning("SFTPLock %r not explicitly unlocked" % (self.path,))
211
131
            self.unlock()
212
132
 
213
133
    def unlock(self):
221
141
            # What specific errors should we catch here?
222
142
            pass
223
143
 
224
 
class SFTPTransport (Transport):
225
 
    """
226
 
    Transport implementation for SFTP access.
227
 
    """
228
 
    _do_prefetch = False # Right now Paramiko's prefetch support causes things to hang
229
 
 
230
 
    def __init__(self, base, clone_from=None):
 
144
 
 
145
class SFTPTransport(ConnectedTransport):
 
146
    """Transport implementation for SFTP access."""
 
147
 
 
148
    _do_prefetch = _default_do_prefetch
 
149
    # TODO: jam 20060717 Conceivably these could be configurable, either
 
150
    #       by auto-tuning at run-time, or by a configuration (per host??)
 
151
    #       but the performance curve is pretty flat, so just going with
 
152
    #       reasonable defaults.
 
153
    _max_readv_combine = 200
 
154
    # Having to round trip to the server means waiting for a response,
 
155
    # so it is better to download extra bytes.
 
156
    # 8KiB had good performance for both local and remote network operations
 
157
    _bytes_to_read_before_seek = 8192
 
158
 
 
159
    # The sftp spec says that implementations SHOULD allow reads
 
160
    # to be at least 32K. paramiko.readv() does an async request
 
161
    # for the chunks. So we need to keep it within a single request
 
162
    # size for paramiko <= 1.6.1. paramiko 1.6.2 will probably chop
 
163
    # up the request itself, rather than us having to worry about it
 
164
    _max_request_size = 32768
 
165
 
 
166
    def __init__(self, base, _from_transport=None):
231
167
        assert base.startswith('sftp://')
232
 
        self._parse_url(base)
233
 
        base = self._unparse_url()
234
 
        super(SFTPTransport, self).__init__(base)
235
 
        if clone_from is None:
236
 
            self._sftp_connect()
237
 
        else:
238
 
            # use the same ssh connection, etc
239
 
            self._sftp = clone_from._sftp
240
 
        # super saves 'self.base'
241
 
    
242
 
    def should_cache(self):
243
 
        """
244
 
        Return True if the data pulled across should be cached locally.
245
 
        """
246
 
        return True
247
 
 
248
 
    def clone(self, offset=None):
249
 
        """
250
 
        Return a new SFTPTransport with root at self.base + offset.
251
 
        We share the same SFTP session between such transports, because it's
252
 
        fairly expensive to set them up.
253
 
        """
254
 
        if offset is None:
255
 
            return SFTPTransport(self.base, self)
256
 
        else:
257
 
            return SFTPTransport(self.abspath(offset), self)
258
 
 
259
 
    def abspath(self, relpath):
260
 
        """
261
 
        Return the full url to the given relative path.
 
168
        super(SFTPTransport, self).__init__(base,
 
169
                                            _from_transport=_from_transport)
 
170
 
 
171
    def _remote_path(self, relpath):
 
172
        """Return the path to be passed along the sftp protocol for relpath.
262
173
        
263
 
        @param relpath: the relative path or path components
264
 
        @type relpath: str or list
265
 
        """
266
 
        return self._unparse_url(self._abspath(relpath))
267
 
    
268
 
    def _abspath(self, relpath):
269
 
        """Return the absolute path segment without the SFTP URL."""
270
 
        # FIXME: share the common code across transports
271
 
        assert isinstance(relpath, basestring)
272
 
        relpath = [urllib.unquote(relpath)]
273
 
        basepath = self._path.split('/')
274
 
        if len(basepath) > 0 and basepath[-1] == '':
275
 
            basepath = basepath[:-1]
276
 
 
277
 
        for p in relpath:
278
 
            if p == '..':
279
 
                if len(basepath) == 0:
280
 
                    # In most filesystems, a request for the parent
281
 
                    # of root, just returns root.
282
 
                    continue
283
 
                basepath.pop()
284
 
            elif p == '.':
285
 
                continue # No-op
286
 
            else:
287
 
                basepath.append(p)
288
 
 
289
 
        path = '/'.join(basepath)
290
 
        # could still be a "relative" path here, but relative on the sftp server
291
 
        return path
292
 
 
293
 
    def relpath(self, abspath):
294
 
        username, password, host, port, path = self._split_url(abspath)
295
 
        error = []
296
 
        if (username != self._username):
297
 
            error.append('username mismatch')
298
 
        if (host != self._host):
299
 
            error.append('host mismatch')
300
 
        if (port != self._port):
301
 
            error.append('port mismatch')
302
 
        if (not path.startswith(self._path)):
303
 
            error.append('path mismatch')
304
 
        if error:
305
 
            extra = ': ' + ', '.join(error)
306
 
            raise PathNotChild(abspath, self.base, extra=extra)
307
 
        pl = len(self._path)
308
 
        return path[pl:].lstrip('/')
 
174
        :param relpath: is a urlencoded string.
 
175
        """
 
176
        relative = urlutils.unescape(relpath).encode('utf-8')
 
177
        remote_path = self._combine_paths(self._path, relative)
 
178
        # the initial slash should be removed from the path, and treated as a
 
179
        # homedir relative path (the path begins with a double slash if it is
 
180
        # absolute).  see draft-ietf-secsh-scp-sftp-ssh-uri-03.txt
 
181
        # RBC 20060118 we are not using this as its too user hostile. instead
 
182
        # we are following lftp and using /~/foo to mean '~/foo'
 
183
        # vila--20070602 and leave absolute paths begin with a single slash.
 
184
        if remote_path.startswith('/~/'):
 
185
            remote_path = remote_path[3:]
 
186
        elif remote_path == '/~':
 
187
            remote_path = ''
 
188
        return remote_path
 
189
 
 
190
    def _create_connection(self, credentials=None):
 
191
        """Create a new connection with the provided credentials.
 
192
 
 
193
        :param credentials: The credentials needed to establish the connection.
 
194
 
 
195
        :return: The created connection and its associated credentials.
 
196
 
 
197
        The credentials are only the password as it may have been entered
 
198
        interactively by the user and may be different from the one provided
 
199
        in base url at transport creation time.
 
200
        """
 
201
        if credentials is None:
 
202
            password = self._password
 
203
        else:
 
204
            password = credentials
 
205
 
 
206
        vendor = ssh._get_ssh_vendor()
 
207
        connection = vendor.connect_sftp(self._user, password,
 
208
                                         self._host, self._port)
 
209
        return connection, password
 
210
 
 
211
    def _get_sftp(self):
 
212
        """Ensures that a connection is established"""
 
213
        connection = self._get_connection()
 
214
        if connection is None:
 
215
            # First connection ever
 
216
            connection, credentials = self._create_connection()
 
217
            self._set_connection(connection, credentials)
 
218
        return connection
309
219
 
310
220
    def has(self, relpath):
311
221
        """
312
222
        Does the target location exist?
313
223
        """
314
224
        try:
315
 
            self._sftp.stat(self._abspath(relpath))
 
225
            self._get_sftp().stat(self._remote_path(relpath))
316
226
            return True
317
227
        except IOError:
318
228
            return False
319
229
 
320
 
    def get(self, relpath, decode=False):
 
230
    def get(self, relpath):
321
231
        """
322
232
        Get the file at the given relative path.
323
233
 
324
234
        :param relpath: The relative path to the file
325
235
        """
326
236
        try:
327
 
            path = self._abspath(relpath)
328
 
            f = self._sftp.file(path, mode='rb')
329
 
            if self._do_prefetch and hasattr(f, 'prefetch'):
 
237
            path = self._remote_path(relpath)
 
238
            f = self._get_sftp().file(path, mode='rb')
 
239
            if self._do_prefetch and (getattr(f, 'prefetch', None) is not None):
330
240
                f.prefetch()
331
241
            return f
332
242
        except (IOError, paramiko.SSHException), e:
 
243
            self._translate_io_exception(e, path, ': error retrieving',
 
244
                failure_exc=errors.ReadError)
 
245
 
 
246
    def _readv(self, relpath, offsets):
 
247
        """See Transport.readv()"""
 
248
        # We overload the default readv() because we want to use a file
 
249
        # that does not have prefetch enabled.
 
250
        # Also, if we have a new paramiko, it implements an async readv()
 
251
        if not offsets:
 
252
            return
 
253
 
 
254
        try:
 
255
            path = self._remote_path(relpath)
 
256
            fp = self._get_sftp().file(path, mode='rb')
 
257
            readv = getattr(fp, 'readv', None)
 
258
            if readv:
 
259
                return self._sftp_readv(fp, offsets, relpath)
 
260
            mutter('seek and read %s offsets', len(offsets))
 
261
            return self._seek_and_read(fp, offsets, relpath)
 
262
        except (IOError, paramiko.SSHException), e:
333
263
            self._translate_io_exception(e, path, ': error retrieving')
334
264
 
335
 
    def get_partial(self, relpath, start, length=None):
336
 
        """
337
 
        Get just part of a file.
338
 
 
339
 
        :param relpath: Path to the file, relative to base
340
 
        :param start: The starting position to read from
341
 
        :param length: The length to read. A length of None indicates
342
 
                       read to the end of the file.
343
 
        :return: A file-like object containing at least the specified bytes.
344
 
                 Some implementations may return objects which can be read
345
 
                 past this length, but this is not guaranteed.
346
 
        """
347
 
        # TODO: implement get_partial_multi to help with knit support
348
 
        f = self.get(relpath)
349
 
        f.seek(start)
350
 
        if self._do_prefetch and hasattr(f, 'prefetch'):
351
 
            f.prefetch()
352
 
        return f
353
 
 
354
 
    def put(self, relpath, f, mode=None):
355
 
        """
356
 
        Copy the file-like or string object into the location.
 
265
    def recommended_page_size(self):
 
266
        """See Transport.recommended_page_size().
 
267
 
 
268
        For SFTP we suggest a large page size to reduce the overhead
 
269
        introduced by latency.
 
270
        """
 
271
        return 64 * 1024
 
272
 
 
273
    def _sftp_readv(self, fp, offsets, relpath='<unknown>'):
 
274
        """Use the readv() member of fp to do async readv.
 
275
 
 
276
        And then read them using paramiko.readv(). paramiko.readv()
 
277
        does not support ranges > 64K, so it caps the request size, and
 
278
        just reads until it gets all the stuff it wants
 
279
        """
 
280
        offsets = list(offsets)
 
281
        sorted_offsets = sorted(offsets)
 
282
 
 
283
        # The algorithm works as follows:
 
284
        # 1) Coalesce nearby reads into a single chunk
 
285
        #    This generates a list of combined regions, the total size
 
286
        #    and the size of the sub regions. This coalescing step is limited
 
287
        #    in the number of nearby chunks to combine, and is allowed to
 
288
        #    skip small breaks in the requests. Limiting it makes sure that
 
289
        #    we can start yielding some data earlier, and skipping means we
 
290
        #    make fewer requests. (Beneficial even when using async)
 
291
        # 2) Break up this combined regions into chunks that are smaller
 
292
        #    than 64KiB. Technically the limit is 65536, but we are a
 
293
        #    little bit conservative. This is because sftp has a maximum
 
294
        #    return chunk size of 64KiB (max size of an unsigned short)
 
295
        # 3) Issue a readv() to paramiko to create an async request for
 
296
        #    all of this data
 
297
        # 4) Read in the data as it comes back, until we've read one
 
298
        #    continuous section as determined in step 1
 
299
        # 5) Break up the full sections into hunks for the original requested
 
300
        #    offsets. And put them in a cache
 
301
        # 6) Check if the next request is in the cache, and if it is, remove
 
302
        #    it from the cache, and yield its data. Continue until no more
 
303
        #    entries are in the cache.
 
304
        # 7) loop back to step 4 until all data has been read
 
305
        #
 
306
        # TODO: jam 20060725 This could be optimized one step further, by
 
307
        #       attempting to yield whatever data we have read, even before
 
308
        #       the first coallesced section has been fully processed.
 
309
 
 
310
        # When coalescing for use with readv(), we don't really need to
 
311
        # use any fudge factor, because the requests are made asynchronously
 
312
        coalesced = list(self._coalesce_offsets(sorted_offsets,
 
313
                               limit=self._max_readv_combine,
 
314
                               fudge_factor=0,
 
315
                               ))
 
316
        requests = []
 
317
        for c_offset in coalesced:
 
318
            start = c_offset.start
 
319
            size = c_offset.length
 
320
 
 
321
            # We need to break this up into multiple requests
 
322
            while size > 0:
 
323
                next_size = min(size, self._max_request_size)
 
324
                requests.append((start, next_size))
 
325
                size -= next_size
 
326
                start += next_size
 
327
 
 
328
        mutter('SFTP.readv() %s offsets => %s coalesced => %s requests',
 
329
                len(offsets), len(coalesced), len(requests))
 
330
 
 
331
        # Queue the current read until we have read the full coalesced section
 
332
        cur_data = []
 
333
        cur_data_len = 0
 
334
        cur_coalesced_stack = iter(coalesced)
 
335
        cur_coalesced = cur_coalesced_stack.next()
 
336
 
 
337
        # Cache the results, but only until they have been fulfilled
 
338
        data_map = {}
 
339
        # turn the list of offsets into a stack
 
340
        offset_stack = iter(offsets)
 
341
        cur_offset_and_size = offset_stack.next()
 
342
 
 
343
        for data in fp.readv(requests):
 
344
            cur_data += data
 
345
            cur_data_len += len(data)
 
346
 
 
347
            if cur_data_len < cur_coalesced.length:
 
348
                continue
 
349
            assert cur_data_len == cur_coalesced.length, \
 
350
                "Somehow we read too much: %s != %s" % (cur_data_len,
 
351
                                                        cur_coalesced.length)
 
352
            all_data = ''.join(cur_data)
 
353
            cur_data = []
 
354
            cur_data_len = 0
 
355
 
 
356
            for suboffset, subsize in cur_coalesced.ranges:
 
357
                key = (cur_coalesced.start+suboffset, subsize)
 
358
                data_map[key] = all_data[suboffset:suboffset+subsize]
 
359
 
 
360
            # Now that we've read some data, see if we can yield anything back
 
361
            while cur_offset_and_size in data_map:
 
362
                this_data = data_map.pop(cur_offset_and_size)
 
363
                yield cur_offset_and_size[0], this_data
 
364
                cur_offset_and_size = offset_stack.next()
 
365
 
 
366
            # We read a coalesced entry, so mark it as done
 
367
            cur_coalesced = None
 
368
            # Now that we've read all of the data for this coalesced section
 
369
            # on to the next
 
370
            cur_coalesced = cur_coalesced_stack.next()
 
371
 
 
372
        if cur_coalesced is not None:
 
373
            raise errors.ShortReadvError(relpath, cur_coalesced.start,
 
374
                cur_coalesced.length, len(data))
 
375
 
 
376
    def put_file(self, relpath, f, mode=None):
 
377
        """
 
378
        Copy the file-like object into the location.
357
379
 
358
380
        :param relpath: Location to put the contents, relative to base.
359
 
        :param f:       File-like or string object.
 
381
        :param f:       File-like object.
360
382
        :param mode: The final mode for the file
361
383
        """
362
 
        final_path = self._abspath(relpath)
363
 
        self._put(final_path, f, mode=mode)
 
384
        final_path = self._remote_path(relpath)
 
385
        return self._put(final_path, f, mode=mode)
364
386
 
365
387
    def _put(self, abspath, f, mode=None):
366
388
        """Helper function so both put() and copy_abspaths can reuse the code"""
371
393
        try:
372
394
            try:
373
395
                fout.set_pipelined(True)
374
 
                self._pump(f, fout)
 
396
                length = self._pump(f, fout)
375
397
            except (IOError, paramiko.SSHException), e:
376
398
                self._translate_io_exception(e, tmp_abspath)
 
399
            # XXX: This doesn't truly help like we would like it to.
 
400
            #      The problem is that openssh strips sticky bits. So while we
 
401
            #      can properly set group write permission, we lose the group
 
402
            #      sticky bit. So it is probably best to stop chmodding, and
 
403
            #      just tell users that they need to set the umask correctly.
 
404
            #      The attr.st_mode = mode, in _sftp_open_exclusive
 
405
            #      will handle when the user wants the final mode to be more 
 
406
            #      restrictive. And then we avoid a round trip. Unless 
 
407
            #      paramiko decides to expose an async chmod()
 
408
 
 
409
            # This is designed to chmod() right before we close.
 
410
            # Because we set_pipelined() earlier, theoretically we might 
 
411
            # avoid the round trip for fout.close()
377
412
            if mode is not None:
378
 
                self._sftp.chmod(tmp_abspath, mode)
 
413
                self._get_sftp().chmod(tmp_abspath, mode)
379
414
            fout.close()
380
415
            closed = True
381
 
            self._rename(tmp_abspath, abspath)
 
416
            self._rename_and_overwrite(tmp_abspath, abspath)
 
417
            return length
382
418
        except Exception, e:
383
419
            # If we fail, try to clean up the temporary file
384
420
            # before we throw the exception
390
426
            try:
391
427
                if not closed:
392
428
                    fout.close()
393
 
                self._sftp.remove(tmp_abspath)
 
429
                self._get_sftp().remove(tmp_abspath)
394
430
            except:
395
431
                # raise the saved except
396
432
                raise e
397
433
            # raise the original with its traceback if we can.
398
434
            raise
399
435
 
 
436
    def _put_non_atomic_helper(self, relpath, writer, mode=None,
 
437
                               create_parent_dir=False,
 
438
                               dir_mode=None):
 
439
        abspath = self._remote_path(relpath)
 
440
 
 
441
        # TODO: jam 20060816 paramiko doesn't publicly expose a way to
 
442
        #       set the file mode at create time. If it does, use it.
 
443
        #       But for now, we just chmod later anyway.
 
444
 
 
445
        def _open_and_write_file():
 
446
            """Try to open the target file, raise error on failure"""
 
447
            fout = None
 
448
            try:
 
449
                try:
 
450
                    fout = self._get_sftp().file(abspath, mode='wb')
 
451
                    fout.set_pipelined(True)
 
452
                    writer(fout)
 
453
                except (paramiko.SSHException, IOError), e:
 
454
                    self._translate_io_exception(e, abspath,
 
455
                                                 ': unable to open')
 
456
 
 
457
                # This is designed to chmod() right before we close.
 
458
                # Because we set_pipelined() earlier, theoretically we might 
 
459
                # avoid the round trip for fout.close()
 
460
                if mode is not None:
 
461
                    self._get_sftp().chmod(abspath, mode)
 
462
            finally:
 
463
                if fout is not None:
 
464
                    fout.close()
 
465
 
 
466
        if not create_parent_dir:
 
467
            _open_and_write_file()
 
468
            return
 
469
 
 
470
        # Try error handling to create the parent directory if we need to
 
471
        try:
 
472
            _open_and_write_file()
 
473
        except NoSuchFile:
 
474
            # Try to create the parent directory, and then go back to
 
475
            # writing the file
 
476
            parent_dir = os.path.dirname(abspath)
 
477
            self._mkdir(parent_dir, dir_mode)
 
478
            _open_and_write_file()
 
479
 
 
480
    def put_file_non_atomic(self, relpath, f, mode=None,
 
481
                            create_parent_dir=False,
 
482
                            dir_mode=None):
 
483
        """Copy the file-like object into the target location.
 
484
 
 
485
        This function is not strictly safe to use. It is only meant to
 
486
        be used when you already know that the target does not exist.
 
487
        It is not safe, because it will open and truncate the remote
 
488
        file. So there may be a time when the file has invalid contents.
 
489
 
 
490
        :param relpath: The remote location to put the contents.
 
491
        :param f:       File-like object.
 
492
        :param mode:    Possible access permissions for new file.
 
493
                        None means do not set remote permissions.
 
494
        :param create_parent_dir: If we cannot create the target file because
 
495
                        the parent directory does not exist, go ahead and
 
496
                        create it, and then try again.
 
497
        """
 
498
        def writer(fout):
 
499
            self._pump(f, fout)
 
500
        self._put_non_atomic_helper(relpath, writer, mode=mode,
 
501
                                    create_parent_dir=create_parent_dir,
 
502
                                    dir_mode=dir_mode)
 
503
 
 
504
    def put_bytes_non_atomic(self, relpath, bytes, mode=None,
 
505
                             create_parent_dir=False,
 
506
                             dir_mode=None):
 
507
        def writer(fout):
 
508
            fout.write(bytes)
 
509
        self._put_non_atomic_helper(relpath, writer, mode=mode,
 
510
                                    create_parent_dir=create_parent_dir,
 
511
                                    dir_mode=dir_mode)
 
512
 
400
513
    def iter_files_recursive(self):
401
514
        """Walk the relative paths of all files in this transport."""
402
515
        queue = list(self.list_dir('.'))
403
516
        while queue:
404
 
            relpath = urllib.quote(queue.pop(0))
 
517
            relpath = queue.pop(0)
405
518
            st = self.stat(relpath)
406
519
            if stat.S_ISDIR(st.st_mode):
407
520
                for i, basename in enumerate(self.list_dir(relpath)):
409
522
            else:
410
523
                yield relpath
411
524
 
 
525
    def _mkdir(self, abspath, mode=None):
 
526
        if mode is None:
 
527
            local_mode = 0777
 
528
        else:
 
529
            local_mode = mode
 
530
        try:
 
531
            self._get_sftp().mkdir(abspath, local_mode)
 
532
            if mode is not None:
 
533
                self._get_sftp().chmod(abspath, mode=mode)
 
534
        except (paramiko.SSHException, IOError), e:
 
535
            self._translate_io_exception(e, abspath, ': unable to mkdir',
 
536
                failure_exc=FileExists)
 
537
 
412
538
    def mkdir(self, relpath, mode=None):
413
539
        """Create a directory at the given path."""
 
540
        self._mkdir(self._remote_path(relpath), mode=mode)
 
541
 
 
542
    def open_write_stream(self, relpath, mode=None):
 
543
        """See Transport.open_write_stream."""
 
544
        # initialise the file to zero-length
 
545
        # this is three round trips, but we don't use this 
 
546
        # api more than once per write_group at the moment so 
 
547
        # it is a tolerable overhead. Better would be to truncate
 
548
        # the file after opening. RBC 20070805
 
549
        self.put_bytes_non_atomic(relpath, "", mode)
 
550
        abspath = self._remote_path(relpath)
 
551
        # TODO: jam 20060816 paramiko doesn't publicly expose a way to
 
552
        #       set the file mode at create time. If it does, use it.
 
553
        #       But for now, we just chmod later anyway.
 
554
        handle = None
414
555
        try:
415
 
            path = self._abspath(relpath)
416
 
            # In the paramiko documentation, it says that passing a mode flag 
417
 
            # will filtered against the server umask.
418
 
            # StubSFTPServer does not do this, which would be nice, because it is
419
 
            # what we really want :)
420
 
            # However, real servers do use umask, so we really should do it that way
421
 
            self._sftp.mkdir(path)
422
 
            if mode is not None:
423
 
                self._sftp.chmod(path, mode=mode)
 
556
            handle = self._get_sftp().file(abspath, mode='wb')
 
557
            handle.set_pipelined(True)
424
558
        except (paramiko.SSHException, IOError), e:
425
 
            self._translate_io_exception(e, path, ': unable to mkdir',
426
 
                failure_exc=FileExists)
 
559
            self._translate_io_exception(e, abspath,
 
560
                                         ': unable to open')
 
561
        _file_streams[self.abspath(relpath)] = handle
 
562
        return FileFileStream(self, relpath, handle)
427
563
 
428
 
    def _translate_io_exception(self, e, path, more_info='', failure_exc=NoSuchFile):
 
564
    def _translate_io_exception(self, e, path, more_info='',
 
565
                                failure_exc=PathError):
429
566
        """Translate a paramiko or IOError into a friendlier exception.
430
567
 
431
568
        :param e: The original exception
435
572
        :param failure_exc: Paramiko has the super fun ability to raise completely
436
573
                           opaque errors that just set "e.args = ('Failure',)" with
437
574
                           no more information.
438
 
                           This sometimes means FileExists, but it also sometimes
439
 
                           means NoSuchFile
 
575
                           If this parameter is set, it defines the exception 
 
576
                           to raise in these cases.
440
577
        """
441
578
        # paramiko seems to generate detailless errors.
442
579
        self._translate_error(e, path, raise_generic=False)
443
 
        if hasattr(e, 'args'):
 
580
        if getattr(e, 'args', None) is not None:
444
581
            if (e.args == ('No such file or directory',) or
445
582
                e.args == ('No such file',)):
446
583
                raise NoSuchFile(path, str(e) + more_info)
450
587
            if (e.args == ('Failure',)):
451
588
                raise failure_exc(path, str(e) + more_info)
452
589
            mutter('Raising exception with args %s', e.args)
453
 
        if hasattr(e, 'errno'):
 
590
        if getattr(e, 'errno', None) is not None:
454
591
            mutter('Raising exception with errno %s', e.errno)
455
592
        raise e
456
593
 
457
 
    def append(self, relpath, f):
 
594
    def append_file(self, relpath, f, mode=None):
458
595
        """
459
596
        Append the text in the file-like object into the final
460
597
        location.
461
598
        """
462
599
        try:
463
 
            path = self._abspath(relpath)
464
 
            fout = self._sftp.file(path, 'ab')
 
600
            path = self._remote_path(relpath)
 
601
            fout = self._get_sftp().file(path, 'ab')
 
602
            if mode is not None:
 
603
                self._get_sftp().chmod(path, mode)
 
604
            result = fout.tell()
465
605
            self._pump(f, fout)
 
606
            return result
466
607
        except (IOError, paramiko.SSHException), e:
467
608
            self._translate_io_exception(e, relpath, ': unable to append')
468
609
 
469
 
    def copy(self, rel_from, rel_to):
470
 
        """Copy the item at rel_from to the location at rel_to"""
471
 
        path_from = self._abspath(rel_from)
472
 
        path_to = self._abspath(rel_to)
473
 
        self._copy_abspaths(path_from, path_to)
474
 
 
475
 
    def _copy_abspaths(self, path_from, path_to, mode=None):
476
 
        """Copy files given an absolute path
477
 
 
478
 
        :param path_from: Path on remote server to read
479
 
        :param path_to: Path on remote server to write
480
 
        :return: None
481
 
 
482
 
        TODO: Should the destination location be atomically created?
483
 
              This has not been specified
484
 
        TODO: This should use some sort of remote copy, rather than
485
 
              pulling the data locally, and then writing it remotely
486
 
        """
 
610
    def rename(self, rel_from, rel_to):
 
611
        """Rename without special overwriting"""
487
612
        try:
488
 
            fin = self._sftp.file(path_from, 'rb')
489
 
            try:
490
 
                self._put(path_to, fin, mode=mode)
491
 
            finally:
492
 
                fin.close()
 
613
            self._get_sftp().rename(self._remote_path(rel_from),
 
614
                              self._remote_path(rel_to))
493
615
        except (IOError, paramiko.SSHException), e:
494
 
            self._translate_io_exception(e, path_from, ': unable copy to: %r' % path_to)
495
 
 
496
 
    def copy_to(self, relpaths, other, mode=None, pb=None):
497
 
        """Copy a set of entries from self into another Transport.
498
 
 
499
 
        :param relpaths: A list/generator of entries to be copied.
500
 
        """
501
 
        if isinstance(other, SFTPTransport) and other._sftp is self._sftp:
502
 
            # Both from & to are on the same remote filesystem
503
 
            # We can use a remote copy, instead of pulling locally, and pushing
504
 
 
505
 
            total = self._get_total(relpaths)
506
 
            count = 0
507
 
            for path in relpaths:
508
 
                path_from = self._abspath(relpath)
509
 
                path_to = other._abspath(relpath)
510
 
                self._update_pb(pb, 'copy-to', count, total)
511
 
                self._copy_abspaths(path_from, path_to, mode=mode)
512
 
                count += 1
513
 
            return count
514
 
        else:
515
 
            return super(SFTPTransport, self).copy_to(relpaths, other, mode=mode, pb=pb)
516
 
 
517
 
    def _rename(self, abs_from, abs_to):
 
616
            self._translate_io_exception(e, rel_from,
 
617
                    ': unable to rename to %r' % (rel_to))
 
618
 
 
619
    def _rename_and_overwrite(self, abs_from, abs_to):
518
620
        """Do a fancy rename on the remote server.
519
621
        
520
622
        Using the implementation provided by osutils.
521
623
        """
522
624
        try:
 
625
            sftp = self._get_sftp()
523
626
            fancy_rename(abs_from, abs_to,
524
 
                    rename_func=self._sftp.rename,
525
 
                    unlink_func=self._sftp.remove)
 
627
                         rename_func=sftp.rename,
 
628
                         unlink_func=sftp.remove)
526
629
        except (IOError, paramiko.SSHException), e:
527
 
            self._translate_io_exception(e, abs_from, ': unable to rename to %r' % (abs_to))
 
630
            self._translate_io_exception(e, abs_from,
 
631
                                         ': unable to rename to %r' % (abs_to))
528
632
 
529
633
    def move(self, rel_from, rel_to):
530
634
        """Move the item at rel_from to the location at rel_to"""
531
 
        path_from = self._abspath(rel_from)
532
 
        path_to = self._abspath(rel_to)
533
 
        self._rename(path_from, path_to)
 
635
        path_from = self._remote_path(rel_from)
 
636
        path_to = self._remote_path(rel_to)
 
637
        self._rename_and_overwrite(path_from, path_to)
534
638
 
535
639
    def delete(self, relpath):
536
640
        """Delete the item at relpath"""
537
 
        path = self._abspath(relpath)
 
641
        path = self._remote_path(relpath)
538
642
        try:
539
 
            self._sftp.remove(path)
 
643
            self._get_sftp().remove(path)
540
644
        except (IOError, paramiko.SSHException), e:
541
645
            self._translate_io_exception(e, path, ': unable to delete')
542
646
            
 
647
    def external_url(self):
 
648
        """See bzrlib.transport.Transport.external_url."""
 
649
        # the external path for SFTP is the base
 
650
        return self.base
 
651
 
543
652
    def listable(self):
544
653
        """Return True if this store supports listing."""
545
654
        return True
549
658
        Return a list of all files at the given location.
550
659
        """
551
660
        # does anything actually use this?
552
 
        path = self._abspath(relpath)
 
661
        # -- Unknown
 
662
        # This is at least used by copy_tree for remote upgrades.
 
663
        # -- David Allouche 2006-08-11
 
664
        path = self._remote_path(relpath)
553
665
        try:
554
 
            return self._sftp.listdir(path)
 
666
            entries = self._get_sftp().listdir(path)
555
667
        except (IOError, paramiko.SSHException), e:
556
668
            self._translate_io_exception(e, path, ': failed to list_dir')
 
669
        return [urlutils.escape(entry) for entry in entries]
 
670
 
 
671
    def rmdir(self, relpath):
 
672
        """See Transport.rmdir."""
 
673
        path = self._remote_path(relpath)
 
674
        try:
 
675
            return self._get_sftp().rmdir(path)
 
676
        except (IOError, paramiko.SSHException), e:
 
677
            self._translate_io_exception(e, path, ': failed to rmdir')
557
678
 
558
679
    def stat(self, relpath):
559
680
        """Return the stat information for a file."""
560
 
        path = self._abspath(relpath)
 
681
        path = self._remote_path(relpath)
561
682
        try:
562
 
            return self._sftp.stat(path)
 
683
            return self._get_sftp().stat(path)
563
684
        except (IOError, paramiko.SSHException), e:
564
685
            self._translate_io_exception(e, path, ': unable to stat')
565
686
 
589
710
        # that we have taken the lock.
590
711
        return SFTPLock(relpath, self)
591
712
 
592
 
 
593
 
    def _unparse_url(self, path=None):
594
 
        if path is None:
595
 
            path = self._path
596
 
        path = urllib.quote(path)
597
 
        if path.startswith('/'):
598
 
            path = '/%2F' + path[1:]
599
 
        else:
600
 
            path = '/' + path
601
 
        netloc = urllib.quote(self._host)
602
 
        if self._username is not None:
603
 
            netloc = '%s@%s' % (urllib.quote(self._username), netloc)
604
 
        if self._port is not None:
605
 
            netloc = '%s:%d' % (netloc, self._port)
606
 
 
607
 
        return urlparse.urlunparse(('sftp', netloc, path, '', '', ''))
608
 
 
609
 
    def _split_url(self, url):
610
 
        if isinstance(url, unicode):
611
 
            url = url.encode('utf-8')
612
 
        (scheme, netloc, path, params,
613
 
         query, fragment) = urlparse.urlparse(url, allow_fragments=False)
614
 
        assert scheme == 'sftp'
615
 
        username = password = host = port = None
616
 
        if '@' in netloc:
617
 
            username, host = netloc.split('@', 1)
618
 
            if ':' in username:
619
 
                username, password = username.split(':', 1)
620
 
                password = urllib.unquote(password)
621
 
            username = urllib.unquote(username)
622
 
        else:
623
 
            host = netloc
624
 
 
625
 
        if ':' in host:
626
 
            host, port = host.rsplit(':', 1)
627
 
            try:
628
 
                port = int(port)
629
 
            except ValueError:
630
 
                # TODO: Should this be ConnectionError?
631
 
                raise TransportError('%s: invalid port number' % port)
632
 
        host = urllib.unquote(host)
633
 
 
634
 
        path = urllib.unquote(path)
635
 
 
636
 
        # the initial slash should be removed from the path, and treated
637
 
        # as a homedir relative path (the path begins with a double slash
638
 
        # if it is absolute).
639
 
        # see draft-ietf-secsh-scp-sftp-ssh-uri-03.txt
640
 
        if path.startswith('/'):
641
 
            path = path[1:]
642
 
 
643
 
        return (username, password, host, port, path)
644
 
 
645
 
    def _parse_url(self, url):
646
 
        (self._username, self._password,
647
 
         self._host, self._port, self._path) = self._split_url(url)
648
 
 
649
 
    def _sftp_connect(self):
650
 
        """Connect to the remote sftp server.
651
 
        After this, self._sftp should have a valid connection (or
652
 
        we raise an TransportError 'could not connect').
653
 
 
654
 
        TODO: Raise a more reasonable ConnectionFailed exception
655
 
        """
656
 
        global _connected_hosts
657
 
 
658
 
        idx = (self._host, self._port, self._username)
659
 
        try:
660
 
            self._sftp = _connected_hosts[idx]
661
 
            return
662
 
        except KeyError:
663
 
            pass
664
 
        
665
 
        vendor = _get_ssh_vendor()
666
 
        if vendor != 'none':
667
 
            sock = SFTPSubprocess(self._host, self._port, self._username)
668
 
            self._sftp = SFTPClient(sock)
669
 
        else:
670
 
            self._paramiko_connect()
671
 
 
672
 
        _connected_hosts[idx] = self._sftp
673
 
 
674
 
    def _paramiko_connect(self):
675
 
        global SYSTEM_HOSTKEYS, BZR_HOSTKEYS
676
 
        
677
 
        load_host_keys()
678
 
 
679
 
        try:
680
 
            t = paramiko.Transport((self._host, self._port or 22))
681
 
            t.start_client()
682
 
        except paramiko.SSHException, e:
683
 
            raise ConnectionError('Unable to reach SSH host %s:%d' %
684
 
                                  (self._host, self._port), e)
685
 
            
686
 
        server_key = t.get_remote_server_key()
687
 
        server_key_hex = paramiko.util.hexify(server_key.get_fingerprint())
688
 
        keytype = server_key.get_name()
689
 
        if SYSTEM_HOSTKEYS.has_key(self._host) and SYSTEM_HOSTKEYS[self._host].has_key(keytype):
690
 
            our_server_key = SYSTEM_HOSTKEYS[self._host][keytype]
691
 
            our_server_key_hex = paramiko.util.hexify(our_server_key.get_fingerprint())
692
 
        elif BZR_HOSTKEYS.has_key(self._host) and BZR_HOSTKEYS[self._host].has_key(keytype):
693
 
            our_server_key = BZR_HOSTKEYS[self._host][keytype]
694
 
            our_server_key_hex = paramiko.util.hexify(our_server_key.get_fingerprint())
695
 
        else:
696
 
            warning('Adding %s host key for %s: %s' % (keytype, self._host, server_key_hex))
697
 
            if not BZR_HOSTKEYS.has_key(self._host):
698
 
                BZR_HOSTKEYS[self._host] = {}
699
 
            BZR_HOSTKEYS[self._host][keytype] = server_key
700
 
            our_server_key = server_key
701
 
            our_server_key_hex = paramiko.util.hexify(our_server_key.get_fingerprint())
702
 
            save_host_keys()
703
 
        if server_key != our_server_key:
704
 
            filename1 = os.path.expanduser('~/.ssh/known_hosts')
705
 
            filename2 = pathjoin(config_dir(), 'ssh_host_keys')
706
 
            raise TransportError('Host keys for %s do not match!  %s != %s' % \
707
 
                (self._host, our_server_key_hex, server_key_hex),
708
 
                ['Try editing %s or %s' % (filename1, filename2)])
709
 
 
710
 
        self._sftp_auth(t)
711
 
        
712
 
        try:
713
 
            self._sftp = t.open_sftp_client()
714
 
        except paramiko.SSHException, e:
715
 
            raise ConnectionError('Unable to start sftp client %s:%d' %
716
 
                                  (self._host, self._port), e)
717
 
 
718
 
    def _sftp_auth(self, transport):
719
 
        # paramiko requires a username, but it might be none if nothing was supplied
720
 
        # use the local username, just in case.
721
 
        # We don't override self._username, because if we aren't using paramiko,
722
 
        # the username might be specified in ~/.ssh/config and we don't want to
723
 
        # force it to something else
724
 
        # Also, it would mess up the self.relpath() functionality
725
 
        username = self._username or getpass.getuser()
726
 
 
727
 
        # Paramiko tries to open a socket.AF_UNIX in order to connect
728
 
        # to ssh-agent. That attribute doesn't exist on win32 (it does in cygwin)
729
 
        # so we get an AttributeError exception. For now, just don't try to
730
 
        # connect to an agent if we are on win32
731
 
        if sys.platform != 'win32':
732
 
            agent = paramiko.Agent()
733
 
            for key in agent.get_keys():
734
 
                mutter('Trying SSH agent key %s' % paramiko.util.hexify(key.get_fingerprint()))
735
 
                try:
736
 
                    transport.auth_publickey(username, key)
737
 
                    return
738
 
                except paramiko.SSHException, e:
739
 
                    pass
740
 
        
741
 
        # okay, try finding id_rsa or id_dss?  (posix only)
742
 
        if self._try_pkey_auth(transport, paramiko.RSAKey, username, 'id_rsa'):
743
 
            return
744
 
        if self._try_pkey_auth(transport, paramiko.DSSKey, username, 'id_dsa'):
745
 
            return
746
 
 
747
 
 
748
 
        if self._password:
749
 
            try:
750
 
                transport.auth_password(username, self._password)
751
 
                return
752
 
            except paramiko.SSHException, e:
753
 
                pass
754
 
 
755
 
            # FIXME: Don't keep a password held in memory if you can help it
756
 
            #self._password = None
757
 
 
758
 
        # give up and ask for a password
759
 
        password = bzrlib.ui.ui_factory.get_password(
760
 
                prompt='SSH %(user)s@%(host)s password',
761
 
                user=username, host=self._host)
762
 
        try:
763
 
            transport.auth_password(username, password)
764
 
        except paramiko.SSHException, e:
765
 
            raise ConnectionError('Unable to authenticate to SSH host as %s@%s' %
766
 
                                  (username, self._host), e)
767
 
 
768
 
    def _try_pkey_auth(self, transport, pkey_class, username, filename):
769
 
        filename = os.path.expanduser('~/.ssh/' + filename)
770
 
        try:
771
 
            key = pkey_class.from_private_key_file(filename)
772
 
            transport.auth_publickey(username, key)
773
 
            return True
774
 
        except paramiko.PasswordRequiredException:
775
 
            password = bzrlib.ui.ui_factory.get_password(
776
 
                    prompt='SSH %(filename)s password',
777
 
                    filename=filename)
778
 
            try:
779
 
                key = pkey_class.from_private_key_file(filename, password)
780
 
                transport.auth_publickey(username, key)
781
 
                return True
782
 
            except paramiko.SSHException:
783
 
                mutter('SSH authentication via %s key failed.' % (os.path.basename(filename),))
784
 
        except paramiko.SSHException:
785
 
            mutter('SSH authentication via %s key failed.' % (os.path.basename(filename),))
786
 
        except IOError:
787
 
            pass
788
 
        return False
789
 
 
790
713
    def _sftp_open_exclusive(self, abspath, mode=None):
791
714
        """Open a remote path exclusively.
792
715
 
801
724
        :param abspath: The remote absolute path where the file should be opened
802
725
        :param mode: The mode permissions bits for the new file
803
726
        """
804
 
        path = self._sftp._adjust_cwd(abspath)
 
727
        # TODO: jam 20060816 Paramiko >= 1.6.2 (probably earlier) supports
 
728
        #       using the 'x' flag to indicate SFTP_FLAG_EXCL.
 
729
        #       However, there is no way to set the permission mode at open 
 
730
        #       time using the sftp_client.file() functionality.
 
731
        path = self._get_sftp()._adjust_cwd(abspath)
 
732
        # mutter('sftp abspath %s => %s', abspath, path)
805
733
        attr = SFTPAttributes()
806
734
        if mode is not None:
807
735
            attr.st_mode = mode
808
736
        omode = (SFTP_FLAG_WRITE | SFTP_FLAG_CREATE 
809
737
                | SFTP_FLAG_TRUNC | SFTP_FLAG_EXCL)
810
738
        try:
811
 
            t, msg = self._sftp._request(CMD_OPEN, path, omode, attr)
 
739
            t, msg = self._get_sftp()._request(CMD_OPEN, path, omode, attr)
812
740
            if t != CMD_HANDLE:
813
741
                raise TransportError('Expected an SFTP handle')
814
742
            handle = msg.get_string()
815
 
            return SFTPFile(self._sftp, handle, 'wb', -1)
 
743
            return SFTPFile(self._get_sftp(), handle, 'wb', -1)
816
744
        except (paramiko.SSHException, IOError), e:
817
745
            self._translate_io_exception(e, abspath, ': unable to open',
818
746
                failure_exc=FileExists)
819
747
 
 
748
    def _can_roundtrip_unix_modebits(self):
 
749
        if sys.platform == 'win32':
 
750
            # anyone else?
 
751
            return False
 
752
        else:
 
753
            return True
 
754
 
 
755
# ------------- server test implementation --------------
 
756
import threading
 
757
 
 
758
from bzrlib.tests.stub_sftp import StubServer, StubSFTPServer
 
759
 
 
760
STUB_SERVER_KEY = """
 
761
-----BEGIN RSA PRIVATE KEY-----
 
762
MIICWgIBAAKBgQDTj1bqB4WmayWNPB+8jVSYpZYk80Ujvj680pOTh2bORBjbIAyz
 
763
oWGW+GUjzKxTiiPvVmxFgx5wdsFvF03v34lEVVhMpouqPAYQ15N37K/ir5XY+9m/
 
764
d8ufMCkjeXsQkKqFbAlQcnWMCRnOoPHS3I4vi6hmnDDeeYTSRvfLbW0fhwIBIwKB
 
765
gBIiOqZYaoqbeD9OS9z2K9KR2atlTxGxOJPXiP4ESqP3NVScWNwyZ3NXHpyrJLa0
 
766
EbVtzsQhLn6rF+TzXnOlcipFvjsem3iYzCpuChfGQ6SovTcOjHV9z+hnpXvQ/fon
 
767
soVRZY65wKnF7IAoUwTmJS9opqgrN6kRgCd3DASAMd1bAkEA96SBVWFt/fJBNJ9H
 
768
tYnBKZGw0VeHOYmVYbvMSstssn8un+pQpUm9vlG/bp7Oxd/m+b9KWEh2xPfv6zqU
 
769
avNwHwJBANqzGZa/EpzF4J8pGti7oIAPUIDGMtfIcmqNXVMckrmzQ2vTfqtkEZsA
 
770
4rE1IERRyiJQx6EJsz21wJmGV9WJQ5kCQQDwkS0uXqVdFzgHO6S++tjmjYcxwr3g
 
771
H0CoFYSgbddOT6miqRskOQF3DZVkJT3kyuBgU2zKygz52ukQZMqxCb1fAkASvuTv
 
772
qfpH87Qq5kQhNKdbbwbmd2NxlNabazPijWuphGTdW0VfJdWfklyS2Kr+iqrs/5wV
 
773
HhathJt636Eg7oIjAkA8ht3MQ+XSl9yIJIS8gVpbPxSw5OMfw0PjVE7tBdQruiSc
 
774
nvuQES5C9BMHjF39LZiGH1iLQy7FgdHyoP+eodI7
 
775
-----END RSA PRIVATE KEY-----
 
776
"""
 
777
 
 
778
 
 
779
class SocketListener(threading.Thread):
 
780
 
 
781
    def __init__(self, callback):
 
782
        threading.Thread.__init__(self)
 
783
        self._callback = callback
 
784
        self._socket = socket.socket()
 
785
        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 
786
        self._socket.bind(('localhost', 0))
 
787
        self._socket.listen(1)
 
788
        self.port = self._socket.getsockname()[1]
 
789
        self._stop_event = threading.Event()
 
790
 
 
791
    def stop(self):
 
792
        # called from outside this thread
 
793
        self._stop_event.set()
 
794
        # use a timeout here, because if the test fails, the server thread may
 
795
        # never notice the stop_event.
 
796
        self.join(5.0)
 
797
        self._socket.close()
 
798
 
 
799
    def run(self):
 
800
        while True:
 
801
            readable, writable_unused, exception_unused = \
 
802
                select.select([self._socket], [], [], 0.1)
 
803
            if self._stop_event.isSet():
 
804
                return
 
805
            if len(readable) == 0:
 
806
                continue
 
807
            try:
 
808
                s, addr_unused = self._socket.accept()
 
809
                # because the loopback socket is inline, and transports are
 
810
                # never explicitly closed, best to launch a new thread.
 
811
                threading.Thread(target=self._callback, args=(s,)).start()
 
812
            except socket.error, x:
 
813
                sys.excepthook(*sys.exc_info())
 
814
                warning('Socket error during accept() within unit test server'
 
815
                        ' thread: %r' % x)
 
816
            except Exception, x:
 
817
                # probably a failed test; unit test thread will log the
 
818
                # failure/error
 
819
                sys.excepthook(*sys.exc_info())
 
820
                warning('Exception from within unit test server thread: %r' % 
 
821
                        x)
 
822
 
 
823
 
 
824
class SocketDelay(object):
 
825
    """A socket decorator to make TCP appear slower.
 
826
 
 
827
    This changes recv, send, and sendall to add a fixed latency to each python
 
828
    call if a new roundtrip is detected. That is, when a recv is called and the
 
829
    flag new_roundtrip is set, latency is charged. Every send and send_all
 
830
    sets this flag.
 
831
 
 
832
    In addition every send, sendall and recv sleeps a bit per character send to
 
833
    simulate bandwidth.
 
834
 
 
835
    Not all methods are implemented, this is deliberate as this class is not a
 
836
    replacement for the builtin sockets layer. fileno is not implemented to
 
837
    prevent the proxy being bypassed. 
 
838
    """
 
839
 
 
840
    simulated_time = 0
 
841
    _proxied_arguments = dict.fromkeys([
 
842
        "close", "getpeername", "getsockname", "getsockopt", "gettimeout",
 
843
        "setblocking", "setsockopt", "settimeout", "shutdown"])
 
844
 
 
845
    def __init__(self, sock, latency, bandwidth=1.0, 
 
846
                 really_sleep=True):
 
847
        """ 
 
848
        :param bandwith: simulated bandwith (MegaBit)
 
849
        :param really_sleep: If set to false, the SocketDelay will just
 
850
        increase a counter, instead of calling time.sleep. This is useful for
 
851
        unittesting the SocketDelay.
 
852
        """
 
853
        self.sock = sock
 
854
        self.latency = latency
 
855
        self.really_sleep = really_sleep
 
856
        self.time_per_byte = 1 / (bandwidth / 8.0 * 1024 * 1024) 
 
857
        self.new_roundtrip = False
 
858
 
 
859
    def sleep(self, s):
 
860
        if self.really_sleep:
 
861
            time.sleep(s)
 
862
        else:
 
863
            SocketDelay.simulated_time += s
 
864
 
 
865
    def __getattr__(self, attr):
 
866
        if attr in SocketDelay._proxied_arguments:
 
867
            return getattr(self.sock, attr)
 
868
        raise AttributeError("'SocketDelay' object has no attribute %r" %
 
869
                             attr)
 
870
 
 
871
    def dup(self):
 
872
        return SocketDelay(self.sock.dup(), self.latency, self.time_per_byte,
 
873
                           self._sleep)
 
874
 
 
875
    def recv(self, *args):
 
876
        data = self.sock.recv(*args)
 
877
        if data and self.new_roundtrip:
 
878
            self.new_roundtrip = False
 
879
            self.sleep(self.latency)
 
880
        self.sleep(len(data) * self.time_per_byte)
 
881
        return data
 
882
 
 
883
    def sendall(self, data, flags=0):
 
884
        if not self.new_roundtrip:
 
885
            self.new_roundtrip = True
 
886
            self.sleep(self.latency)
 
887
        self.sleep(len(data) * self.time_per_byte)
 
888
        return self.sock.sendall(data, flags)
 
889
 
 
890
    def send(self, data, flags=0):
 
891
        if not self.new_roundtrip:
 
892
            self.new_roundtrip = True
 
893
            self.sleep(self.latency)
 
894
        bytes_sent = self.sock.send(data, flags)
 
895
        self.sleep(bytes_sent * self.time_per_byte)
 
896
        return bytes_sent
 
897
 
 
898
 
 
899
class SFTPServer(Server):
 
900
    """Common code for SFTP server facilities."""
 
901
 
 
902
    def __init__(self, server_interface=StubServer):
 
903
        self._original_vendor = None
 
904
        self._homedir = None
 
905
        self._server_homedir = None
 
906
        self._listener = None
 
907
        self._root = None
 
908
        self._vendor = ssh.ParamikoVendor()
 
909
        self._server_interface = server_interface
 
910
        # sftp server logs
 
911
        self.logs = []
 
912
        self.add_latency = 0
 
913
 
 
914
    def _get_sftp_url(self, path):
 
915
        """Calculate an sftp url to this server for path."""
 
916
        return 'sftp://foo:bar@localhost:%d/%s' % (self._listener.port, path)
 
917
 
 
918
    def log(self, message):
 
919
        """StubServer uses this to log when a new server is created."""
 
920
        self.logs.append(message)
 
921
 
 
922
    def _run_server_entry(self, sock):
 
923
        """Entry point for all implementations of _run_server.
 
924
        
 
925
        If self.add_latency is > 0.000001 then sock is given a latency adding
 
926
        decorator.
 
927
        """
 
928
        if self.add_latency > 0.000001:
 
929
            sock = SocketDelay(sock, self.add_latency)
 
930
        return self._run_server(sock)
 
931
 
 
932
    def _run_server(self, s):
 
933
        ssh_server = paramiko.Transport(s)
 
934
        key_file = pathjoin(self._homedir, 'test_rsa.key')
 
935
        f = open(key_file, 'w')
 
936
        f.write(STUB_SERVER_KEY)
 
937
        f.close()
 
938
        host_key = paramiko.RSAKey.from_private_key_file(key_file)
 
939
        ssh_server.add_server_key(host_key)
 
940
        server = self._server_interface(self)
 
941
        ssh_server.set_subsystem_handler('sftp', paramiko.SFTPServer,
 
942
                                         StubSFTPServer, root=self._root,
 
943
                                         home=self._server_homedir)
 
944
        event = threading.Event()
 
945
        ssh_server.start_server(event, server)
 
946
        event.wait(5.0)
 
947
    
 
948
    def setUp(self, backing_server=None):
 
949
        # XXX: TODO: make sftpserver back onto backing_server rather than local
 
950
        # disk.
 
951
        assert (backing_server is None or
 
952
                isinstance(backing_server, local.LocalURLServer)), (
 
953
            "backing_server should not be %r, because this can only serve the "
 
954
            "local current working directory." % (backing_server,))
 
955
        self._original_vendor = ssh._ssh_vendor_manager._cached_ssh_vendor
 
956
        ssh._ssh_vendor_manager._cached_ssh_vendor = self._vendor
 
957
        if sys.platform == 'win32':
 
958
            # Win32 needs to use the UNICODE api
 
959
            self._homedir = getcwd()
 
960
        else:
 
961
            # But Linux SFTP servers should just deal in bytestreams
 
962
            self._homedir = os.getcwd()
 
963
        if self._server_homedir is None:
 
964
            self._server_homedir = self._homedir
 
965
        self._root = '/'
 
966
        if sys.platform == 'win32':
 
967
            self._root = ''
 
968
        self._listener = SocketListener(self._run_server_entry)
 
969
        self._listener.setDaemon(True)
 
970
        self._listener.start()
 
971
 
 
972
    def tearDown(self):
 
973
        """See bzrlib.transport.Server.tearDown."""
 
974
        self._listener.stop()
 
975
        ssh._ssh_vendor_manager._cached_ssh_vendor = self._original_vendor
 
976
 
 
977
    def get_bogus_url(self):
 
978
        """See bzrlib.transport.Server.get_bogus_url."""
 
979
        # this is chosen to try to prevent trouble with proxies, wierd dns, etc
 
980
        # we bind a random socket, so that we get a guaranteed unused port
 
981
        # we just never listen on that port
 
982
        s = socket.socket()
 
983
        s.bind(('localhost', 0))
 
984
        return 'sftp://%s:%s/' % s.getsockname()
 
985
 
 
986
 
 
987
class SFTPFullAbsoluteServer(SFTPServer):
 
988
    """A test server for sftp transports, using absolute urls and ssh."""
 
989
 
 
990
    def get_url(self):
 
991
        """See bzrlib.transport.Server.get_url."""
 
992
        homedir = self._homedir
 
993
        if sys.platform != 'win32':
 
994
            # Remove the initial '/' on all platforms but win32
 
995
            homedir = homedir[1:]
 
996
        return self._get_sftp_url(urlutils.escape(homedir))
 
997
 
 
998
 
 
999
class SFTPServerWithoutSSH(SFTPServer):
 
1000
    """An SFTP server that uses a simple TCP socket pair rather than SSH."""
 
1001
 
 
1002
    def __init__(self):
 
1003
        super(SFTPServerWithoutSSH, self).__init__()
 
1004
        self._vendor = ssh.LoopbackVendor()
 
1005
 
 
1006
    def _run_server(self, sock):
 
1007
        # Re-import these as locals, so that they're still accessible during
 
1008
        # interpreter shutdown (when all module globals get set to None, leading
 
1009
        # to confusing errors like "'NoneType' object has no attribute 'error'".
 
1010
        class FakeChannel(object):
 
1011
            def get_transport(self):
 
1012
                return self
 
1013
            def get_log_channel(self):
 
1014
                return 'paramiko'
 
1015
            def get_name(self):
 
1016
                return '1'
 
1017
            def get_hexdump(self):
 
1018
                return False
 
1019
            def close(self):
 
1020
                pass
 
1021
 
 
1022
        server = paramiko.SFTPServer(FakeChannel(), 'sftp', StubServer(self), StubSFTPServer,
 
1023
                                     root=self._root, home=self._server_homedir)
 
1024
        try:
 
1025
            server.start_subsystem('sftp', None, sock)
 
1026
        except socket.error, e:
 
1027
            if (len(e.args) > 0) and (e.args[0] == errno.EPIPE):
 
1028
                # it's okay for the client to disconnect abruptly
 
1029
                # (bug in paramiko 1.6: it should absorb this exception)
 
1030
                pass
 
1031
            else:
 
1032
                raise
 
1033
        except Exception, e:
 
1034
            # This typically seems to happen during interpreter shutdown, so
 
1035
            # most of the useful ways to report this error are won't work.
 
1036
            # Writing the exception type, and then the text of the exception,
 
1037
            # seems to be the best we can do.
 
1038
            import sys
 
1039
            sys.stderr.write('\nEXCEPTION %r: ' % (e.__class__,))
 
1040
            sys.stderr.write('%s\n\n' % (e,))
 
1041
        server.finish_subsystem()
 
1042
 
 
1043
 
 
1044
class SFTPAbsoluteServer(SFTPServerWithoutSSH):
 
1045
    """A test server for sftp transports, using absolute urls."""
 
1046
 
 
1047
    def get_url(self):
 
1048
        """See bzrlib.transport.Server.get_url."""
 
1049
        homedir = self._homedir
 
1050
        if sys.platform != 'win32':
 
1051
            # Remove the initial '/' on all platforms but win32
 
1052
            homedir = homedir[1:]
 
1053
        return self._get_sftp_url(urlutils.escape(homedir))
 
1054
 
 
1055
 
 
1056
class SFTPHomeDirServer(SFTPServerWithoutSSH):
 
1057
    """A test server for sftp transports, using homedir relative urls."""
 
1058
 
 
1059
    def get_url(self):
 
1060
        """See bzrlib.transport.Server.get_url."""
 
1061
        return self._get_sftp_url("~/")
 
1062
 
 
1063
 
 
1064
class SFTPSiblingAbsoluteServer(SFTPAbsoluteServer):
 
1065
    """A test server for sftp transports where only absolute paths will work.
 
1066
 
 
1067
    It does this by serving from a deeply-nested directory that doesn't exist.
 
1068
    """
 
1069
 
 
1070
    def setUp(self, backing_server=None):
 
1071
        self._server_homedir = '/dev/noone/runs/tests/here'
 
1072
        super(SFTPSiblingAbsoluteServer, self).setUp(backing_server)
 
1073
 
 
1074
 
 
1075
def get_test_permutations():
 
1076
    """Return the permutations to be used in testing."""
 
1077
    return [(SFTPTransport, SFTPAbsoluteServer),
 
1078
            (SFTPTransport, SFTPHomeDirServer),
 
1079
            (SFTPTransport, SFTPSiblingAbsoluteServer),
 
1080
            ]