← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/transport/sftp.py

  • Committer: Martin Pool
  • Date: 2009-01-13 03:11:04 UTC
  • mto: This revision was merged to the branch mainline in revision 3937.
  • Revision ID: mbp@sourcefrog.net-20090113031104-03my054s02i9l2pe
Bump version to 1.12 and add news template

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/transport/sftp.py
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005 Robey Pointer <robey@lag.net>
 
2
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
2
3
#
3
4
# This program is free software; you can redistribute it and/or modify
4
5
# it under the terms of the GNU General Public License as published by
12
13
#
13
14
# You should have received a copy of the GNU General Public License
14
15
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
17
 
17
18
"""Implementation of Transport over SFTP, using paramiko."""
18
19
 
28
29
import itertools
29
30
import os
30
31
import random
 
32
import select
 
33
import socket
31
34
import stat
32
35
import sys
33
36
import time
37
40
 
38
41
from bzrlib import (
39
42
    config,
40
 
    debug,
41
43
    errors,
42
44
    urlutils,
43
45
    )
82
84
else:
83
85
    from paramiko.sftp import (SFTP_FLAG_WRITE, SFTP_FLAG_CREATE,
84
86
                               SFTP_FLAG_EXCL, SFTP_FLAG_TRUNC,
85
 
                               SFTP_OK, CMD_HANDLE, CMD_OPEN)
 
87
                               CMD_HANDLE, CMD_OPEN)
86
88
    from paramiko.sftp_attr import SFTPAttributes
87
89
    from paramiko.sftp_file import SFTPFile
88
90
 
94
96
 
95
97
class SFTPLock(object):
96
98
    """This fakes a lock in a remote location.
97
 
 
 
99
    
98
100
    A present lock is indicated just by the existence of a file.  This
99
 
    doesn't work well on all transports and they are only used in
 
101
    doesn't work well on all transports and they are only used in 
100
102
    deprecated storage formats.
101
103
    """
102
 
 
 
104
    
103
105
    __slots__ = ['path', 'lock_path', 'lock_file', 'transport']
104
106
 
105
107
    def __init__(self, path, transport):
138
140
    # See _get_requests for an explanation.
139
141
    _max_request_size = 32768
140
142
 
141
 
    def __init__(self, original_offsets, relpath, _report_activity):
 
143
    def __init__(self, original_offsets, relpath):
142
144
        """Create a new readv helper.
143
145
 
144
146
        :param original_offsets: The original requests given by the caller of
145
147
            readv()
146
148
        :param relpath: The name of the file (if known)
147
 
        :param _report_activity: A Transport._report_activity bound method,
148
 
            to be called as data arrives.
149
149
        """
150
150
        self.original_offsets = list(original_offsets)
151
151
        self.relpath = relpath
152
 
        self._report_activity = _report_activity
153
152
 
154
153
    def _get_requests(self):
155
154
        """Break up the offsets into individual requests over sftp.
181
180
                requests.append((start, next_size))
182
181
                size -= next_size
183
182
                start += next_size
184
 
        if 'sftp' in debug.debug_flags:
185
 
            mutter('SFTP.readv(%s) %s offsets => %s coalesced => %s requests',
186
 
                self.relpath, len(sorted_offsets), len(coalesced),
187
 
                len(requests))
 
183
        mutter('SFTP.readv(%s) %s offsets => %s coalesced => %s requests',
 
184
               self.relpath, len(sorted_offsets), len(coalesced),
 
185
               len(requests))
188
186
        return requests
189
187
 
190
188
    def request_and_yield_offsets(self, fp):
221
219
            if len(data) != length:
222
220
                raise errors.ShortReadvError(self.relpath,
223
221
                    start, length, len(data))
224
 
            self._report_activity(length, 'read')
225
222
            if last_end is None:
226
223
                # This is the first request, just buffer it
227
224
                buffered_data = [data]
286
283
            del buffered_data[:]
287
284
            data_chunks.append((input_start, buffered))
288
285
        if data_chunks:
289
 
            if 'sftp' in debug.debug_flags:
290
 
                mutter('SFTP readv left with %d out-of-order bytes',
291
 
                    sum(map(lambda x: len(x[1]), data_chunks)))
 
286
            mutter('SFTP readv left with %d out-of-order bytes',
 
287
                   sum(map(lambda x: len(x[1]), data_chunks)))
292
288
            # We've processed all the readv data, at this point, anything we
293
289
            # couldn't process is in data_chunks. This doesn't happen often, so
294
290
            # this code path isn't optimized
347
343
 
348
344
    def _remote_path(self, relpath):
349
345
        """Return the path to be passed along the sftp protocol for relpath.
