2 import os, tempfile, heapq, binascii, traceback, array, stat, struct
3 from stat import S_IFREG, S_IFDIR
4 from time import time, strftime, localtime
6 from zope.interface import implements
7 from twisted.python import components
8 from twisted.application import service, strports
9 from twisted.conch.ssh import factory, keys, session
10 from twisted.conch.ssh.filetransfer import FileTransferServer, SFTPError, \
11 FX_NO_SUCH_FILE, FX_OP_UNSUPPORTED, FX_PERMISSION_DENIED, FX_EOF, \
12 FX_BAD_MESSAGE, FX_FAILURE, FX_OK
13 from twisted.conch.ssh.filetransfer import FXF_READ, FXF_WRITE, FXF_APPEND, \
14 FXF_CREAT, FXF_TRUNC, FXF_EXCL
15 from twisted.conch.interfaces import ISFTPServer, ISFTPFile, IConchUser, ISession
16 from twisted.conch.avatar import ConchUser
17 from twisted.conch.openssh_compat import primes
18 from twisted.cred import portal
19 from twisted.internet.error import ProcessDone, ProcessTerminated
20 from twisted.python.failure import Failure
21 from twisted.internet.interfaces import ITransport
23 from twisted.internet import defer
24 from twisted.internet.interfaces import IFinishableConsumer
25 from foolscap.api import eventually
26 from allmydata.util import deferredutil
28 from allmydata.util.consumer import download_to_data
29 from allmydata.interfaces import IFileNode, IDirectoryNode, ExistingChildError, \
30 NoSuchChildError, ChildOfWrongTypeError
31 from allmydata.mutable.common import NotWriteableError
32 from allmydata.immutable.upload import FileHandle
34 from pycryptopp.cipher.aes import AES
37 use_foolscap_logging = True
39 from allmydata.util.log import NOISY, OPERATIONAL, WEIRD, \
40 msg as _msg, err as _err, PrefixingLogMixin as _PrefixingLogMixin
42 if use_foolscap_logging:
43 (logmsg, logerr, PrefixingLogMixin) = (_msg, _err, _PrefixingLogMixin)
44 else: # pragma: no cover
45 def logmsg(s, level=None):
47 def logerr(s, level=None):
49 class PrefixingLogMixin:
50 def __init__(self, facility=None, prefix=''):
52 def log(self, s, level=None):
53 print "%r %s" % (self.prefix, s)
56 def eventually_callback(d):
57 return lambda res: eventually(d.callback, res)
59 def eventually_errback(d):
60 return lambda err: eventually(d.errback, err)
64 if isinstance(x, unicode):
65 return x.encode('utf-8')
66 if isinstance(x, str):
72 """SFTP times are unsigned 32-bit integers representing UTC seconds
73 (ignoring leap seconds) since the Unix epoch, January 1 1970 00:00 UTC.
74 A Tahoe time is the corresponding float."""
75 return long(t) & 0xFFFFFFFFL
78 def _convert_error(res, request):
79 if not isinstance(res, Failure):
81 if isinstance(res, str): logged_res = "<data of length %r>" % (len(res),)
82 logmsg("SUCCESS %r %r" % (request, logged_res,), level=OPERATIONAL)
86 logmsg("RAISE %r %r" % (request, err.value), level=OPERATIONAL)
88 if noisy: logmsg(traceback.format_exc(err.value), level=NOISY)
89 except: # pragma: no cover
92 # The message argument to SFTPError must not reveal information that
93 # might compromise anonymity.
95 if err.check(SFTPError):
96 # original raiser of SFTPError has responsibility to ensure anonymity
98 if err.check(NoSuchChildError):
99 childname = _utf8(err.value.args[0])
100 raise SFTPError(FX_NO_SUCH_FILE, childname)
101 if err.check(NotWriteableError) or err.check(ChildOfWrongTypeError):
102 msg = _utf8(err.value.args[0])
103 raise SFTPError(FX_PERMISSION_DENIED, msg)
104 if err.check(ExistingChildError):
105 # Versions of SFTP after v3 (which is what twisted.conch implements)
106 # define a specific error code for this case: FX_FILE_ALREADY_EXISTS.
107 # However v3 doesn't; instead, other servers such as sshd return
108 # FX_FAILURE. The gvfs SFTP backend, for example, depends on this
109 # to translate the error to the equivalent of POSIX EEXIST, which is
110 # necessary for some picky programs (such as gedit).
111 msg = _utf8(err.value.args[0])
112 raise SFTPError(FX_FAILURE, msg)
113 if err.check(NotImplementedError):
114 raise SFTPError(FX_OP_UNSUPPORTED, _utf8(err.value))
115 if err.check(EOFError):
116 raise SFTPError(FX_EOF, "end of file reached")
117 if err.check(defer.FirstError):
118 _convert_error(err.value.subFailure, request)
120 # We assume that the error message is not anonymity-sensitive.
121 raise SFTPError(FX_FAILURE, _utf8(err.value))
124 def _repr_flags(flags):
125 return "|".join([f for f in
126 [(flags & FXF_READ) and "FXF_READ" or None,
127 (flags & FXF_WRITE) and "FXF_WRITE" or None,
128 (flags & FXF_APPEND) and "FXF_APPEND" or None,
129 (flags & FXF_CREAT) and "FXF_CREAT" or None,
130 (flags & FXF_TRUNC) and "FXF_TRUNC" or None,
131 (flags & FXF_EXCL) and "FXF_EXCL" or None,
136 def _lsLine(name, attrs):
139 st_mtime = attrs.get("mtime", 0)
140 st_mode = attrs["permissions"]
141 # TODO: check that clients are okay with this being a "?".
142 # (They should be because the longname is intended for human
144 st_size = attrs.get("size", "?")
145 # We don't know how many links there really are to this object.
148 # Based on <http://twistedmatrix.com/trac/browser/trunk/twisted/conch/ls.py?rev=25412>.
149 # We can't call the version in Twisted because we might have a version earlier than
150 # <http://twistedmatrix.com/trac/changeset/25412> (released in Twisted 8.2).
153 perms = array.array('c', '-'*10)
154 ft = stat.S_IFMT(mode)
155 if stat.S_ISDIR(ft): perms[0] = 'd'
156 elif stat.S_ISREG(ft): perms[0] = '-'
159 if mode&stat.S_IRUSR: perms[1] = 'r'
160 if mode&stat.S_IWUSR: perms[2] = 'w'
161 if mode&stat.S_IXUSR: perms[3] = 'x'
163 if mode&stat.S_IRGRP: perms[4] = 'r'
164 if mode&stat.S_IWGRP: perms[5] = 'w'
165 if mode&stat.S_IXGRP: perms[6] = 'x'
167 if mode&stat.S_IROTH: perms[7] = 'r'
168 if mode&stat.S_IWOTH: perms[8] = 'w'
169 if mode&stat.S_IXOTH: perms[9] = 'x'
170 # suid/sgid never set
173 l += str(st_nlink).rjust(5) + ' '
184 if st_mtime + sixmo < now or st_mtime > now + day:
185 # mtime is more than 6 months ago, or more than one day in the future
186 l += strftime("%b %d %Y ", localtime(st_mtime))
188 l += strftime("%b %d %H:%M ", localtime(st_mtime))
193 def _is_readonly(parent_readonly, child):
194 """Whether child should be listed as having read-only permissions in parent."""
196 if child.is_unknown():
198 elif child.is_mutable():
199 return child.is_readonly()
201 return parent_readonly
204 def _populate_attrs(childnode, metadata, size=None):
207 # The permissions must have the S_IFDIR (040000) or S_IFREG (0100000)
208 # bits, otherwise the client may refuse to open a directory.
209 # Also, sshfs run as a non-root user requires files and directories
210 # to be world-readable/writeable.
212 # Directories and unknown nodes have no size, and SFTP doesn't
213 # require us to make one up.
215 # childnode might be None, meaning that the file doesn't exist yet,
216 # but we're going to write it later.
218 if childnode and childnode.is_unknown():
220 elif childnode and IDirectoryNode.providedBy(childnode):
221 perms = S_IFDIR | 0777
223 # For files, omit the size if we don't immediately know it.
224 if childnode and size is None:
225 size = childnode.get_size()
227 assert isinstance(size, (int, long)) and not isinstance(size, bool), repr(size)
229 perms = S_IFREG | 0666
232 assert 'readonly' in metadata, metadata
233 if metadata['readonly']:
234 perms &= S_IFDIR | S_IFREG | 0555 # clear 'w' bits
236 # see webapi.txt for what these times mean
237 if 'linkmotime' in metadata.get('tahoe', {}):
238 attrs['mtime'] = _to_sftp_time(metadata['tahoe']['linkmotime'])
239 elif 'mtime' in metadata:
240 # We would prefer to omit atime, but SFTP version 3 can only
241 # accept mtime if atime is also set.
242 attrs['mtime'] = _to_sftp_time(metadata['mtime'])
243 attrs['atime'] = attrs['mtime']
245 if 'linkcrtime' in metadata.get('tahoe', {}):
246 attrs['createtime'] = _to_sftp_time(metadata['tahoe']['linkcrtime'])
248 if 'ctime' in metadata:
249 attrs['ctime'] = _to_sftp_time(metadata['ctime'])
251 attrs['permissions'] = perms
253 # twisted.conch.ssh.filetransfer only implements SFTP version 3,
254 # which doesn't include SSH_FILEXFER_ATTR_FLAGS.
259 class EncryptedTemporaryFile(PrefixingLogMixin):
260 # not implemented: next, readline, readlines, xreadlines, writelines
263 PrefixingLogMixin.__init__(self, facility="tahoe.sftp")
264 self.file = tempfile.TemporaryFile()
265 self.key = os.urandom(16) # AES-128
267 def _crypt(self, offset, data):
268 # TODO: use random-access AES (pycryptopp ticket #18)
269 offset_big = offset // 16
270 offset_small = offset % 16
271 iv = binascii.unhexlify("%032x" % offset_big)
272 cipher = AES(self.key, iv=iv)
273 cipher.process("\x00"*offset_small)
274 return cipher.process(data)
282 def seek(self, offset, whence=os.SEEK_SET):
283 if noisy: self.log(".seek(%r, %r)" % (offset, whence), level=NOISY)
284 self.file.seek(offset, whence)
287 offset = self.file.tell()
288 if noisy: self.log(".tell() = %r" % (offset,), level=NOISY)
291 def read(self, size=-1):
292 if noisy: self.log(".read(%r)" % (size,), level=NOISY)
293 index = self.file.tell()
294 ciphertext = self.file.read(size)
295 plaintext = self._crypt(index, ciphertext)
298 def write(self, plaintext):
299 if noisy: self.log(".write(<data of length %r>)" % (len(plaintext),), level=NOISY)
300 index = self.file.tell()
301 ciphertext = self._crypt(index, plaintext)
302 self.file.write(ciphertext)
304 def truncate(self, newsize):
305 if noisy: self.log(".truncate(%r)" % (newsize,), level=NOISY)
306 self.file.truncate(newsize)
309 class OverwriteableFileConsumer(PrefixingLogMixin):
310 implements(IFinishableConsumer)
311 """I act both as a consumer for the download of the original file contents, and as a
312 wrapper for a temporary file that records the downloaded data and any overwrites.
313 I use a priority queue to keep track of which regions of the file have been overwritten
314 but not yet downloaded, so that the download does not clobber overwritten data.
315 I use another priority queue to record milestones at which to make callbacks
316 indicating that a given number of bytes have been downloaded.
318 The temporary file reflects the contents of the file that I represent, except that:
319 - regions that have neither been downloaded nor overwritten, if present,
321 - the temporary file may be shorter than the represented file (it is never longer).
322 The latter's current size is stored in self.current_size.
324 This abstraction is mostly independent of SFTP. Consider moving it, if it is found
325 useful for other frontends."""
327 def __init__(self, download_size, tempfile_maker):
328 PrefixingLogMixin.__init__(self, facility="tahoe.sftp")
329 if noisy: self.log(".__init__(%r, %r)" % (download_size, tempfile_maker), level=NOISY)
330 self.download_size = download_size
331 self.current_size = download_size
332 self.f = tempfile_maker()
334 self.milestones = [] # empty heap of (offset, d)
335 self.overwrites = [] # empty heap of (start, end)
336 self.is_closed = False
337 self.done = self.when_reached(download_size) # adds a milestone
339 def _signal_done(ign):
340 if noisy: self.log("DONE", level=NOISY)
342 self.done.addCallback(_signal_done)
348 def get_current_size(self):
349 return self.current_size
351 def set_current_size(self, size):
352 if noisy: self.log(".set_current_size(%r), current_size = %r, downloaded = %r" %
353 (size, self.current_size, self.downloaded), level=NOISY)
354 if size < self.current_size or size < self.downloaded:
355 self.f.truncate(size)
356 if size > self.current_size:
357 self.overwrite(self.current_size, "\x00" * (size - self.current_size))
358 self.current_size = size
360 # invariant: self.download_size <= self.current_size
361 if size < self.download_size:
362 self.download_size = size
363 if self.downloaded >= self.download_size:
366 def registerProducer(self, p, streaming):
367 if noisy: self.log(".registerProducer(%r, streaming=%r)" % (p, streaming), level=NOISY)
370 # call resumeProducing once to start things off
379 def write(self, data):
380 if noisy: self.log(".write(<data of length %r>)" % (len(data),), level=NOISY)
384 if self.downloaded >= self.download_size:
387 next_downloaded = self.downloaded + len(data)
388 if next_downloaded > self.download_size:
389 data = data[:(self.download_size - self.downloaded)]
391 while len(self.overwrites) > 0:
392 (start, end) = self.overwrites[0]
393 if start >= next_downloaded:
394 # This and all remaining overwrites are after the data we just downloaded.
396 if start > self.downloaded:
397 # The data we just downloaded has been partially overwritten.
398 # Write the prefix of it that precedes the overwritten region.
399 self.f.seek(self.downloaded)
400 self.f.write(data[:(start - self.downloaded)])
402 # This merges consecutive overwrites if possible, which allows us to detect the
403 # case where the download can be stopped early because the remaining region
404 # to download has already been fully overwritten.
405 heapq.heappop(self.overwrites)
406 while len(self.overwrites) > 0:
407 (start1, end1) = self.overwrites[0]
411 heapq.heappop(self.overwrites)
413 if end >= next_downloaded:
414 # This overwrite extends past the downloaded data, so there is no
415 # more data to consider on this call.
416 heapq.heappush(self.overwrites, (next_downloaded, end))
417 self._update_downloaded(next_downloaded)
419 elif end >= self.downloaded:
420 data = data[(end - self.downloaded):]
421 self._update_downloaded(end)
423 self.f.seek(self.downloaded)
425 self._update_downloaded(next_downloaded)
427 def _update_downloaded(self, new_downloaded):
428 self.downloaded = new_downloaded
429 milestone = new_downloaded
430 if len(self.overwrites) > 0:
431 (start, end) = self.overwrites[0]
432 if start <= new_downloaded and end > milestone:
435 while len(self.milestones) > 0:
436 (next, d) = self.milestones[0]
439 if noisy: self.log("MILESTONE %r %r" % (next, d), level=NOISY)
440 heapq.heappop(self.milestones)
441 eventually_callback(d)(None)
443 if milestone >= self.download_size:
446 def overwrite(self, offset, data):
447 if noisy: self.log(".overwrite(%r, <data of length %r>)" % (offset, len(data)), level=NOISY)
448 if offset > self.current_size:
449 # Normally writing at an offset beyond the current end-of-file
450 # would leave a hole that appears filled with zeroes. However, an
451 # EncryptedTemporaryFile doesn't behave like that (if there is a
452 # hole in the file on disk, the zeroes that are read back will be
453 # XORed with the keystream). So we must explicitly write zeroes in
454 # the gap between the current EOF and the offset.
456 self.f.seek(self.current_size)
457 self.f.write("\x00" * (offset - self.current_size))
458 start = self.current_size
464 end = offset + len(data)
465 self.current_size = max(self.current_size, end)
466 if end > self.downloaded:
467 heapq.heappush(self.overwrites, (start, end))
469 def read(self, offset, length):
470 """When the data has been read, callback the Deferred that we return with this data.
471 Otherwise errback the Deferred that we return.
472 The caller must perform no more overwrites until the Deferred has fired."""
474 if noisy: self.log(".read(%r, %r), current_size = %r" % (offset, length, self.current_size), level=NOISY)
475 if offset >= self.current_size:
476 def _eof(): raise EOFError("read past end of file")
477 return defer.execute(_eof)
479 if offset + length > self.current_size:
480 length = self.current_size - offset
481 if noisy: self.log("truncating read to %r bytes" % (length,), level=NOISY)
483 needed = min(offset + length, self.download_size)
484 d = self.when_reached(needed)
486 # It is not necessarily the case that self.downloaded >= needed, because
487 # the file might have been truncated (thus truncating the download) and
490 assert self.current_size >= offset + length, (self.current_size, offset, length)
491 if noisy: self.log("self.f = %r" % (self.f,), level=NOISY)
493 return self.f.read(length)
494 d.addCallback(_reached)
497 def when_reached(self, index):
498 if noisy: self.log(".when_reached(%r)" % (index,), level=NOISY)
499 if index <= self.downloaded: # already reached
500 if noisy: self.log("already reached %r" % (index,), level=NOISY)
501 return defer.succeed(None)
504 if noisy: self.log("reached %r" % (index,), level=NOISY)
506 d.addCallback(_reached)
507 heapq.heappush(self.milestones, (index, d))
514 while len(self.milestones) > 0:
515 (next, d) = self.milestones[0]
516 if noisy: self.log("MILESTONE FINISH %r %r" % (next, d), level=NOISY)
517 heapq.heappop(self.milestones)
518 # The callback means that the milestone has been reached if
519 # it is ever going to be. Note that the file may have been
520 # truncated to before the milestone.
521 eventually_callback(d)(None)
523 # FIXME: causes spurious failures
524 #self.unregisterProducer()
527 self.is_closed = True
529 if not self.is_closed:
532 except BaseException as e:
533 self.log("suppressed %r from close of temporary file %r" % (e, self.f), level=WEIRD)
535 def unregisterProducer(self):
537 self.producer.stopProducing()
541 SIZE_THRESHOLD = 1000
544 class ShortReadOnlySFTPFile(PrefixingLogMixin):
545 implements(ISFTPFile)
546 """I represent a file handle to a particular file on an SFTP connection.
547 I am used only for short immutable files opened in read-only mode.
548 The file contents are downloaded to memory when I am created."""
550 def __init__(self, userpath, filenode, metadata):
551 PrefixingLogMixin.__init__(self, facility="tahoe.sftp", prefix=userpath)
552 if noisy: self.log(".__init__(%r, %r, %r)" % (userpath, filenode, metadata), level=NOISY)
554 assert IFileNode.providedBy(filenode), filenode
555 self.filenode = filenode
556 self.metadata = metadata
557 self.async = download_to_data(filenode)
560 def readChunk(self, offset, length):
561 request = ".readChunk(%r, %r)" % (offset, length)
562 self.log(request, level=OPERATIONAL)
565 def _closed(): raise SFTPError(FX_BAD_MESSAGE, "cannot read from a closed file handle")
566 return defer.execute(_closed)
570 if noisy: self.log("_read(<data of length %r>) in readChunk(%r, %r)" % (len(data), offset, length), level=NOISY)
572 # "In response to this request, the server will read as many bytes as it
573 # can from the file (up to 'len'), and return them in a SSH_FXP_DATA
574 # message. If an error occurs or EOF is encountered before reading any
575 # data, the server will respond with SSH_FXP_STATUS. For normal disk
576 # files, it is guaranteed that this will read the specified number of
577 # bytes, or up to end of file."
579 # i.e. we respond with an EOF error iff offset is already at EOF.
581 if offset >= len(data):
582 eventually_errback(d)(SFTPError(FX_EOF, "read at or past end of file"))
584 eventually_callback(d)(data[offset:min(offset+length, len(data))])
586 self.async.addCallbacks(_read, eventually_errback(d))
587 d.addBoth(_convert_error, request)
590 def writeChunk(self, offset, data):
591 self.log(".writeChunk(%r, <data of length %r>) denied" % (offset, len(data)), level=OPERATIONAL)
593 def _denied(): raise SFTPError(FX_PERMISSION_DENIED, "file handle was not opened for writing")
594 return defer.execute(_denied)
597 self.log(".close()", level=OPERATIONAL)
600 return defer.succeed(None)
603 request = ".getAttrs()"
604 self.log(request, level=OPERATIONAL)
607 def _closed(): raise SFTPError(FX_BAD_MESSAGE, "cannot get attributes for a closed file handle")
608 return defer.execute(_closed)
610 d = defer.execute(_populate_attrs, self.filenode, self.metadata)
611 d.addBoth(_convert_error, request)
614 def setAttrs(self, attrs):
615 self.log(".setAttrs(%r) denied" % (attrs,), level=OPERATIONAL)
616 def _denied(): raise SFTPError(FX_PERMISSION_DENIED, "file handle was not opened for writing")
617 return defer.execute(_denied)
620 class GeneralSFTPFile(PrefixingLogMixin):
621 implements(ISFTPFile)
622 """I represent a file handle to a particular file on an SFTP connection.
623 I wrap an instance of OverwriteableFileConsumer, which is responsible for
624 storing the file contents. In order to allow write requests to be satisfied
625 immediately, there is effectively a FIFO queue between requests made to this
626 file handle, and requests to my OverwriteableFileConsumer. This queue is
627 implemented by the callback chain of self.async.
629 When first constructed, I am in an 'unopened' state that causes most
630 operations to be delayed until 'open' is called."""
632 def __init__(self, userpath, flags, close_notify, convergence):
633 PrefixingLogMixin.__init__(self, facility="tahoe.sftp", prefix=userpath)
634 if noisy: self.log(".__init__(%r, %r = %r, %r, <convergence censored>)" %
635 (userpath, flags, _repr_flags(flags), close_notify), level=NOISY)
637 self.userpath = userpath
639 self.close_notify = close_notify
640 self.convergence = convergence
641 self.async = defer.Deferred()
642 # Creating or truncating the file is a change, but if FXF_EXCL is set, a zero-length file has already been created.
643 self.has_changed = (flags & (FXF_CREAT | FXF_TRUNC)) and not (flags & FXF_EXCL)
645 self.abandoned = False
647 self.childname = None
651 # self.consumer should only be relied on in callbacks for self.async, since it might
652 # not be set before then.
655 def open(self, parent=None, childname=None, filenode=None, metadata=None):
656 self.log(".open(parent=%r, childname=%r, filenode=%r, metadata=%r)" %
657 (parent, childname, filenode, metadata), level=OPERATIONAL)
659 # If the file has been renamed, the new (parent, childname) takes precedence.
660 if self.parent is None:
662 if self.childname is None:
663 self.childname = childname
664 self.filenode = filenode
665 self.metadata = metadata
668 tempfile_maker = EncryptedTemporaryFile
670 if (self.flags & FXF_TRUNC) or not filenode:
671 # We're either truncating or creating the file, so we don't need the old contents.
672 self.consumer = OverwriteableFileConsumer(0, tempfile_maker)
673 self.consumer.finish()
675 assert IFileNode.providedBy(filenode), filenode
677 # TODO: use download interface described in #993 when implemented.
678 if filenode.is_mutable():
679 self.async.addCallback(lambda ign: filenode.download_best_version())
680 def _downloaded(data):
681 self.consumer = OverwriteableFileConsumer(len(data), tempfile_maker)
682 self.consumer.write(data)
683 self.consumer.finish()
685 self.async.addCallback(_downloaded)
687 download_size = filenode.get_size()
688 assert download_size is not None, "download_size is None"
689 self.consumer = OverwriteableFileConsumer(download_size, tempfile_maker)
691 if noisy: self.log("_read immutable", level=NOISY)
692 filenode.read(self.consumer, 0, None)
693 self.async.addCallback(_read)
695 eventually_callback(self.async)(None)
697 if noisy: self.log("open done", level=NOISY)
700 def rename(self, new_userpath, new_parent, new_childname):
701 self.log(".rename(%r, %r, %r)" % (new_userpath, new_parent, new_childname), level=OPERATIONAL)
703 self.userpath = new_userpath
704 self.parent = new_parent
705 self.childname = new_childname
708 self.log(".abandon()", level=OPERATIONAL)
710 self.abandoned = True
713 self.log(".sync()", level=OPERATIONAL)
716 self.async.addBoth(eventually_callback(d))
719 def readChunk(self, offset, length):
720 request = ".readChunk(%r, %r)" % (offset, length)
721 self.log(request, level=OPERATIONAL)
723 if not (self.flags & FXF_READ):
724 def _denied(): raise SFTPError(FX_PERMISSION_DENIED, "file handle was not opened for reading")
725 return defer.execute(_denied)
728 def _closed(): raise SFTPError(FX_BAD_MESSAGE, "cannot read from a closed file handle")
729 return defer.execute(_closed)
733 if noisy: self.log("_read in readChunk(%r, %r)" % (offset, length), level=NOISY)
734 d2 = self.consumer.read(offset, length)
735 d2.addCallbacks(eventually_callback(d), eventually_errback(d))
736 # It is correct to drop d2 here.
738 self.async.addCallbacks(_read, eventually_errback(d))
739 d.addBoth(_convert_error, request)
742 def writeChunk(self, offset, data):
743 self.log(".writeChunk(%r, <data of length %r>)" % (offset, len(data)), level=OPERATIONAL)
745 if not (self.flags & FXF_WRITE):
746 def _denied(): raise SFTPError(FX_PERMISSION_DENIED, "file handle was not opened for writing")
747 return defer.execute(_denied)
750 def _closed(): raise SFTPError(FX_BAD_MESSAGE, "cannot write to a closed file handle")
751 return defer.execute(_closed)
753 self.has_changed = True
755 # Note that we return without waiting for the write to occur. Reads and
756 # close wait for prior writes, and will fail if any prior operation failed.
757 # This is ok because SFTP makes no guarantee that the write completes
758 # before the request does. In fact it explicitly allows write errors to be
759 # delayed until close:
760 # "One should note that on some server platforms even a close can fail.
761 # This can happen e.g. if the server operating system caches writes,
762 # and an error occurs while flushing cached writes during the close."
765 if noisy: self.log("_write in .writeChunk(%r, <data of length %r>), current_size = %r" %
766 (offset, len(data), self.consumer.get_current_size()), level=NOISY)
767 # FXF_APPEND means that we should always write at the current end of file.
768 write_offset = offset
769 if self.flags & FXF_APPEND:
770 write_offset = self.consumer.get_current_size()
772 self.consumer.overwrite(write_offset, data)
773 if noisy: self.log("overwrite done", level=NOISY)
775 self.async.addCallback(_write)
776 # don't addErrback to self.async, just allow subsequent async ops to fail.
777 return defer.succeed(None)
781 self.log(request, level=OPERATIONAL)
784 return defer.succeed(None)
786 # This means that close has been called, not that the close has succeeded.
789 if not (self.flags & (FXF_WRITE | FXF_CREAT)):
790 def _readonly_close():
792 self.consumer.close()
793 return defer.execute(_readonly_close)
795 # We must capture the abandoned, parent, and childname variables synchronously
796 # at the close call. This is needed by the correctness arguments in the comments
797 # for _abandon_any_heisenfiles and _rename_heisenfiles.
798 abandoned = self.abandoned
800 childname = self.childname
802 # has_changed is set when writeChunk is called, not when the write occurs, so
803 # it is correct to optimize out the commit if it is False at the close call.
804 has_changed = self.has_changed
807 if noisy: self.log("_committed(%r)" % (res,), level=NOISY)
809 self.consumer.close()
811 # We must close_notify before re-firing self.async.
812 if self.close_notify:
813 self.close_notify(self.userpath, self.parent, self.childname, self)
817 d2 = self.consumer.when_done()
818 if self.filenode and self.filenode.is_mutable():
819 self.log("update mutable file %r childname=%r" % (self.filenode, self.childname,), level=OPERATIONAL)
820 d2.addCallback(lambda ign: self.consumer.get_current_size())
821 d2.addCallback(lambda size: self.consumer.read(0, size))
822 d2.addCallback(lambda new_contents: self.filenode.overwrite(new_contents))
825 self.log("_add_file childname=%r" % (childname,), level=OPERATIONAL)
826 u = FileHandle(self.consumer.get_file(), self.convergence)
827 return parent.add_file(childname, u)
828 d2.addCallback(_add_file)
830 d2.addBoth(_committed)
835 # If the file has been abandoned, we don't want the close operation to get "stuck",
836 # even if self.async fails to re-fire. Doing the close independently of self.async
837 # in that case ensures that dropping an ssh connection is sufficient to abandon
838 # any heisenfiles that were not explicitly closed in that connection.
839 if abandoned or not has_changed:
840 d.addCallback(_committed)
842 self.async.addCallback(_close)
844 self.async.addCallbacks(eventually_callback(d), eventually_errback(d))
845 d.addBoth(_convert_error, request)
849 request = ".getAttrs()"
850 self.log(request, level=OPERATIONAL)
853 def _closed(): raise SFTPError(FX_BAD_MESSAGE, "cannot get attributes for a closed file handle")
854 return defer.execute(_closed)
856 # Optimization for read-only handles, when we already know the metadata.
857 if not(self.flags & (FXF_WRITE | FXF_CREAT)) and self.metadata and self.filenode and not self.filenode.is_mutable():
858 return defer.succeed(_populate_attrs(self.filenode, self.metadata))
862 # self.filenode might be None, but that's ok.
863 attrs = _populate_attrs(self.filenode, self.metadata, size=self.consumer.get_current_size())
864 eventually_callback(d)(attrs)
866 self.async.addCallbacks(_get, eventually_errback(d))
867 d.addBoth(_convert_error, request)
870 def setAttrs(self, attrs):
871 request = ".setAttrs(attrs) %r" % (attrs,)
872 self.log(request, level=OPERATIONAL)
874 if not (self.flags & FXF_WRITE):
875 def _denied(): raise SFTPError(FX_PERMISSION_DENIED, "file handle was not opened for writing")
876 return defer.execute(_denied)
879 def _closed(): raise SFTPError(FX_BAD_MESSAGE, "cannot set attributes for a closed file handle")
880 return defer.execute(_closed)
882 if not "size" in attrs:
883 return defer.succeed(None)
886 if not isinstance(size, (int, long)) or size < 0:
887 def _bad(): raise SFTPError(FX_BAD_MESSAGE, "new size is not a valid nonnegative integer")
888 return defer.execute(_bad)
892 self.consumer.set_current_size(size)
893 eventually_callback(d)(None)
895 self.async.addCallbacks(_resize, eventually_errback(d))
896 d.addBoth(_convert_error, request)
901 def __init__(self, items):
911 def __init__(self, value):
915 # A "heisenfile" is a file that has been opened with write flags
916 # (FXF_WRITE and/or FXF_CREAT) and not yet close-notified.
917 # 'all_heisenfiles' maps from a direntry string to
918 # (list_of_GeneralSFTPFile, open_time_utc).
919 # A direntry string is parent_write_uri + "/" + childname_utf8 for
920 # an immutable file, or file_write_uri for a mutable file.
921 # Updates to this dict are single-threaded.
926 class SFTPUserHandler(ConchUser, PrefixingLogMixin):
927 implements(ISFTPServer)
928 def __init__(self, client, rootnode, username):
929 ConchUser.__init__(self)
930 PrefixingLogMixin.__init__(self, facility="tahoe.sftp", prefix=username)
931 if noisy: self.log(".__init__(%r, %r, %r)" % (client, rootnode, username), level=NOISY)
933 self.channelLookup["session"] = session.SSHSession
934 self.subsystemLookup["sftp"] = FileTransferServer
936 self._client = client
937 self._root = rootnode
938 self._username = username
939 self._convergence = client.convergence
941 # maps from UTF-8 paths for this user, to files written and still open
942 self._heisenfiles = {}
944 def gotVersion(self, otherVersion, extData):
945 self.log(".gotVersion(%r, %r)" % (otherVersion, extData), level=OPERATIONAL)
947 # advertise the same extensions as the OpenSSH SFTP server
948 # <http://www.openbsd.org/cgi-bin/cvsweb/src/usr.bin/ssh/PROTOCOL?rev=1.15>
949 return {'posix-rename@openssh.com': '1',
950 'statvfs@openssh.com': '2',
951 'fstatvfs@openssh.com': '2',
955 self.log(".logout()", level=OPERATIONAL)
957 for files in self._heisenfiles.itervalues():
961 def _add_heisenfiles_by_path(self, userpath, files):
962 if noisy: self.log("._add_heisenfiles_by_path(%r, %r)" % (userpath, files), level=NOISY)
964 if userpath in self._heisenfiles:
965 self._heisenfiles[userpath] += files
967 self._heisenfiles[userpath] = files
969 def _add_heisenfiles_by_direntry(self, direntry, files_to_add):
970 if noisy: self.log("._add_heisenfiles_by_direntry(%r, %r)" % (direntry, files_to_add), level=NOISY)
973 if direntry in all_heisenfiles:
974 (old_files, opentime) = all_heisenfiles[direntry]
975 all_heisenfiles[direntry] = (old_files + files_to_add, opentime)
977 all_heisenfiles[direntry] = (files_to_add, time())
979 def _abandon_any_heisenfiles(self, userpath, direntry):
980 if noisy: self.log("._abandon_any_heisenfiles(%r, %r)" % (userpath, direntry), level=NOISY)
982 # First we synchronously mark all heisenfiles matching the userpath or direntry
983 # as abandoned, and remove them from the two heisenfile dicts. Then we .sync()
984 # each file that we abandoned.
986 # For each file, the call to .abandon() occurs:
987 # * before the file is closed, in which case it will never be committed
988 # (uploaded+linked or published); or
989 # * after it is closed but before it has been close_notified, in which case the
990 # .sync() ensures that it has been committed (successfully or not) before we
993 # This avoids a race that might otherwise cause the file to be committed after
994 # the remove operation has completed.
996 # We return a Deferred that fires with True if any files were abandoned (this
997 # does not mean that they were not committed; it is used to determine whether
998 # a NoSuchChildError from the attempt to delete the file should be suppressed).
1001 if direntry in all_heisenfiles:
1002 (files, opentime) = all_heisenfiles[direntry]
1003 del all_heisenfiles[direntry]
1004 if userpath in self._heisenfiles:
1005 files += self._heisenfiles[userpath]
1006 del self._heisenfiles[userpath]
1011 d = defer.succeed(None)
1013 d.addBoth(lambda ign: f.sync())
1015 d.addBoth(lambda ign: len(files) > 0)
1018 def _rename_heisenfiles(self, from_userpath, from_parent, from_childname,
1019 to_userpath, to_parent, to_childname, overwrite=True):
1020 if noisy: self.log("._rename_heisenfiles(%r, %r, %r, %r, %r, %r, overwrite=%r)" %
1021 (from_userpath, from_parent, from_childname,
1022 to_userpath, to_parent, to_childname, overwrite), level=NOISY)
1024 # First we synchronously rename all heisenfiles matching the userpath or direntry.
1025 # Then we .sync() each file that we renamed.
1027 # For each file, the call to .rename occurs:
1028 # * before the file is closed, in which case it will be committed at the
1030 # * after it is closed but before it has been close_notified, in which case the
1031 # .sync() ensures that it has been committed (successfully or not) before we
1034 # This avoids a race that might otherwise cause the file to be committed at the
1035 # old name after the rename operation has completed.
1037 # Note that if overwrite is False, the caller should already have checked
1038 # whether a real direntry exists at the destination. It is possible that another
1039 # direntry (heisen or real) comes to exist at the destination after that check,
1040 # but in that case it is correct for the rename to succeed (and for the commit
1041 # of the heisenfile at the destination to possibly clobber the other entry, since
1042 # that can happen anyway when we have concurrent write handles to the same direntry).
1044 # We return a Deferred that fires with True if any files were renamed (this
1045 # does not mean that they were not committed; it is used to determine whether
1046 # a NoSuchChildError from the rename attempt should be suppressed). If overwrite
1047 # is False and there were already heisenfiles at the destination userpath or
1048 # direntry, we return a Deferred that fails with SFTPError(FX_PERMISSION_DENIED).
1050 from_direntry = self._direntry_for(from_parent, from_childname)
1051 to_direntry = self._direntry_for(to_parent, to_childname)
1053 if not overwrite and (to_userpath in self._heisenfiles or to_direntry in all_heisenfiles):
1054 def _existing(): raise SFTPError(FX_PERMISSION_DENIED, "cannot rename to existing path " + to_userpath)
1055 return defer.execute(_existing)
1058 if from_direntry in all_heisenfiles:
1059 (from_files, opentime) = all_heisenfiles[from_direntry]
1060 del all_heisenfiles[from_direntry]
1061 if from_userpath in self._heisenfiles:
1062 from_files += self._heisenfiles[from_userpath]
1063 del self._heisenfiles[from_userpath]
1065 self._add_heisenfiles_by_direntry(to_direntry, from_files)
1066 self._add_heisenfiles_by_path(to_userpath, from_files)
1068 for f in from_files:
1069 f.rename(to_userpath, to_parent, to_childname)
1071 d = defer.succeed(None)
1072 for f in from_files:
1073 d.addBoth(lambda ign: f.sync())
1075 d.addBoth(lambda ign: len(from_files) > 0)
1078 def _sync_heisenfiles(self, userpath, direntry, ignore=None):
1079 request = "._sync_heisenfiles(%r, %r, ignore=%r)" % (userpath, direntry, ignore)
1080 self.log(request, level=OPERATIONAL)
1083 if direntry in all_heisenfiles:
1084 (files, opentime) = all_heisenfiles[direntry]
1085 if userpath in self._heisenfiles:
1086 files += self._heisenfiles[userpath]
1088 if noisy: self.log("files = %r in %r" % (files, request), level=NOISY)
1090 d = defer.succeed(None)
1094 if noisy: self.log("_sync %r in %r" % (f, request), level=NOISY)
1099 self.log("done %r" % (request,), level=OPERATIONAL)
1104 def _remove_heisenfile(self, userpath, parent, childname, file_to_remove):
1105 if noisy: self.log("._remove_heisenfile(%r, %r, %r, %r)" % (userpath, parent, childname, file_to_remove), level=NOISY)
1107 direntry = self._direntry_for(parent, childname)
1108 if direntry in all_heisenfiles:
1109 (all_old_files, opentime) = all_heisenfiles[direntry]
1110 all_new_files = [f for f in all_old_files if f is not file_to_remove]
1111 if len(all_new_files) > 0:
1112 all_heisenfiles[direntry] = (all_new_files, opentime)
1114 del all_heisenfiles[direntry]
1116 if userpath in self._heisenfiles:
1117 old_files = self._heisenfiles[userpath]
1118 new_files = [f for f in old_files if f is not file_to_remove]
1119 if len(new_files) > 0:
1120 self._heisenfiles[userpath] = new_files
1122 del self._heisenfiles[userpath]
1124 def _direntry_for(self, filenode_or_parent, childname=None):
1125 if filenode_or_parent:
1126 rw_uri = filenode_or_parent.get_write_uri()
1127 if rw_uri and childname:
1128 return rw_uri + "/" + childname.encode('utf-8')
1134 def _make_file(self, existing_file, userpath, flags, parent=None, childname=None, filenode=None, metadata=None):
1135 if noisy: self.log("._make_file(%r, %r, %r = %r, parent=%r, childname=%r, filenode=%r, metadata=%r)" %
1136 (existing_file, userpath, flags, _repr_flags(flags), parent, childname, filenode, metadata),
1139 assert metadata is None or 'readonly' in metadata, metadata
1141 writing = (flags & (FXF_WRITE | FXF_CREAT)) != 0
1143 direntry = self._direntry_for(parent, childname)
1145 direntry = self._direntry_for(filenode)
1147 d = self._sync_heisenfiles(userpath, direntry, ignore=existing_file)
1149 if not writing and (flags & FXF_READ) and filenode and not filenode.is_mutable() and filenode.get_size() <= SIZE_THRESHOLD:
1150 d.addCallback(lambda ign: ShortReadOnlySFTPFile(userpath, filenode, metadata))
1154 close_notify = self._remove_heisenfile
1156 d.addCallback(lambda ign: existing_file or GeneralSFTPFile(userpath, flags, close_notify, self._convergence))
1157 def _got_file(file):
1159 self._add_heisenfiles_by_direntry(direntry, [file])
1160 return file.open(parent=parent, childname=childname, filenode=filenode, metadata=metadata)
1161 d.addCallback(_got_file)
1164 def openFile(self, pathstring, flags, attrs):
1165 request = ".openFile(%r, %r = %r, %r)" % (pathstring, flags, _repr_flags(flags), attrs)
1166 self.log(request, level=OPERATIONAL)
1168 # This is used for both reading and writing.
1169 # First exclude invalid combinations of flags, and empty paths.
1171 if not (flags & (FXF_READ | FXF_WRITE)):
1172 def _bad_readwrite():
1173 raise SFTPError(FX_BAD_MESSAGE, "invalid file open flags: at least one of FXF_READ and FXF_WRITE must be set")
1174 return defer.execute(_bad_readwrite)
1176 if (flags & FXF_EXCL) and not (flags & FXF_CREAT):
1177 def _bad_exclcreat():
1178 raise SFTPError(FX_BAD_MESSAGE, "invalid file open flags: FXF_EXCL cannot be set without FXF_CREAT")
1179 return defer.execute(_bad_exclcreat)
1181 path = self._path_from_string(pathstring)
1183 def _emptypath(): raise SFTPError(FX_NO_SUCH_FILE, "path cannot be empty")
1184 return defer.execute(_emptypath)
1186 # The combination of flags is potentially valid.
1188 # To work around clients that have race condition bugs, a getAttr, rename, or
1189 # remove request following an 'open' request with FXF_WRITE or FXF_CREAT flags,
1190 # should succeed even if the 'open' request has not yet completed. So we now
1191 # synchronously add a file object into the self._heisenfiles dict, indexed
1192 # by its UTF-8 userpath. (We can't yet add it to the all_heisenfiles dict,
1193 # because we don't yet have a user-independent path for the file.) The file
1194 # object does not know its filenode, parent, or childname at this point.
1196 userpath = self._path_to_utf8(path)
1198 if flags & (FXF_WRITE | FXF_CREAT):
1199 file = GeneralSFTPFile(userpath, flags, self._remove_heisenfile, self._convergence)
1200 self._add_heisenfiles_by_path(userpath, [file])
1202 # We haven't decided which file implementation to use yet.
1205 # Now there are two major cases:
1207 # 1. The path is specified as /uri/FILECAP, with no parent directory.
1208 # If the FILECAP is mutable and writeable, then we can open it in write-only
1209 # or read/write mode (non-exclusively), otherwise we can only open it in
1210 # read-only mode. The open should succeed immediately as long as FILECAP is
1211 # a valid known filecap that grants the required permission.
1213 # 2. The path is specified relative to a parent. We find the parent dirnode and
1214 # get the child's URI and metadata if it exists. There are four subcases:
1215 # a. the child does not exist: FXF_CREAT must be set, and we must be able
1216 # to write to the parent directory.
1217 # b. the child exists but is not a valid known filecap: fail
1218 # c. the child is mutable: if we are trying to open it write-only or
1219 # read/write, then we must be able to write to the file.
1220 # d. the child is immutable: if we are trying to open it write-only or
1221 # read/write, then we must be able to write to the parent directory.
1223 # To reduce latency, open normally succeeds as soon as these conditions are
1224 # met, even though there might be a failure in downloading the existing file
1225 # or uploading a new one. However, there is an exception: if a file has been
1226 # written, then closed, and is now being reopened, then we have to delay the
1227 # open until the previous upload/publish has completed. This is necessary
1228 # because sshfs does not wait for the result of an FXF_CLOSE message before
1229 # reporting to the client that a file has been closed. It applies both to
1230 # mutable files, and to directory entries linked to an immutable file.
1232 # Note that the permission checks below are for more precise error reporting on
1233 # the open call; later operations would fail even if we did not make these checks.
1235 d = self._get_root(path)
1236 def _got_root( (root, path) ):
1237 if root.is_unknown():
1238 raise SFTPError(FX_PERMISSION_DENIED,
1239 "cannot open an unknown cap (or child of an unknown directory). "
1240 "Upgrading the gateway to a later Tahoe-LAFS version may help")
1243 if noisy: self.log("case 1: root = %r, path[:-1] = %r" % (root, path[:-1]), level=NOISY)
1244 if not IFileNode.providedBy(root):
1245 raise SFTPError(FX_PERMISSION_DENIED,
1246 "cannot open a directory cap")
1247 if (flags & FXF_WRITE) and root.is_readonly():
1248 raise SFTPError(FX_PERMISSION_DENIED,
1249 "cannot write to a non-writeable filecap without a parent directory")
1250 if flags & FXF_EXCL:
1251 raise SFTPError(FX_FAILURE,
1252 "cannot create a file exclusively when it already exists")
1254 # The file does not need to be added to all_heisenfiles, because it is not
1255 # associated with a directory entry that needs to be updated.
1257 return self._make_file(file, userpath, flags, filenode=root)
1260 childname = path[-1]
1261 if noisy: self.log("case 2: root = %r, childname = %r, path[:-1] = %r" %
1262 (root, childname, path[:-1]), level=NOISY)
1263 d2 = root.get_child_at_path(path[:-1])
1264 def _got_parent(parent):
1265 if noisy: self.log("_got_parent(%r)" % (parent,), level=NOISY)
1266 if parent.is_unknown():
1267 raise SFTPError(FX_PERMISSION_DENIED,
1268 "cannot open an unknown cap (or child of an unknown directory). "
1269 "Upgrading the gateway to a later Tahoe-LAFS version may help")
1271 parent_readonly = parent.is_readonly()
1272 d3 = defer.succeed(None)
1273 if flags & FXF_EXCL:
1274 # FXF_EXCL means that the link to the file (not the file itself) must
1275 # be created atomically wrt updates by this storage client.
1276 # That is, we need to create the link before returning success to the
1277 # SFTP open request (and not just on close, as would normally be the
1278 # case). We make the link initially point to a zero-length LIT file,
1279 # which is consistent with what might happen on a POSIX filesystem.
1282 raise SFTPError(FX_FAILURE,
1283 "cannot create a file exclusively when the parent directory is read-only")
1285 # 'overwrite=False' ensures failure if the link already exists.
1286 # FIXME: should use a single call to set_uri and return (child, metadata) (#1035)
1288 zero_length_lit = "URI:LIT:"
1289 if noisy: self.log("%r.set_uri(%r, None, readcap=%r, overwrite=False)" %
1290 (parent, zero_length_lit, childname), level=NOISY)
1291 d3.addCallback(lambda ign: parent.set_uri(childname, None, readcap=zero_length_lit, overwrite=False))
1292 def _seturi_done(child):
1293 if noisy: self.log("%r.get_metadata_for(%r)" % (parent, childname), level=NOISY)
1294 d4 = parent.get_metadata_for(childname)
1295 d4.addCallback(lambda metadata: (child, metadata))
1297 d3.addCallback(_seturi_done)
1299 if noisy: self.log("%r.get_child_and_metadata(%r)" % (parent, childname), level=NOISY)
1300 d3.addCallback(lambda ign: parent.get_child_and_metadata(childname))
1302 def _got_child( (filenode, metadata) ):
1303 if noisy: self.log("_got_child( (%r, %r) )" % (filenode, metadata), level=NOISY)
1305 if filenode.is_unknown():
1306 raise SFTPError(FX_PERMISSION_DENIED,
1307 "cannot open an unknown cap. Upgrading the gateway "
1308 "to a later Tahoe-LAFS version may help")
1309 if not IFileNode.providedBy(filenode):
1310 raise SFTPError(FX_PERMISSION_DENIED,
1311 "cannot open a directory as if it were a file")
1312 if (flags & FXF_WRITE) and filenode.is_mutable() and filenode.is_readonly():
1313 raise SFTPError(FX_PERMISSION_DENIED,
1314 "cannot open a read-only mutable file for writing")
1315 if (flags & FXF_WRITE) and parent_readonly:
1316 raise SFTPError(FX_PERMISSION_DENIED,
1317 "cannot open a file for writing when the parent directory is read-only")
1319 metadata['readonly'] = _is_readonly(parent_readonly, filenode)
1320 return self._make_file(file, userpath, flags, parent=parent, childname=childname,
1321 filenode=filenode, metadata=metadata)
1323 if noisy: self.log("_no_child(%r)" % (f,), level=NOISY)
1324 f.trap(NoSuchChildError)
1326 if not (flags & FXF_CREAT):
1327 raise SFTPError(FX_NO_SUCH_FILE,
1328 "the file does not exist, and was not opened with the creation (CREAT) flag")
1330 raise SFTPError(FX_PERMISSION_DENIED,
1331 "cannot create a file when the parent directory is read-only")
1333 return self._make_file(file, userpath, flags, parent=parent, childname=childname)
1334 d3.addCallbacks(_got_child, _no_child)
1337 d2.addCallback(_got_parent)
1340 d.addCallback(_got_root)
1341 def _remove_on_error(err):
1343 self._remove_heisenfile(userpath, None, None, file)
1345 d.addErrback(_remove_on_error)
1346 d.addBoth(_convert_error, request)
1349 def renameFile(self, from_pathstring, to_pathstring, overwrite=False):
1350 request = ".renameFile(%r, %r)" % (from_pathstring, to_pathstring)
1351 self.log(request, level=OPERATIONAL)
1353 from_path = self._path_from_string(from_pathstring)
1354 to_path = self._path_from_string(to_pathstring)
1355 from_userpath = self._path_to_utf8(from_path)
1356 to_userpath = self._path_to_utf8(to_path)
1358 # the target directory must already exist
1359 d = deferredutil.gatherResults([self._get_parent_or_node(from_path),
1360 self._get_parent_or_node(to_path)])
1361 def _got( (from_pair, to_pair) ):
1362 if noisy: self.log("_got( (%r, %r) ) in .renameFile(%r, %r, overwrite=%r)" %
1363 (from_pair, to_pair, from_pathstring, to_pathstring, overwrite), level=NOISY)
1364 (from_parent, from_childname) = from_pair
1365 (to_parent, to_childname) = to_pair
1367 if from_childname is None:
1368 raise SFTPError(FX_NO_SUCH_FILE, "cannot rename a source object specified by URI")
1369 if to_childname is None:
1370 raise SFTPError(FX_NO_SUCH_FILE, "cannot rename to a destination specified by URI")
1372 # <http://tools.ietf.org/html/draft-ietf-secsh-filexfer-02#section-6.5>
1373 # "It is an error if there already exists a file with the name specified
1375 # OpenSSH's SFTP server returns FX_PERMISSION_DENIED for this error.
1377 # For the standard SSH_FXP_RENAME operation, overwrite=False.
1378 # We also support the posix-rename@openssh.com extension, which uses overwrite=True.
1380 d2 = defer.fail(NoSuchChildError())
1382 d2.addCallback(lambda ign: to_parent.get(to_childname))
1383 def _expect_fail(res):
1384 if not isinstance(res, Failure):
1385 raise SFTPError(FX_PERMISSION_DENIED, "cannot rename to existing path " + to_userpath)
1387 # It is OK if we fail for errors other than NoSuchChildError, since that probably
1388 # indicates some problem accessing the destination directory.
1389 res.trap(NoSuchChildError)
1390 d2.addBoth(_expect_fail)
1392 # If there are heisenfiles to be written at the 'from' direntry, then ensure
1393 # they will now be written at the 'to' direntry instead.
1394 d2.addCallback(lambda ign:
1395 self._rename_heisenfiles(from_userpath, from_parent, from_childname,
1396 to_userpath, to_parent, to_childname, overwrite=overwrite))
1399 # FIXME: use move_child_to_path to avoid possible data loss due to #943
1400 #d3 = from_parent.move_child_to_path(from_childname, to_root, to_path, overwrite=overwrite)
1402 d3 = from_parent.move_child_to(from_childname, to_parent, to_childname, overwrite=overwrite)
1404 if noisy: self.log("_check(%r) in .renameFile(%r, %r, overwrite=%r)" %
1405 (err, from_pathstring, to_pathstring, overwrite), level=NOISY)
1407 if not isinstance(err, Failure) or (renamed and err.check(NoSuchChildError)):
1409 if not overwrite and err.check(ExistingChildError):
1410 raise SFTPError(FX_PERMISSION_DENIED, "cannot rename to existing path " + to_userpath)
1415 d2.addCallback(_move)
1418 d.addBoth(_convert_error, request)
1421 def makeDirectory(self, pathstring, attrs):
1422 request = ".makeDirectory(%r, %r)" % (pathstring, attrs)
1423 self.log(request, level=OPERATIONAL)
1425 path = self._path_from_string(pathstring)
1426 metadata = self._attrs_to_metadata(attrs)
1427 d = self._get_root(path)
1428 d.addCallback(lambda (root, path):
1429 self._get_or_create_directories(root, path, metadata))
1430 d.addBoth(_convert_error, request)
1433 def _get_or_create_directories(self, node, path, metadata):
1434 if not IDirectoryNode.providedBy(node):
1435 # TODO: provide the name of the blocking file in the error message.
1436 def _blocked(): raise SFTPError(FX_FAILURE, "cannot create directory because there "
1437 "is a file in the way") # close enough
1438 return defer.execute(_blocked)
1441 return defer.succeed(node)
1442 d = node.get(path[0])
1443 def _maybe_create(f):
1444 f.trap(NoSuchChildError)
1445 return node.create_subdirectory(path[0])
1446 d.addErrback(_maybe_create)
1447 d.addCallback(self._get_or_create_directories, path[1:], metadata)
1450 def removeFile(self, pathstring):
1451 request = ".removeFile(%r)" % (pathstring,)
1452 self.log(request, level=OPERATIONAL)
1454 path = self._path_from_string(pathstring)
1455 d = self._remove_object(path, must_be_file=True)
1456 d.addBoth(_convert_error, request)
1459 def removeDirectory(self, pathstring):
1460 request = ".removeDirectory(%r)" % (pathstring,)
1461 self.log(request, level=OPERATIONAL)
1463 path = self._path_from_string(pathstring)
1464 d = self._remove_object(path, must_be_directory=True)
1465 d.addBoth(_convert_error, request)
1468 def _remove_object(self, path, must_be_directory=False, must_be_file=False):
1469 userpath = self._path_to_utf8(path)
1470 d = self._get_parent_or_node(path)
1471 def _got_parent( (parent, childname) ):
1472 if childname is None:
1473 raise SFTPError(FX_NO_SUCH_FILE, "cannot remove an object specified by URI")
1475 direntry = self._direntry_for(parent, childname)
1476 d2 = defer.succeed(False)
1477 if not must_be_directory:
1478 d2.addCallback(lambda ign: self._abandon_any_heisenfiles(userpath, direntry))
1480 d2.addCallback(lambda abandoned:
1481 parent.delete(childname, must_exist=not abandoned,
1482 must_be_directory=must_be_directory, must_be_file=must_be_file))
1484 d.addCallback(_got_parent)
1487 def openDirectory(self, pathstring):
1488 request = ".openDirectory(%r)" % (pathstring,)
1489 self.log(request, level=OPERATIONAL)
1491 path = self._path_from_string(pathstring)
1492 d = self._get_parent_or_node(path)
1493 def _got_parent_or_node( (parent_or_node, childname) ):
1494 if noisy: self.log("_got_parent_or_node( (%r, %r) ) in openDirectory(%r)" %
1495 (parent_or_node, childname, pathstring), level=NOISY)
1496 if childname is None:
1497 return parent_or_node
1499 return parent_or_node.get(childname)
1500 d.addCallback(_got_parent_or_node)
1502 if dirnode.is_unknown():
1503 raise SFTPError(FX_PERMISSION_DENIED,
1504 "cannot list an unknown cap as a directory. Upgrading the gateway "
1505 "to a later Tahoe-LAFS version may help")
1506 if not IDirectoryNode.providedBy(dirnode):
1507 raise SFTPError(FX_PERMISSION_DENIED,
1508 "cannot list a file as if it were a directory")
1511 def _render(children):
1512 parent_readonly = dirnode.is_readonly()
1514 for filename, (child, metadata) in children.iteritems():
1515 # The file size may be cached or absent.
1516 metadata['readonly'] = _is_readonly(parent_readonly, child)
1517 attrs = _populate_attrs(child, metadata)
1518 filename_utf8 = filename.encode('utf-8')
1519 longname = _lsLine(filename_utf8, attrs)
1520 results.append( (filename_utf8, longname, attrs) )
1521 return StoppableList(results)
1522 d2.addCallback(_render)
1524 d.addCallback(_list)
1525 d.addBoth(_convert_error, request)
1528 def getAttrs(self, pathstring, followLinks):
1529 request = ".getAttrs(%r, followLinks=%r)" % (pathstring, followLinks)
1530 self.log(request, level=OPERATIONAL)
1532 # When asked about a specific file, report its current size.
1533 # TODO: the modification time for a mutable file should be
1534 # reported as the update time of the best version. But that
1535 # information isn't currently stored in mutable shares, I think.
1537 # Some clients will incorrectly try to get the attributes
1538 # of a file immediately after opening it, before it has been put
1539 # into the all_heisenfiles table. This is a race condition bug in
1540 # the client, but we probably need to handle it anyway.
1542 path = self._path_from_string(pathstring)
1543 userpath = self._path_to_utf8(path)
1544 d = self._get_parent_or_node(path)
1545 def _got_parent_or_node( (parent_or_node, childname) ):
1546 if noisy: self.log("_got_parent_or_node( (%r, %r) )" % (parent_or_node, childname), level=NOISY)
1548 direntry = self._direntry_for(parent_or_node, childname)
1549 d2 = self._sync_heisenfiles(userpath, direntry)
1551 if childname is None:
1552 node = parent_or_node
1553 d2.addCallback(lambda ign: node.get_current_size())
1554 d2.addCallback(lambda size:
1555 _populate_attrs(node, {'readonly': node.is_unknown() or node.is_readonly()}, size=size))
1557 parent = parent_or_node
1558 d2.addCallback(lambda ign: parent.get_child_and_metadata_at_path([childname]))
1559 def _got( (child, metadata) ):
1560 if noisy: self.log("_got( (%r, %r) )" % (child, metadata), level=NOISY)
1561 assert IDirectoryNode.providedBy(parent), parent
1562 metadata['readonly'] = _is_readonly(parent.is_readonly(), child)
1563 d3 = child.get_current_size()
1564 d3.addCallback(lambda size: _populate_attrs(child, metadata, size=size))
1567 if noisy: self.log("_nosuch(%r)" % (err,), level=NOISY)
1568 err.trap(NoSuchChildError)
1569 direntry = self._direntry_for(parent, childname)
1570 if noisy: self.log("checking open files:\nself._heisenfiles = %r\nall_heisenfiles = %r\ndirentry=%r" %
1571 (self._heisenfiles, all_heisenfiles, direntry), level=NOISY)
1572 if direntry in all_heisenfiles:
1573 (files, opentime) = all_heisenfiles[direntry]
1574 sftptime = _to_sftp_time(opentime)
1575 # A file that has been opened for writing necessarily has permissions rw-rw-rw-.
1576 return {'permissions': S_IFREG | 0666,
1578 'createtime': sftptime,
1584 d2.addCallbacks(_got, _nosuch)
1586 d.addCallback(_got_parent_or_node)
1587 d.addBoth(_convert_error, request)
1590 def setAttrs(self, pathstring, attrs):
1591 self.log(".setAttrs(%r, %r)" % (pathstring, attrs), level=OPERATIONAL)
1594 # this would require us to download and re-upload the truncated/extended
1596 def _unsupported(): raise SFTPError(FX_OP_UNSUPPORTED, "setAttrs wth size attribute unsupported")
1597 return defer.execute(_unsupported)
1598 return defer.succeed(None)
1600 def readLink(self, pathstring):
1601 self.log(".readLink(%r)" % (pathstring,), level=OPERATIONAL)
1603 def _unsupported(): raise SFTPError(FX_OP_UNSUPPORTED, "readLink")
1604 return defer.execute(_unsupported)
1606 def makeLink(self, linkPathstring, targetPathstring):
1607 self.log(".makeLink(%r, %r)" % (linkPathstring, targetPathstring), level=OPERATIONAL)
1609 # If this is implemented, note the reversal of arguments described in point 7 of
1610 # <http://www.openbsd.org/cgi-bin/cvsweb/src/usr.bin/ssh/PROTOCOL?rev=1.15>.
1612 def _unsupported(): raise SFTPError(FX_OP_UNSUPPORTED, "makeLink")
1613 return defer.execute(_unsupported)
1615 def extendedRequest(self, extensionName, extensionData):
1616 self.log(".extendedRequest(%r, <data of length %r>)" % (extensionName, len(extensionData)), level=OPERATIONAL)
1618 # We implement the three main OpenSSH SFTP extensions; see
1619 # <http://www.openbsd.org/cgi-bin/cvsweb/src/usr.bin/ssh/PROTOCOL?rev=1.15>
1621 if extensionName == 'posix-rename@openssh.com':
1622 def _bad(): raise SFTPError(FX_BAD_MESSAGE, "could not parse posix-rename@openssh.com request")
1624 (fromPathLen,) = struct.unpack('>L', extensionData[0:4])
1625 if 8 + fromPathLen > len(extensionData): return defer.execute(_bad)
1627 (toPathLen,) = struct.unpack('>L', extensionData[(4 + fromPathLen):(8 + fromPathLen)])
1628 if 8 + fromPathLen + toPathLen != len(extensionData): return defer.execute(_bad)
1630 fromPathstring = extensionData[4:(4 + fromPathLen)]
1631 toPathstring = extensionData[(8 + fromPathLen):]
1632 d = self.renameFile(fromPathstring, toPathstring, overwrite=True)
1634 # Twisted conch assumes that the response from an extended request is either
1635 # an error, or an FXP_EXTENDED_REPLY. But it happens to do the right thing
1636 # (respond with an FXP_STATUS message) if we return a Failure with code FX_OK.
1637 def _succeeded(ign):
1638 raise SFTPError(FX_OK, "request succeeded")
1639 d.addCallback(_succeeded)
1642 if extensionName == 'statvfs@openssh.com' or extensionName == 'fstatvfs@openssh.com':
1643 return defer.succeed(struct.pack('>11Q',
1644 1024, # uint64 f_bsize /* file system block size */
1645 1024, # uint64 f_frsize /* fundamental fs block size */
1646 628318530, # uint64 f_blocks /* number of blocks (unit f_frsize) */
1647 314159265, # uint64 f_bfree /* free blocks in file system */
1648 314159265, # uint64 f_bavail /* free blocks for non-root */
1649 200000000, # uint64 f_files /* total file inodes */
1650 100000000, # uint64 f_ffree /* free file inodes */
1651 100000000, # uint64 f_favail /* free file inodes for non-root */
1652 0x1AF5, # uint64 f_fsid /* file system id */
1653 2, # uint64 f_flag /* bit mask = ST_NOSUID; not ST_RDONLY */
1654 65535, # uint64 f_namemax /* maximum filename length */
1657 def _unsupported(): raise SFTPError(FX_OP_UNSUPPORTED, "unsupported %r request <data of length %r>" %
1658 (extensionName, len(extensionData)))
1659 return defer.execute(_unsupported)
1661 def realPath(self, pathstring):
1662 self.log(".realPath(%r)" % (pathstring,), level=OPERATIONAL)
1664 return self._path_to_utf8(self._path_from_string(pathstring))
1666 def _path_to_utf8(self, path):
1667 return (u"/" + u"/".join(path)).encode('utf-8')
1669 def _path_from_string(self, pathstring):
1670 if noisy: self.log("CONVERT %r" % (pathstring,), level=NOISY)
1672 # The home directory is the root directory.
1673 pathstring = pathstring.strip("/")
1674 if pathstring == "" or pathstring == ".":
1677 path_utf8 = pathstring.split("/")
1679 # <http://tools.ietf.org/html/draft-ietf-secsh-filexfer-02#section-6.2>
1680 # "Servers SHOULD interpret a path name component ".." as referring to
1681 # the parent directory, and "." as referring to the current directory."
1683 for p_utf8 in path_utf8:
1685 # ignore excess .. components at the root
1690 p = p_utf8.decode('utf-8', 'strict')
1691 except UnicodeError:
1692 raise SFTPError(FX_NO_SUCH_FILE, "path could not be decoded as UTF-8")
1695 if noisy: self.log(" PATH %r" % (path,), level=NOISY)
1698 def _get_root(self, path):
1699 # return Deferred (root, remaining_path)
1700 d = defer.succeed(None)
1701 if path and path[0] == u"uri":
1702 d.addCallback(lambda ign: self._client.create_node_from_uri(path[1].encode('utf-8')))
1703 d.addCallback(lambda root: (root, path[2:]))
1705 d.addCallback(lambda ign: (self._root, path))
1708 def _get_parent_or_node(self, path):
1709 # return Deferred (parent, childname) or (node, None)
1710 d = self._get_root(path)
1711 def _got_root( (root, remaining_path) ):
1712 if not remaining_path:
1715 d2 = root.get_child_at_path(remaining_path[:-1])
1716 d2.addCallback(lambda parent: (parent, remaining_path[-1]))
1718 d.addCallback(_got_root)
1721 def _attrs_to_metadata(self, attrs):
1725 if key == "mtime" or key == "ctime" or key == "createtime":
1726 metadata[key] = long(attrs[key])
1727 elif key.startswith("ext_"):
1728 metadata[key] = str(attrs[key])
1733 class SFTPUser(ConchUser, PrefixingLogMixin):
1734 implements(ISession)
1735 def __init__(self, check_abort, client, rootnode, username, convergence):
1736 ConchUser.__init__(self)
1737 PrefixingLogMixin.__init__(self, facility="tahoe.sftp")
1739 self.channelLookup["session"] = session.SSHSession
1740 self.subsystemLookup["sftp"] = FileTransferServer
1742 self.check_abort = check_abort
1743 self.client = client
1744 self.root = rootnode
1745 self.username = username
1746 self.convergence = convergence
1748 def getPty(self, terminal, windowSize, attrs):
1749 self.log(".getPty(%r, %r, %r)" % (terminal, windowSize, attrs), level=OPERATIONAL)
1750 raise NotImplementedError
1752 def openShell(self, protocol):
1753 self.log(".openShell(%r)" % (protocol,), level=OPERATIONAL)
1754 raise NotImplementedError
1756 def execCommand(self, protocol, cmd):
1757 self.log(".execCommand(%r, %r)" % (protocol, cmd), level=OPERATIONAL)
1758 raise NotImplementedError
1760 def windowChanged(self, newWindowSize):
1761 self.log(".windowChanged(%r)" % (newWindowSize,), level=OPERATIONAL)
1764 self.log(".eofReceived()", level=OPERATIONAL)
1767 self.log(".closed()", level=OPERATIONAL)
1770 # if you have an SFTPUser, and you want something that provides ISFTPServer,
1771 # then you get SFTPHandler(user)
1772 components.registerAdapter(SFTPHandler, SFTPUser, ISFTPServer)
1774 from auth import AccountURLChecker, AccountFileChecker, NeedRootcapLookupScheme
1777 implements(portal.IRealm)
1778 def __init__(self, client):
1779 self._client = client
1781 def requestAvatar(self, avatarID, mind, interface):
1782 assert interface == IConchUser, interface
1783 rootnode = self._client.create_node_from_uri(avatarID.rootcap)
1784 handler = SFTPUserHandler(self._client, rootnode, avatarID.username)
1785 return (interface, handler, handler.logout)
1788 class SFTPServer(service.MultiService):
1789 def __init__(self, client, accountfile, accounturl,
1790 sftp_portstr, pubkey_file, privkey_file):
1791 service.MultiService.__init__(self)
1793 r = Dispatcher(client)
1794 p = portal.Portal(r)
1797 c = AccountFileChecker(self, accountfile)
1798 p.registerChecker(c)
1800 c = AccountURLChecker(self, accounturl)
1801 p.registerChecker(c)
1802 if not accountfile and not accounturl:
1803 # we could leave this anonymous, with just the /uri/CAP form
1804 raise NeedRootcapLookupScheme("must provide an account file or URL")
1806 pubkey = keys.Key.fromFile(pubkey_file)
1807 privkey = keys.Key.fromFile(privkey_file)
1808 class SSHFactory(factory.SSHFactory):
1809 publicKeys = {pubkey.sshType(): pubkey}
1810 privateKeys = {privkey.sshType(): privkey}
1811 def getPrimes(self):
1813 # if present, this enables diffie-hellman-group-exchange
1814 return primes.parseModuliFile("/etc/ssh/moduli")
1821 s = strports.service(sftp_portstr, f)
1822 s.setServiceParent(self)