/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/transport/sftp.py

  • Committer: Michael Hudson
  • Date: 2009-02-26 22:28:08 UTC
  • mto: This revision was merged to the branch mainline in revision 4121.
  • Revision ID: michael.hudson@canonical.com-20090226222808-url1udsulbge2r28
_this_ works

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005, 2006, 2007, 2008, 2009 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Implementation of Transport over SFTP, using paramiko."""
18
18
 
28
28
import itertools
29
29
import os
30
30
import random
 
31
import select
 
32
import socket
31
33
import stat
32
34
import sys
33
35
import time
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
 
404
406
        """
405
407
        try:
406
408
            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
409
            return True
410
410
        except IOError:
411
411
            return False
416
416
        :param relpath: The relative path to the file
417
417
        """
418
418
        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
419
            path = self._remote_path(relpath)
425
420
            f = self._get_sftp().file(path, mode='rb')
426
421
            if self._do_prefetch and (getattr(f, 'prefetch', None) is not None):
617
612
 
618
613
    def iter_files_recursive(self):
619
614
        """Walk the relative paths of all files in this transport."""
620
 
        # progress is handled by list_dir
621
615
        queue = list(self.list_dir('.'))
622
616
        while queue:
623
617
            relpath = queue.pop(0)
634
628
        else:
635
629
            local_mode = mode
636
630
        try:
637
 
            self._report_activity(len(abspath), 'write')
638
631
            self._get_sftp().mkdir(abspath, local_mode)
639
 
            self._report_activity(1, 'read')
640
632
            if mode is not None:
641
633
                # chmod a dir through sftp will erase any sgid bit set
642
634
                # on the server side.  So, if the bit mode are already
709
701
            # strange but true, for the paramiko server.
710
702
            if (e.args == ('Failure',)):
711
703
                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
704
            mutter('Raising exception with args %s', e.args)
719
705
        if getattr(e, 'errno', None) is not None:
720
706
            mutter('Raising exception with errno %s', e.errno)
793
779
        path = self._remote_path(relpath)
794
780
        try:
795
781
            entries = self._get_sftp().listdir(path)
796
 
            self._report_activity(sum(map(len, entries)), 'read')
797
782
        except (IOError, paramiko.SSHException), e:
798
783
            self._translate_io_exception(e, path, ': failed to list_dir')
799
784
        return [urlutils.escape(entry) for entry in entries]
810
795
        """Return the stat information for a file."""
811
796
        path = self._remote_path(relpath)
812
797
        try:
813
 
            return self._get_sftp().lstat(path)
 
798
            return self._get_sftp().stat(path)
814
799
        except (IOError, paramiko.SSHException), e:
815
800
            self._translate_io_exception(e, path, ': unable to stat')
816
801
 
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
802
    def lock_read(self, relpath):
840
803
        """
841
804
        Lock the given file for shared (read) access.
904
867
        else:
905
868
            return True
906
869
 
 
870
# ------------- server test implementation --------------
 
871
import threading
 
872
 
 
873
from bzrlib.tests.stub_sftp import StubServer, StubSFTPServer
 