350
 
 
 
346
        
351
347
        :param relpath: is a urlencoded string.
352
348
        """
353
349
        relative = urlutils.unescape(relpath).encode('utf-8')
404
400
        """
405
401
        try:
406
402
            self._get_sftp().stat(self._remote_path(relpath))
407
 
            # stat result is about 20 bytes, let's say
408
 
            self._report_activity(20, 'read')
409
403
            return True
410
404
        except IOError:
411
405
            return False
412
406
 
413
407
    def get(self, relpath):
414
 
        """Get the file at the given relative path.
 
408
        """
 
409
        Get the file at the given relative path.
415
410
 
416
411
        :param relpath: The relative path to the file
417
412
        """
418
413
        try:
419
 
            # FIXME: by returning the file directly, we don't pass this
420
 
            # through to report_activity.  We could try wrapping the object
421
 
            # before it's returned.  For readv and get_bytes it's handled in
422
 
            # the higher-level function.
423
 
            # -- mbp 20090126
424
414
            path = self._remote_path(relpath)
425
415
            f = self._get_sftp().file(path, mode='rb')
426
416
            if self._do_prefetch and (getattr(f, 'prefetch', None) is not None):
430
420
            self._translate_io_exception(e, path, ': error retrieving',
431
421
                failure_exc=errors.ReadError)
432
422
 
433
 
    def get_bytes(self, relpath):
434
 
        # reimplement this here so that we can report how many bytes came back
435
 
        f = self.get(relpath)
436
 
        try:
437
 
            bytes = f.read()
438
 
            self._report_activity(len(bytes), 'read')
439
 
            return bytes
440
 
        finally:
441
 
            f.close()
442
 
 
443
423
    def _readv(self, relpath, offsets):
444
424
        """See Transport.readv()"""
445
425
        # We overload the default readv() because we want to use a file
454
434
            readv = getattr(fp, 'readv', None)
455
435
            if readv:
456
436
                return self._sftp_readv(fp, offsets, relpath)
457
 
            if 'sftp' in debug.debug_flags:
458
 
                mutter('seek and read %s offsets', len(offsets))
 
437
            mutter('seek and read %s offsets', len(offsets))
459
438
            return self._seek_and_read(fp, offsets, relpath)
460
439
        except (IOError, paramiko.SSHException), e:
461
440
            self._translate_io_exception(e, path, ': error retrieving')
468
447
        """
469
448
        return 64 * 1024
470
449
 
471
 
    def _sftp_readv(self, fp, offsets, relpath):
 
450
    def _sftp_readv(self, fp, offsets, relpath='<unknown>'):
472
451
        """Use the readv() member of fp to do async readv.
473
452
 
474
 
        Then read them using paramiko.readv(). paramiko.readv()
 
453
        And then read them using paramiko.readv(). paramiko.readv()
475
454
        does not support ranges > 64K, so it caps the request size, and
476
 
        just reads until it gets all the stuff it wants.
 
455
        just reads until it gets all the stuff it wants
477
456
        """
478
 
        helper = _SFTPReadvHelper(offsets, relpath, self._report_activity)
 
457
        helper = _SFTPReadvHelper(offsets, relpath)
479
458
        return helper.request_and_yield_offsets(fp)
480
459
 
481
460
    def put_file(self, relpath, f, mode=None):
507
486
            #      sticky bit. So it is probably best to stop chmodding, and
