1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
1 |
# Copyright (C) 2005 Robey Pointer <robey@lag.net>, Canonical Ltd
|
2 |
||
3 |
# This program is free software; you can redistribute it and/or modify
|
|
4 |
# it under the terms of the GNU General Public License as published by
|
|
5 |
# the Free Software Foundation; either version 2 of the License, or
|
|
6 |
# (at your option) any later version.
|
|
7 |
||
8 |
# This program is distributed in the hope that it will be useful,
|
|
9 |
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
10 |
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
11 |
# GNU General Public License for more details.
|
|
12 |
||
13 |
# You should have received a copy of the GNU General Public License
|
|
14 |
# along with this program; if not, write to the Free Software
|
|
15 |
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
|
|
16 |
||
17 |
"""
|
|
18 |
A stub SFTP server for loopback SFTP testing.
|
|
19 |
Adapted from the one in paramiko's unit tests.
|
|
20 |
"""
|
|
21 |
||
22 |
import os |
|
23 |
from paramiko import ServerInterface, SFTPServerInterface, SFTPServer, SFTPAttributes, \ |
|
24 |
SFTPHandle, SFTP_OK, AUTH_SUCCESSFUL, OPEN_SUCCEEDED |
|
1666.1.6
by Robert Collins
Make knit the default format. |
25 |
|
1185.31.32
by John Arbash Meinel
Updated the bzr sourcecode to use bzrlib.osutils.pathjoin rather than os.path.join to enforce internal use of / instead of \ |
26 |
from bzrlib.osutils import pathjoin |
1185.58.2
by John Arbash Meinel
Added mode to the appropriate transport functions, and tests to make sure they work. |
27 |
from bzrlib.trace import mutter |
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
28 |
|
29 |
||
30 |
class StubServer (ServerInterface): |
|
1666.1.6
by Robert Collins
Make knit the default format. |
31 |
|
1185.49.10
by John Arbash Meinel
Use a weakref dictionary to enable re-use of a connection (for sftp). |
32 |
def __init__(self, test_case): |
33 |
ServerInterface.__init__(self) |
|
34 |
self._test_case = test_case |
|
35 |
||
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
36 |
def check_auth_password(self, username, password): |
37 |
# all are allowed
|
|
1185.49.10
by John Arbash Meinel
Use a weakref dictionary to enable re-use of a connection (for sftp). |
38 |
self._test_case.log('sftpserver - authorizing: %s' % (username,)) |
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
39 |
return AUTH_SUCCESSFUL |
40 |
||
41 |
def check_channel_request(self, kind, chanid): |
|
1185.49.10
by John Arbash Meinel
Use a weakref dictionary to enable re-use of a connection (for sftp). |
42 |
self._test_case.log('sftpserver - channel request: %s, %s' % (kind, chanid)) |
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
43 |
return OPEN_SUCCEEDED |
44 |
||
45 |
||
46 |
class StubSFTPHandle (SFTPHandle): |
|
47 |
def stat(self): |
|
48 |
try: |
|
49 |
return SFTPAttributes.from_stat(os.fstat(self.readfile.fileno())) |
|
50 |
except OSError, e: |
|
51 |
return SFTPServer.convert_errno(e.errno) |
|
52 |
||
53 |
def chattr(self, attr): |
|
54 |
# python doesn't have equivalents to fchown or fchmod, so we have to
|
|
55 |
# use the stored filename
|
|
1185.58.2
by John Arbash Meinel
Added mode to the appropriate transport functions, and tests to make sure they work. |
56 |
mutter('Changing permissions on %s to %s', self.filename, attr) |
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
57 |
try: |
58 |
SFTPServer.set_file_attr(self.filename, attr) |
|
59 |
except OSError, e: |
|
60 |
return SFTPServer.convert_errno(e.errno) |
|
61 |
||
62 |
||
63 |
class StubSFTPServer (SFTPServerInterface): |
|
1666.1.6
by Robert Collins
Make knit the default format. |
64 |
|
1524.1.1
by Robert Collins
Test sftp with relative, absolute-in-homedir and absolute-not-in-homedir |
65 |
def __init__(self, server, root, home=None): |
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
66 |
SFTPServerInterface.__init__(self, server) |
67 |
self.root = root |
|
1524.1.1
by Robert Collins
Test sftp with relative, absolute-in-homedir and absolute-not-in-homedir |
68 |
if home is None: |
69 |
self.home = self.root |
|
70 |
else: |
|
71 |
self.home = home[len(self.root):] |
|
72 |
if (len(self.home) > 0) and (self.home[0] == '/'): |
|
73 |
self.home = self.home[1:] |
|
1547.1.1
by Robey Pointer
modify the sftp unit tests to perform sftp over a direct localhost socket instead of over an actual ssh2 transport |
74 |
server._test_case.log('sftpserver - new connection') |
1524.1.1
by Robert Collins
Test sftp with relative, absolute-in-homedir and absolute-not-in-homedir |
75 |
|
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
76 |
def _realpath(self, path): |
77 |
return self.root + self.canonicalize(path) |
|
78 |
||
1524.1.1
by Robert Collins
Test sftp with relative, absolute-in-homedir and absolute-not-in-homedir |
79 |
def canonicalize(self, path): |
80 |
if os.path.isabs(path): |
|
81 |
return os.path.normpath(path) |
|
82 |
else: |
|
83 |
return os.path.normpath('/' + os.path.join(self.home, path)) |
|
84 |
||
1185.58.2
by John Arbash Meinel
Added mode to the appropriate transport functions, and tests to make sure they work. |
85 |
def chattr(self, path, attr): |
86 |
try: |
|
87 |
SFTPServer.set_file_attr(path, attr) |
|
88 |
except OSError, e: |
|
89 |
return SFTPServer.convert_errno(e.errno) |
|
90 |
return SFTP_OK |
|
91 |
||
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
92 |
def list_folder(self, path): |
93 |
path = self._realpath(path) |
|
94 |
try: |
|
95 |
out = [ ] |
|
96 |
flist = os.listdir(path) |
|
97 |
for fname in flist: |
|
1185.31.32
by John Arbash Meinel
Updated the bzr sourcecode to use bzrlib.osutils.pathjoin rather than os.path.join to enforce internal use of / instead of \ |
98 |
attr = SFTPAttributes.from_stat(os.stat(pathjoin(path, fname))) |
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
99 |
attr.filename = fname |
100 |
out.append(attr) |
|
101 |
return out |
|
102 |
except OSError, e: |
|
103 |
return SFTPServer.convert_errno(e.errno) |
|
104 |
||
105 |
def stat(self, path): |
|
106 |
path = self._realpath(path) |
|
107 |
try: |
|
108 |
return SFTPAttributes.from_stat(os.stat(path)) |
|
109 |
except OSError, e: |
|
110 |
return SFTPServer.convert_errno(e.errno) |
|
111 |
||
112 |
def lstat(self, path): |
|
113 |
path = self._realpath(path) |
|
114 |
try: |
|
115 |
return SFTPAttributes.from_stat(os.lstat(path)) |
|
116 |
except OSError, e: |
|
117 |
return SFTPServer.convert_errno(e.errno) |
|
118 |
||
119 |
def open(self, path, flags, attr): |
|
120 |
path = self._realpath(path) |
|
121 |
try: |
|
1185.31.52
by John Arbash Meinel
O_BINARY doesn't exist outside of win32, don't pass it if it isn't there. |
122 |
if hasattr(os, 'O_BINARY'): |
123 |
flags |= os.O_BINARY |
|
1540.1.9
by John Arbash Meinel
Cleanup getattr() code, since getattr(None, '', None) still works |
124 |
if getattr(attr, 'st_mode', None): |
1185.58.10
by John Arbash Meinel
[patch] Robey Pointer to fix sftp server using umask for files (failing tests for directories) |
125 |
fd = os.open(path, flags, attr.st_mode) |
126 |
else: |
|
127 |
fd = os.open(path, flags) |
|
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
128 |
except OSError, e: |
129 |
return SFTPServer.convert_errno(e.errno) |
|
130 |
if (flags & os.O_CREAT) and (attr is not None): |
|
1185.58.10
by John Arbash Meinel
[patch] Robey Pointer to fix sftp server using umask for files (failing tests for directories) |
131 |
attr._flags &= ~attr.FLAG_PERMISSIONS |
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
132 |
SFTPServer.set_file_attr(path, attr) |
133 |
if flags & os.O_WRONLY: |
|
1185.31.51
by John Arbash Meinel
Setting binary flags for sftp. |
134 |
fstr = 'wb' |
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
135 |
elif flags & os.O_RDWR: |
1185.31.51
by John Arbash Meinel
Setting binary flags for sftp. |
136 |
fstr = 'rb+' |
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
137 |
else: |
138 |
# O_RDONLY (== 0)
|
|
1185.31.51
by John Arbash Meinel
Setting binary flags for sftp. |
139 |
fstr = 'rb' |
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
140 |
try: |
141 |
f = os.fdopen(fd, fstr) |
|
142 |
except OSError, e: |
|
143 |
return SFTPServer.convert_errno(e.errno) |
|
144 |
fobj = StubSFTPHandle() |
|
145 |
fobj.filename = path |
|
146 |
fobj.readfile = f |
|
147 |
fobj.writefile = f |
|
148 |
return fobj |
|
149 |
||
150 |
def remove(self, path): |
|
151 |
path = self._realpath(path) |
|
152 |
try: |
|
153 |
os.remove(path) |
|
154 |
except OSError, e: |
|
155 |
return SFTPServer.convert_errno(e.errno) |
|
156 |
return SFTP_OK |
|
157 |
||
158 |
def rename(self, oldpath, newpath): |
|
159 |
oldpath = self._realpath(oldpath) |
|
160 |
newpath = self._realpath(newpath) |
|
161 |
try: |
|
162 |
os.rename(oldpath, newpath) |
|
163 |
except OSError, e: |
|
164 |
return SFTPServer.convert_errno(e.errno) |
|
165 |
return SFTP_OK |
|
166 |
||
167 |
def mkdir(self, path, attr): |
|
168 |
path = self._realpath(path) |
|
169 |
try: |
|
1540.1.5
by John Arbash Meinel
Bugfix to allow using paramiko > 1.5.2 |
170 |
# Using getattr() in case st_mode is None or 0
|
171 |
# both evaluate to False
|
|
1540.1.9
by John Arbash Meinel
Cleanup getattr() code, since getattr(None, '', None) still works |
172 |
if getattr(attr, 'st_mode', None): |
1185.58.11
by John Arbash Meinel
Made the StubSFTPServer use umask even for mkdir() |
173 |
os.mkdir(path, attr.st_mode) |
174 |
else: |
|
175 |
os.mkdir(path) |
|
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
176 |
if attr is not None: |
1185.58.11
by John Arbash Meinel
Made the StubSFTPServer use umask even for mkdir() |
177 |
attr._flags &= ~attr.FLAG_PERMISSIONS |
1185.16.127
by Martin Pool
[patch] paramiko sftp tests (robey) |
178 |
SFTPServer.set_file_attr(path, attr) |
179 |
except OSError, e: |
|
180 |
return SFTPServer.convert_errno(e.errno) |
|
181 |
return SFTP_OK |
|
182 |
||
183 |
def rmdir(self, path): |
|
184 |
path = self._realpath(path) |
|
185 |
try: |
|
186 |
os.rmdir(path) |
|
187 |
except OSError, e: |
|
188 |
return SFTPServer.convert_errno(e.errno) |
|
189 |
return SFTP_OK |
|
190 |
||
191 |
# removed: chattr, symlink, readlink
|
|
192 |
# (nothing in bzr's sftp transport uses those)
|