874
 
 
875
STUB_SERVER_KEY = """
 
876
-----BEGIN RSA PRIVATE KEY-----
 
877
MIICWgIBAAKBgQDTj1bqB4WmayWNPB+8jVSYpZYk80Ujvj680pOTh2bORBjbIAyz
 
878
oWGW+GUjzKxTiiPvVmxFgx5wdsFvF03v34lEVVhMpouqPAYQ15N37K/ir5XY+9m/
 
879
d8ufMCkjeXsQkKqFbAlQcnWMCRnOoPHS3I4vi6hmnDDeeYTSRvfLbW0fhwIBIwKB
 
880
gBIiOqZYaoqbeD9OS9z2K9KR2atlTxGxOJPXiP4ESqP3NVScWNwyZ3NXHpyrJLa0
 
881
EbVtzsQhLn6rF+TzXnOlcipFvjsem3iYzCpuChfGQ6SovTcOjHV9z+hnpXvQ/fon
 
882
soVRZY65wKnF7IAoUwTmJS9opqgrN6kRgCd3DASAMd1bAkEA96SBVWFt/fJBNJ9H
 
883
tYnBKZGw0VeHOYmVYbvMSstssn8un+pQpUm9vlG/bp7Oxd/m+b9KWEh2xPfv6zqU
 
884
avNwHwJBANqzGZa/EpzF4J8pGti7oIAPUIDGMtfIcmqNXVMckrmzQ2vTfqtkEZsA
 
885
4rE1IERRyiJQx6EJsz21wJmGV9WJQ5kCQQDwkS0uXqVdFzgHO6S++tjmjYcxwr3g
 
886
H0CoFYSgbddOT6miqRskOQF3DZVkJT3kyuBgU2zKygz52ukQZMqxCb1fAkASvuTv
 
887
qfpH87Qq5kQhNKdbbwbmd2NxlNabazPijWuphGTdW0VfJdWfklyS2Kr+iqrs/5wV
 
888
HhathJt636Eg7oIjAkA8ht3MQ+XSl9yIJIS8gVpbPxSw5OMfw0PjVE7tBdQruiSc
 
889
nvuQES5C9BMHjF39LZiGH1iLQy7FgdHyoP+eodI7
 
890
-----END RSA PRIVATE KEY-----
 
891
"""
 
892
 
 
893
 
 
894
class SocketListener(threading.Thread):
 
895
 
 
896
    def __init__(self, callback):
 
897
        threading.Thread.__init__(self)
 
898
        self._callback = callback
 
899
        self._socket = socket.socket()
 
900
        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 
901
        self._socket.bind(('localhost', 0))
 
902
        self._socket.listen(1)
 
903
        self.port = self._socket.getsockname()[1]
 
904
        self._stop_event = threading.Event()
 
905
 
 
906
    def stop(self):
 
907
        # called from outside this thread
 
908
        self._stop_event.set()
 
909
        # use a timeout here, because if the test fails, the server thread may
 
910
        # never notice the stop_event.
 
911
        self.join(5.0)
 
912
        self._socket.close()
 
913
 
 
914
    def run(self):
 
915
        while True:
 
916
            readable, writable_unused, exception_unused = \
 
917
                select.select([self._socket], [], [], 0.1)
 
918
            if self._stop_event.isSet():
 
919
                return
 
920
            if len(readable) == 0:
 
921
                continue
 
922
            try:
 
923
                s, addr_unused = self._socket.accept()
 
924
                # because the loopback socket is inline, and transports are
 
925
                # never explicitly closed, best to launch a new thread.
 
926
                threading.Thread(target=self._callback, args=(s,)).start()
 
927
            except socket.error, x:
 
928
                sys.excepthook(*sys.exc_info())
 
929
                warning('Socket error during accept() within unit test server'
 
930
                        ' thread: %r' % x)
 
931
            except Exception, x:
 
932
                # probably a failed test; unit test thread will log the
 
933
                # failure/error
 
934
                sys.excepthook(*sys.exc_info())
 
935
                warning('Exception from within unit test server thread: %r' %
 
936
                        x)
 
937
 
 
938
 
 
939
class SocketDelay(object):
 
940
    """A socket decorator to make TCP appear slower.
 
941
 
 
942
    This changes recv, send, and sendall to add a fixed latency to each python
 
943
    call if a new roundtrip is detected. That is, when a recv is called and the
 
944
    flag new_roundtrip is set, latency is charged. Every send and send_all
 
945
    sets this flag.
 
946
 
 
947
    In addition every send, sendall and recv sleeps a bit per character send to
 
948
    simulate bandwidth.
 
949
 
 
950
    Not all methods are implemented, this is deliberate as this class is not a
 
951
    replacement for the builtin sockets layer. fileno is not implemented to
 
952
    prevent the proxy being bypassed.
 
953
    """
 