508
487
            #      just tell users that they need to set the umask correctly.
509
488
            #      The attr.st_mode = mode, in _sftp_open_exclusive
510
 
            #      will handle when the user wants the final mode to be more
511
 
            #      restrictive. And then we avoid a round trip. Unless
 
489
            #      will handle when the user wants the final mode to be more 
 
490
            #      restrictive. And then we avoid a round trip. Unless 
512
491
            #      paramiko decides to expose an async chmod()
513
492
 
514
493
            # This is designed to chmod() right before we close.
515
 
            # Because we set_pipelined() earlier, theoretically we might
 
494
            # Because we set_pipelined() earlier, theoretically we might 
516
495
            # avoid the round trip for fout.close()
517
496
            if mode is not None:
518
497
                self._get_sftp().chmod(tmp_abspath, mode)
560
539
                                                 ': unable to open')
561
540
 
562
541
                # This is designed to chmod() right before we close.
563
 
                # Because we set_pipelined() earlier, theoretically we might
 
542
                # Because we set_pipelined() earlier, theoretically we might 
564
543
                # avoid the round trip for fout.close()
565
544
                if mode is not None:
566
545
                    self._get_sftp().chmod(abspath, mode)
617
596
 
618
597
    def iter_files_recursive(self):
619
598
        """Walk the relative paths of all files in this transport."""
620
 
        # progress is handled by list_dir
621
599
        queue = list(self.list_dir('.'))
622
600
        while queue:
623
601
            relpath = queue.pop(0)
634
612
        else:
635
613
            local_mode = mode
636
614
        try:
637
 
            self._report_activity(len(abspath), 'write')
638
615
            self._get_sftp().mkdir(abspath, local_mode)
639
 
            self._report_activity(1, 'read')
640
616
            if mode is not None:
641
617
                # chmod a dir through sftp will erase any sgid bit set
642
618
                # on the server side.  So, if the bit mode are already
664
640
    def open_write_stream(self, relpath, mode=None):
665
641
        """See Transport.open_write_stream."""
666
642
        # initialise the file to zero-length
667
 
        # this is three round trips, but we don't use this
668
 
        # api more than once per write_group at the moment so
 
643
        # this is three round trips, but we don't use this 
 
644
        # api more than once per write_group at the moment so 
669
645
        # it is a tolerable overhead. Better would be to truncate
670
646
        # the file after opening. RBC 20070805
671
647
        self.put_bytes_non_atomic(relpath, "", mode)
694
670
        :param failure_exc: Paramiko has the super fun ability to raise completely
695
671
                           opaque errors that just set "e.args = ('Failure',)" with
696
672
                           no more information.
697
 
                           If this parameter is set, it defines the exception
 
673
                           If this parameter is set, it defines the exception 
698
674
                           to raise in these cases.
699
675
        """
700
676
        # paramiko seems to generate detailless errors.
709
685
            # strange but true, for the paramiko server.
710
686
            if (e.args == ('Failure',)):
711
687
                raise failure_exc(path, str(e) + more_info)
712
 
            # Can be something like args = ('Directory not empty:
713
 
            # '/srv/bazaar.launchpad.net/blah...: '
714
 
            # [Errno 39] Directory not empty',)
715
 
            if (e.args[0].startswith('Directory not empty: ')
716
 
                or getattr(e, 'errno', None) == errno.ENOTEMPTY):
717
 
                raise errors.DirectoryNotEmpty(path, str(e))
718
688
            mutter('Raising exception with args %s', e.args)
719
689
        if getattr(e, 'errno', None) is not None:
720
690
            mutter('Raising exception with errno %s', e.errno)
747
717
 
748
718
    def _rename_and_overwrite(self, abs_from, abs_to):
749
719
        """Do a fancy rename on the remote server.
750
 
 
 
720
        
751
721
        Using the implementation provided by osutils.
752
722
        """
753
723
        try:
772
742
            self._get_sftp().remove(path)
773
743
        except (IOError, paramiko.SSHException), e:
774
744
            self._translate_io_exception(e, path, ': unable to delete')
775
 
 
 
745
            
776
746
    def external_url(self):
777
747
        """See bzrlib.transport.Transport.external_url."""
778
748
        # the external path for SFTP is the base
793
763
        path = self._remote_path(relpath)
794
764
        try:
795
765
            entries = self._get_sftp().listdir(path)
796
 
            self._report_activity(sum(map(len, entries)), 'read')
797
766
        except (IOError, paramiko.SSHException), e:
798
767
            self._translate_io_exception(e, path, ': failed to list_dir')
799
768
        return [urlutils.escape(entry) for entry in entries]
810
779
        """Return the stat information for a file."""
811
780
        path = self._remote_path(relpath)
812
781
        try:
813
 
            return self._get_sftp().lstat(path)
 
782
            return self._get_sftp().stat(path)
814
783
        except (IOError, paramiko.SSHException), e:
815
784
            self._translate_io_exception(e, path, ': unable to stat')
816
785
 
817
 
    def readlink(self, relpath):
818
 
        """See Transport.readlink."""
819
 
        path = self._remote_path(relpath)
820
 
        try:
821
 
            return self._get_sftp().readlink(path)
822
 
        except (IOError, paramiko.SSHException), e:
823
 
            self._translate_io_exception(e, path, ': unable to readlink')
824
 
 
825
 
    def symlink(self, source, link_name):
826
 
        """See Transport.symlink."""
827
 
        try:
828
 
            conn = self._get_sftp()
829
 
            sftp_retval = conn.symlink(source, link_name)
830
 
            if SFTP_OK != sftp_retval:
831
 
                raise TransportError(
832
 
                    '%r: unable to create symlink to %r' % (link_name, source),
833
 
                    sftp_retval
834
 
                )
835
 
        except (IOError, paramiko.SSHException), e:
836
 
            self._translate_io_exception(e, link_name,
837
 
                                         ': unable to create symlink to %r' % (source))
838
 
 
839
786
    def lock_read(self, relpath):
840
787
        """
841
788
        Lock the given file for shared (read) access.
878
825
        """
879
826
        # TODO: jam 20060816 Paramiko >= 1.6.2 (probably earlier) supports
880
827
        #       using the 'x' flag to indicate SFTP_FLAG_EXCL.
881
 
        #       However, there is no way to set the permission mode at open
 
828
        #       However, there is no way to set the permission mode at open 
882
829
        #       time using the sftp_client.file() functionality.
883
830
        path = self._get_sftp()._adjust_cwd(abspath)
884
831
        # mutter('sftp abspath %s => %s', abspath, path)
885
832
        attr = SFTPAttributes()
886
833
        if mode is not None:
887
834
            attr.st_mode = mode
888
 
        omode = (SFTP_FLAG_WRITE | SFTP_FLAG_CREATE
 
835
        omode = (SFTP_FLAG_WRITE | SFTP_FLAG_CREATE 
889
836
                | SFTP_FLAG_TRUNC | SFTP_FLAG_EXCL)
890
837
        try:
891
838
            t, msg = self._get_sftp()._request(CMD_OPEN, path, omode, attr)
904
851
        else:
905
852
            return True
906
853
 
 
854
# ------------- server test implementation --------------
 
855
import threading
 
856
 
 
857
from bzrlib.tests.stub_sftp import StubServer, StubSFTPServer
 
858
 
 
859
STUB_SERVER_KEY = """
 
860
-----BEGIN RSA PRIVATE KEY-----
 
861
MIICWgIBAAKBgQDTj1bqB4WmayWNPB+8jVSYpZYk80Ujvj680pOTh2bORBjbIAyz
 
862
oWGW+GUjzKxTiiPvVmxFgx5wdsFvF03v34lEVVhMpouqPAYQ15N37K/ir5XY+9m/
 
863
d8ufMCkjeXsQkKqFbAlQcnWMCRnOoPHS3I4vi6hmnDDeeYTSRvfLbW0fhwIBIwKB
 