954
 
 
955
    simulated_time = 0
 
956
    _proxied_arguments = dict.fromkeys([
 
957
        "close", "getpeername", "getsockname", "getsockopt", "gettimeout",
 
958
        "setblocking", "setsockopt", "settimeout", "shutdown"])
 
959
 
 
960
    def __init__(self, sock, latency, bandwidth=1.0,
 
961
                 really_sleep=True):
 
962
        """
 
963
        :param bandwith: simulated bandwith (MegaBit)
 
964
        :param really_sleep: If set to false, the SocketDelay will just
 
965
        increase a counter, instead of calling time.sleep. This is useful for
 
966
        unittesting the SocketDelay.
 
967
        """
 
968
        self.sock = sock
 
969
        self.latency = latency
 
970
        self.really_sleep = really_sleep
 
971
        self.time_per_byte = 1 / (bandwidth / 8.0 * 1024 * 1024)
 
972
        self.new_roundtrip = False
 
973
 
 
974
    def sleep(self, s):
 
975
        if self.really_sleep:
 
976
            time.sleep(s)
 
977
        else:
 
978
            SocketDelay.simulated_time += s
 
979
 
 
980
    def __getattr__(self, attr):
 
981
        if attr in SocketDelay._proxied_arguments:
 
982
            return getattr(self.sock, attr)
 
983
        raise AttributeError("'SocketDelay' object has no attribute %r" %
 
984
                             attr)
 
985
 
 
986
    def dup(self):
 
987
        return SocketDelay(self.sock.dup(), self.latency, self.time_per_byte,
 
988
                           self._sleep)
 
989
 
 
990
    def recv(self, *args):
 
991
        data = self.sock.recv(*args)
 
992
        if data and self.new_roundtrip:
 
993
            self.new_roundtrip = False
 
994
            self.sleep(self.latency)
 
995
        self.sleep(len(data) * self.time_per_byte)
 
996
        return data
 
997
 
 
998
    def sendall(self, data, flags=0):
 
999
        if not self.new_roundtrip:
 
1000
            self.new_roundtrip = True
 
1001
            self.sleep(self.latency)
 
1002
        self.sleep(len(data) * self.time_per_byte)
 
1003
        return self.sock.sendall(data, flags)
 
1004
 
 
1005
    def send(self, data, flags=0):
 
1006
        if not self.new_roundtrip:
 
1007
            self.new_roundtrip = True
 
1008
            self.sleep(self.latency)
 
1009
        bytes_sent = self.sock.send(data, flags)
 
1010
        self.sleep(bytes_sent * self.time_per_byte)
 
1011
        return bytes_sent
 
1012
 
 
1013
 
 
1014
class SFTPServer(Server):
 
1015
    """Common code for SFTP server facilities."""
 
1016
 
 
1017
    def __init__(self, server_interface=StubServer):
 
1018
        self._original_vendor = None
 
1019
        self._homedir = None
 
1020
        self._server_homedir = None
 
1021
        self._listener = None
 
1022
        self._root = None
 
1023
        self._vendor = ssh.ParamikoVendor()
 
1024
        self._server_interface = server_interface
 
1025
        # sftp server logs
 
1026
        self.logs = []
 
1027
        self.add_latency = 0
 
1028
 
 
1029
    def _get_sftp_url(self, path):
 
1030
        """Calculate an sftp url to this server for path."""
 
1031
        return 'sftp://foo:bar@localhost:%d/%s' % (self._listener.port, path)
 
1032
 
 
1033
    def log(self, message):
 
1034
        """StubServer uses this to log when a new server is created."""
 
1035
        self.logs.append(message)
 
1036
 
 
1037
    def _run_server_entry(self, sock):
 
1038
        """Entry point for all implementations of _run_server.
 
1039
 
 
1040
        If self.add_latency is > 0.000001 then sock is given a latency adding
 
1041
        decorator.
 
1042
        """
 