864
gBIiOqZYaoqbeD9OS9z2K9KR2atlTxGxOJPXiP4ESqP3NVScWNwyZ3NXHpyrJLa0
 
865
EbVtzsQhLn6rF+TzXnOlcipFvjsem3iYzCpuChfGQ6SovTcOjHV9z+hnpXvQ/fon
 
866
soVRZY65wKnF7IAoUwTmJS9opqgrN6kRgCd3DASAMd1bAkEA96SBVWFt/fJBNJ9H
 
867
tYnBKZGw0VeHOYmVYbvMSstssn8un+pQpUm9vlG/bp7Oxd/m+b9KWEh2xPfv6zqU
 
868
avNwHwJBANqzGZa/EpzF4J8pGti7oIAPUIDGMtfIcmqNXVMckrmzQ2vTfqtkEZsA
 
869
4rE1IERRyiJQx6EJsz21wJmGV9WJQ5kCQQDwkS0uXqVdFzgHO6S++tjmjYcxwr3g
 
870
H0CoFYSgbddOT6miqRskOQF3DZVkJT3kyuBgU2zKygz52ukQZMqxCb1fAkASvuTv
 
871
qfpH87Qq5kQhNKdbbwbmd2NxlNabazPijWuphGTdW0VfJdWfklyS2Kr+iqrs/5wV
 
872
HhathJt636Eg7oIjAkA8ht3MQ+XSl9yIJIS8gVpbPxSw5OMfw0PjVE7tBdQruiSc
 
873
nvuQES5C9BMHjF39LZiGH1iLQy7FgdHyoP+eodI7
 
874
-----END RSA PRIVATE KEY-----
 
875
"""
 
876
 
 
877
 
 
878
class SocketListener(threading.Thread):
 
879
 
 
880
    def __init__(self, callback):
 
881
        threading.Thread.__init__(self)
 
882
        self._callback = callback
 
883
        self._socket = socket.socket()
 
884
        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 
885
        self._socket.bind(('localhost', 0))
 
886
        self._socket.listen(1)
 
887
        self.port = self._socket.getsockname()[1]
 
888
        self._stop_event = threading.Event()
 
889
 
 
890
    def stop(self):
 
891
        # called from outside this thread
 
892
        self._stop_event.set()
 
893
        # use a timeout here, because if the test fails, the server thread may
 
894
        # never notice the stop_event.
 
895
        self.join(5.0)
 
896
        self._socket.close()
 
897
 
 
898
    def run(self):
 
899
        while True:
 
900
            readable, writable_unused, exception_unused = \
 
901
                select.select([self._socket], [], [], 0.1)
 
902
            if self._stop_event.isSet():
 
903
                return
 
904
            if len(readable) == 0:
 
905
                continue
 
906
            try:
 
907
                s, addr_unused = self._socket.accept()
 
908
                # because the loopback socket is inline, and transports are
 
909
                # never explicitly closed, best to launch a new thread.
 
910
                threading.Thread(target=self._callback, args=(s,)).start()
 
911
            except socket.error, x:
 
912
                sys.excepthook(*sys.exc_info())
 
913
                warning('Socket error during accept() within unit test server'
 
914
                        ' thread: %r' % x)
 
915
            except Exception, x:
 
916
                # probably a failed test; unit test thread will log the
 
917
                # failure/error
 
918
                sys.excepthook(*sys.exc_info())
 
919
                warning('Exception from within unit test server thread: %r' % 
 
920
                        x)
 
921
 
 
922
 
 
923
class SocketDelay(object):
 
924
    """A socket decorator to make TCP appear slower.
 
925
 
 
926
    This changes recv, send, and sendall to add a fixed latency to each python
 
927
    call if a new roundtrip is detected. That is, when a recv is called and the
 
928
    flag new_roundtrip is set, latency is charged. Every send and send_all
 
929
    sets this flag.
 
930
 
 
931
    In addition every send, sendall and recv sleeps a bit per character send to
 
932
    simulate bandwidth.
 
933
 
 
934
    Not all methods are implemented, this is deliberate as this class is not a
 
935
    replacement for the builtin sockets layer. fileno is not implemented to
 