1043
        if self.add_latency > 0.000001:
 
1044
            sock = SocketDelay(sock, self.add_latency)
 
1045
        return self._run_server(sock)
 
1046
 
 
1047
    def _run_server(self, s):
 
1048
        ssh_server = paramiko.Transport(s)
 
1049
        key_file = pathjoin(self._homedir, 'test_rsa.key')
 
1050
        f = open(key_file, 'w')
 
1051
        f.write(STUB_SERVER_KEY)
 
1052
        f.close()
 
1053
        host_key = paramiko.RSAKey.from_private_key_file(key_file)
 
1054
        ssh_server.add_server_key(host_key)
 
1055
        server = self._server_interface(self)
 
1056
        ssh_server.set_subsystem_handler('sftp', paramiko.SFTPServer,
 
1057
                                         StubSFTPServer, root=self._root,
 
1058
                                         home=self._server_homedir)
 
1059
        event = threading.Event()
 
1060
        ssh_server.start_server(event, server)
 
1061
        event.wait(5.0)
 
1062
 
 
1063
    def setUp(self, backing_server=None):
 
1064
        # XXX: TODO: make sftpserver back onto backing_server rather than local
 
1065
        # disk.
 
1066
        if not (backing_server is None or
 
1067
                isinstance(backing_server, local.LocalURLServer)):
 
1068
            raise AssertionError(
 
1069
                "backing_server should not be %r, because this can only serve the "
 
1070
                "local current working directory." % (backing_server,))
 
1071
        self._original_vendor = ssh._ssh_vendor_manager._cached_ssh_vendor
 
1072
        ssh._ssh_vendor_manager._cached_ssh_vendor = self._vendor
 
1073
        if sys.platform == 'win32':
 
1074
            # Win32 needs to use the UNICODE api
 
1075
            self._homedir = getcwd()
 
1076
        else:
 
1077
            # But Linux SFTP servers should just deal in bytestreams
 
1078
            self._homedir = os.getcwd()
 
1079
        if self._server_homedir is None:
 
1080
            self._server_homedir = self._homedir
 
1081
        self._root = '/'
 
1082
        if sys.platform == 'win32':
 
1083
            self._root = ''
 
1084
        self._listener = SocketListener(self._run_server_entry)
 
1085
        self._listener.setDaemon(True)
 
1086
        self._listener.start()
 
1087
 
 
1088
    def tearDown(self):
 
1089
        """See bzrlib.transport.Server.tearDown."""
 
1090
        self._listener.stop()
 
1091
        ssh._ssh_vendor_manager._cached_ssh_vendor = self._original_vendor
 
1092
 
 
1093
    def get_bogus_url(self):
 
1094
        """See bzrlib.transport.Server.get_bogus_url."""
 
1095
        # this is chosen to try to prevent trouble with proxies, wierd dns, etc
 
1096
        # we bind a random socket, so that we get a guaranteed unused port
 
1097
        # we just never listen on that port
 
1098
        s = socket.socket()
 
1099
        s.bind(('localhost', 0))
 
1100
        return 'sftp://%s:%s/' % s.getsockname()
 
1101
 
 
1102
 
 
1103
class SFTPFullAbsoluteServer(SFTPServer):
 
1104
    """A test server for sftp transports, using absolute urls and ssh."""
 
1105
 
 
1106
    def get_url(self):
 
1107
        """See bzrlib.transport.Server.get_url."""
 
1108
        homedir = self._homedir
 
1109
        if sys.platform != 'win32':
 
1110
            # Remove the initial '/' on all platforms but win32
 
1111
            homedir = homedir[1:]
 
1112
        return self._get_sftp_url(urlutils.escape(homedir))
 
1113
 
 
1114
 
 
1115
class SFTPServerWithoutSSH(SFTPServer):
 
1116
    """An SFTP server that uses a simple TCP socket pair rather than SSH."""
 