936
    prevent the proxy being bypassed. 
 
937
    """
 
938
 
 
939
    simulated_time = 0
 
940
    _proxied_arguments = dict.fromkeys([
 
941
        "close", "getpeername", "getsockname", "getsockopt", "gettimeout",
 
942
        "setblocking", "setsockopt", "settimeout", "shutdown"])
 
943
 
 
944
    def __init__(self, sock, latency, bandwidth=1.0, 
 
945
                 really_sleep=True):
 
946
        """ 
 
947
        :param bandwith: simulated bandwith (MegaBit)
 
948
        :param really_sleep: If set to false, the SocketDelay will just
 
949
        increase a counter, instead of calling time.sleep. This is useful for
 
950
        unittesting the SocketDelay.
 
951
        """
 
952
        self.sock = sock
 
953
        self.latency = latency
 
954
        self.really_sleep = really_sleep
 
955
        self.time_per_byte = 1 / (bandwidth / 8.0 * 1024 * 1024) 
 
956
        self.new_roundtrip = False
 
957
 
 
958
    def sleep(self, s):
 
959
        if self.really_sleep:
 
960
            time.sleep(s)
 
961
        else:
 
962
            SocketDelay.simulated_time += s
 
963
 
 
964
    def __getattr__(self, attr):
 
965
        if attr in SocketDelay._proxied_arguments:
 
966
            return getattr(self.sock, attr)
 
967
        raise AttributeError("'SocketDelay' object has no attribute %r" %
 
968
                             attr)
 
969
 
 
970
    def dup(self):
 
971
        return SocketDelay(self.sock.dup(), self.latency, self.time_per_byte,
 
972
                           self._sleep)
 
973
 
 
974
    def recv(self, *args):
 
975
        data = self.sock.recv(*args)
 
976
        if data and self.new_roundtrip:
 
977
            self.new_roundtrip = False
 
978
            self.sleep(self.latency)
 
979
        self.sleep(len(data) * self.time_per_byte)
 
980
        return data
 
981
 
 
982
    def sendall(self, data, flags=0):
 
983
        if not self.new_roundtrip:
 
984
            self.new_roundtrip = True
 
985
            self.sleep(self.latency)
 
986
        self.sleep(len(data) * self.time_per_byte)
 
987
        return self.sock.sendall(data, flags)
 
988
 
 
989
    def send(self, data, flags=0):
 
990
        if not self.new_roundtrip:
 
991
            self.new_roundtrip = True
 
992
            self.sleep(self.latency)
 
993
        bytes_sent = self.sock.send(data, flags)
 
994
        self.sleep(bytes_sent * self.time_per_byte)
 
995
        return bytes_sent
 
996
 
 
997
 
 
998
class SFTPServer(Server):
 
999
    """Common code for SFTP server facilities."""
 
1000
 
 
1001
    def __init__(self, server_interface=StubServer):
 
1002
        self._original_vendor = None
 
1003
        self._homedir = None
 
1004
        self._server_homedir = None
 
1005
        self._listener = None
 
1006
        self._root = None
 
1007
        self._vendor = ssh.ParamikoVendor()
 
1008
        self._server_interface = server_interface
 
1009
        # sftp server logs
 
1010
        self.logs = []
 
1011
        self.add_latency = 0
 
1012
 
 
1013
    def _get_sftp_url(self, path):
 
1014
        """Calculate an sftp url to this server for path."""
 
1015
        return 'sftp://foo:bar@localhost:%d/%s' % (self._listener.port, path)
 
1016
 
 
1017
    def log(self, message):
 
1018
        """StubServer uses this to log when a new server is created."""
 
1019
        self.logs.append(message)
 
1020
 
 
1021
    def _run_server_entry(self, sock):
 
1022
        """Entry point for all implementations of _run_server.
 
1023
        
 
1024
        If self.add_latency is > 0.000001 then sock is given a latency adding
 
1025
        decorator.
 
1026
        """
 
1027
        if self.add_latency > 0.000001:
 
1028
            sock = SocketDelay(sock, self.add_latency)
 
1029
        return self._run_server(sock)
 
1030
 
 
1031
    def _run_server(self, s):
 
1032
        ssh_server = paramiko.Transport(s)
 
1033
        key_file = pathjoin(self._homedir, 'test_rsa.key')
 
1034
        f = open(key_file, 'w')
 
1035
        f.write(STUB_SERVER_KEY)
 
1036
        f.close()
 
1037
        host_key = paramiko.RSAKey.from_private_key_file(key_file)
 
1038
        ssh_server.add_server_key(host_key)
 
1039
        server = self._server_interface(self)
 
1040
        ssh_server.set_subsystem_handler('sftp', paramiko.SFTPServer,
 
1041
                                         StubSFTPServer, root=self._root,
 
1042
                                         home=self._server_homedir)
 
1043
        event = threading.Event()
 
1044
        ssh_server.start_server(event, server)
 
1045
        event.wait(5.0)
 
1046
    
 
1047
    def setUp(self, backing_server=None):
 
1048
        # XXX: TODO: make sftpserver back onto backing_server rather than local
 
1049
        # disk.
 
1050
        if not (backing_server is None or
 
1051
                isinstance(backing_server, local.LocalURLServer)):
 
1052
            raise AssertionError(
 
1053
                "backing_server should not be %r, because this can only serve the "
 
1054
                "local current working directory." % (backing_server,))
 
1055
        self._original_vendor = ssh._ssh_vendor_manager._cached_ssh_vendor
 
1056
        ssh._ssh_vendor_manager._cached_ssh_vendor = self._vendor
 
1057
        if sys.platform == 'win32':
 
1058
            # Win32 needs to use the UNICODE api
 
1059
            self._homedir = getcwd()
 
1060
        else:
 
1061
            # But Linux SFTP servers should just deal in bytestreams
 
1062
            self._homedir = os.getcwd()
 
1063
        if self._server_homedir is None:
 
1064
            self._server_homedir = self._homedir
 
1065
        self._root = '/'
 
1066
        if sys.platform == 'win32':
 
1067
            self._root = ''
 
1068
        self._listener = SocketListener(self._run_server_entry)
 
1069
        self._listener.setDaemon(True)
 
1070
        self._listener.start()
 
1071
 
 
1072
    def tearDown(self):
 
1073
        """See bzrlib.transport.Server.tearDown."""
 
1074
        self._listener.stop()
 
1075
        ssh._ssh_vendor_manager._cached_ssh_vendor = self._original_vendor
 
1076
 
 
1077
    def get_bogus_url(self):
 
1078
        """See bzrlib.transport.Server.get_bogus_url."""
 
1079
        # this is chosen to try to prevent trouble with proxies, wierd dns, etc
 
1080
        # we bind a random socket, so that we get a guaranteed unused port
 
1081
        # we just never listen on that port
 
1082
        s = socket.socket()
 
1083
        s.bind(('localhost', 0))
 
1084
        return 'sftp://%s:%s/' % s.getsockname()
 
1085
 
 
1086
 
 
1087
class SFTPFullAbsoluteServer(SFTPServer):
 
1088
    """A test server for sftp transports, using absolute urls and ssh."""
 
1089
 
 
1090
    def get_url(self):
 
1091
        """See bzrlib.transport.Server.get_url."""
 
1092
        homedir = self._homedir
 
1093
        if sys.platform != 'win32':
 
1094
            # Remove the initial '/' on all platforms but win32
 
1095
            homedir = homedir[1:]
 
1096
        return self._get_sftp_url(urlutils.escape(homedir))
 
1097
 
 
1098
 
 
1099
class SFTPServerWithoutSSH(SFTPServer):
 
1100
    """An SFTP server that uses a simple TCP socket pair rather than SSH."""
 
1101
 
 
1102
    def __init__(self):
 
1103
        super(SFTPServerWithoutSSH, self).__init__()
 
1104
        self._vendor = ssh.LoopbackVendor()
 
1105
 
 
1106
    def _run_server(self, sock):
 
1107
        # Re-import these as locals, so that they're still accessible during
 
1108
        # interpreter shutdown (when all module globals get set to None, leading
 
1109
        # to confusing errors like "'NoneType' object has no attribute 'error'".
 
1110
        class FakeChannel(object):
 
1111
            def get_transport(self):
 
1112
                return self
 
1113
            def get_log_channel(self):
 
1114
                return 'paramiko'
 
1115
            def get_name(self):
 
1116
                return '1'
 
1117
            def get_hexdump(self):
 
1118
                return False
 
1119
            def close(self):
 
1120
                pass
 
1121
 
 
1122
        server = paramiko.SFTPServer(
 
1123
            FakeChannel(), 'sftp', StubServer(self), StubSFTPServer,
 
1124
            root=self._root, home=self._server_homedir)
 
1125
        try:
 
1126
            server.start_subsystem(
 
1127
                'sftp', None, ssh.SocketAsChannelAdapter(sock))
 
1128
        except socket.error, e:
 
1129
            if (len(e.args) > 0) and (e.args[0] == errno.EPIPE):
 
1130
                # it's okay for the client to disconnect abruptly
 
1131
                # (bug in paramiko 1.6: it should absorb this exception)
 
1132
                pass
 
1133
            else:
 
1134
                raise
 
1135
        except Exception, e:
 
1136
            # This typically seems to happen during interpreter shutdown, so
 
1137
            # most of the useful ways to report this error are won't work.
 
1138
            # Writing the exception type, and then the text of the exception,
 
1139
            # seems to be the best we can do.
 
1140
            import sys
 
1141
            sys.stderr.write('\nEXCEPTION %r: ' % (e.__class__,))
 
1142
            sys.stderr.write('%s\n\n' % (e,))
 
1143
        server.finish_subsystem()
 
1144
 
 
1145
 
 
1146
class SFTPAbsoluteServer(SFTPServerWithoutSSH):
 
1147
    """A test server for sftp transports, using absolute urls."""
 
1148
 
 
1149
    def get_url(self):
 
1150
        """See bzrlib.transport.Server.get_url."""
 
1151
        homedir = self._homedir
 
1152
        if sys.platform != 'win32':
 
1153
            # Remove the initial '/' on all platforms but win32
 
1154
            homedir = homedir[1:]
 
1155
        return self._get_sftp_url(urlutils.escape(homedir))
 
1156
 
 
1157
 
 
1158
class SFTPHomeDirServer(SFTPServerWithoutSSH):
 
1159
    """A test server for sftp transports, using homedir relative urls."""
 
1160
 
 
1161
    def get_url(self):
 
1162
        """See bzrlib.transport.Server.get_url."""
 
1163
        return self._get_sftp_url("~/")
 
1164
 
 
1165
 
 
1166
class SFTPSiblingAbsoluteServer(SFTPAbsoluteServer):
 
1167
    """A test server for sftp transports where only absolute paths will work.
 
1168
 
 
1169
    It does this by serving from a deeply-nested directory that doesn't exist.
 
1170
    """
 
1171
 
 
1172
    def setUp(self, backing_server=None):
 
1173
        self._server_homedir = '/dev/noone/runs/tests/here'
 
1174
        super(SFTPSiblingAbsoluteServer, self).setUp(backing_server)
 
1175
 
907
1176
 
908
1177
def get_test_permutations():
909
1178
    """Return the permutations to be used in testing."""
910
 
    from bzrlib.tests import stub_sftp
911
 
    return [(SFTPTransport, stub_sftp.SFTPAbsoluteServer),
912
 
            (SFTPTransport, stub_sftp.SFTPHomeDirServer),
913
 
            (SFTPTransport, stub_sftp.SFTPSiblingAbsoluteServer),
 
1179
    return [(SFTPTransport, SFTPAbsoluteServer),
 
1180
            (SFTPTransport, SFTPHomeDirServer),
 
1181
            (SFTPTransport, SFTPSiblingAbsoluteServer),
914
1182
            ]