1117
 
 
1118
    def __init__(self):
 
1119
        super(SFTPServerWithoutSSH, self).__init__()
 
1120
        self._vendor = ssh.LoopbackVendor()
 
1121
 
 
1122
    def _run_server(self, sock):
 
1123
        # Re-import these as locals, so that they're still accessible during
 
1124
        # interpreter shutdown (when all module globals get set to None, leading
 
1125
        # to confusing errors like "'NoneType' object has no attribute 'error'".
 
1126
        class FakeChannel(object):
 
1127
            def get_transport(self):
 
1128
                return self
 
1129
            def get_log_channel(self):
 
1130
                return 'paramiko'
 
1131
            def get_name(self):
 
1132
                return '1'
 
1133
            def get_hexdump(self):
 
1134
                return False
 
1135
            def close(self):
 
1136
                pass
 
1137
 
 
1138
        server = paramiko.SFTPServer(
 
1139
            FakeChannel(), 'sftp', StubServer(self), StubSFTPServer,
 
1140
            root=self._root, home=self._server_homedir)
 
1141
        try:
 
1142
            server.start_subsystem(
 
1143
                'sftp', None, ssh.SocketAsChannelAdapter(sock))
 
1144
        except socket.error, e:
 
1145
            if (len(e.args) > 0) and (e.args[0] == errno.EPIPE):
 
1146
                # it's okay for the client to disconnect abruptly
 
1147
                # (bug in paramiko 1.6: it should absorb this exception)
 
1148
                pass
 
1149
            else:
 
1150
                raise
 
1151
        except Exception, e:
 
1152
            # This typically seems to happen during interpreter shutdown, so
 
1153
            # most of the useful ways to report this error are won't work.
 
1154
            # Writing the exception type, and then the text of the exception,
 
1155
            # seems to be the best we can do.
 
1156
            import sys
 
1157
            sys.stderr.write('\nEXCEPTION %r: ' % (e.__class__,))
 
1158
            sys.stderr.write('%s\n\n' % (e,))
 
1159
        server.finish_subsystem()
 
1160
 
 
1161
 
 
1162
class SFTPAbsoluteServer(SFTPServerWithoutSSH):
 
1163
    """A test server for sftp transports, using absolute urls."""
 
1164
 
 
1165
    def get_url(self):
 
1166
        """See bzrlib.transport.Server.get_url."""
 
1167
        homedir = self._homedir
 
1168
        if sys.platform != 'win32':
 
1169
            # Remove the initial '/' on all platforms but win32
 
1170
            homedir = homedir[1:]
 
1171
        return self._get_sftp_url(urlutils.escape(homedir))
 
1172
 
 
1173
 
 
1174
class SFTPHomeDirServer(SFTPServerWithoutSSH):
 
1175
    """A test server for sftp transports, using homedir relative urls."""
 
1176
 
 
1177
    def get_url(self):
 
1178
        """See bzrlib.transport.Server.get_url."""
 
1179
        return self._get_sftp_url("~/")
 
1180
 
 
1181
 
 
1182
class SFTPSiblingAbsoluteServer(SFTPAbsoluteServer):
 
1183
    """A test server for sftp transports where only absolute paths will work.
 
1184
 
 
1185
    It does this by serving from a deeply-nested directory that doesn't exist.
 
1186
    """
 
1187
 
 
1188
    def setUp(self, backing_server=None):
 
1189
        self._server_homedir = '/dev/noone/runs/tests/here'
 
1190
        super(SFTPSiblingAbsoluteServer, self).setUp(backing_server)
 
1191
 
907
1192
 
908
1193
def get_test_permutations():
909
1194
    """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),
 
1195
    return [(SFTPTransport, SFTPAbsoluteServer),
 
1196
            (SFTPTransport, SFTPHomeDirServer),
 
1197
            (SFTPTransport, SFTPSiblingAbsoluteServer),
914
1198
            ]