tahoe-lafs/contrib/fuse/impl_c/blackmatch.py

#!/usr/bin/env python

#-----------------------------------------------------------------------------------------------
from allmydata.uri import CHKFileURI, NewDirectoryURI, LiteralFileURI
from allmydata.scripts.common_http import do_http as do_http_req
from allmydata.util.hashutil import tagged_hash
from allmydata.util.assertutil import precondition
from allmydata.util import base32, fileutil
from allmydata.scripts.common import get_aliases

from twisted.python import usage
from twisted.internet.protocol import Factory, Protocol
from twisted.internet import reactor, defer

import base64
import errno
import sha
import socket
import stat
import subprocess
import sys
import os
#import pprint

# one needs either python-fuse to have been installed in sys.path, or
# suitable affordances to be made in the build or runtime environment
import fuse

import time
import traceback
import simplejson
import urllib

VERSIONSTR="0.7"

USAGE = 'usage: tahoe fuse [dir_cap_name] [fuse_options] mountpoint'
DEFAULT_DIRECTORY_VALIDITY=26

if not hasattr(fuse, '__version__'):
    raise RuntimeError, \
        "your fuse-py doesn't know of fuse.__version__, probably it's too old."

fuse.fuse_python_api = (0, 2)
fuse.feature_assert('stateful_files', 'has_init')

class TahoeFuseOptions(usage.Options):
    optParameters = [
        ["node-directory", None, "~/.tahoe",
         "Look here to find out which Tahoe node should be used for all "
         "operations. The directory should either contain a full Tahoe node, "
         "or a file named node.url which points to some other Tahoe node. "
         "It should also contain a file named private/aliases which contains "
         "the mapping from alias name to root dirnode URI."
         ],
        ["node-url", None, None,
         "URL of the tahoe node to use, a URL like \"http://127.0.0.1:8123\". "
         "This overrides the URL found in the --node-directory ."],
        ["alias", None, None,
         "Which alias should be mounted."],
        ["root-uri", None, None,
         "Which root directory uri should be mounted."],
        ["cache-timeout", None, 20,
         "Time, in seconds, to cache directory data."],
        ]
    optFlags = [
        ['no-split', None,
         'run stand-alone; no splitting into client and server'],
        ['server', None,
         'server mode (should not be used by end users)'],
        ['server-shutdown', None,
         'shutdown server (should not be used by end users)'],
         ]

    def __init__(self):
        usage.Options.__init__(self)
        self.fuse_options = []
        self.mountpoint = None

    def opt_option(self, fuse_option):
        """
        Pass mount options directly to fuse.  See below.
        """
        self.fuse_options.append(fuse_option)
        
    opt_o = opt_option

    def parseArgs(self, mountpoint=''):
        self.mountpoint = mountpoint

    def getSynopsis(self):
        return "%s [options] mountpoint" % (os.path.basename(sys.argv[0]),)

logfile = file('tfuse.log', 'ab')

def reopen_logfile(fname):
    global logfile
    log('switching to %s' % (fname,))
    logfile.close()
    logfile = file(fname, 'ab')

def log(msg):
    logfile.write("%s: %s\n" % (time.asctime(), msg))
    #time.sleep(0.1)
    logfile.flush()

fuse.flog = log

def unicode_to_utf8(u):
    precondition(isinstance(u, unicode), repr(u))
    return u.encode('utf-8')

def do_http(method, url, body=''):
    resp = do_http_req(method, url, body)
    log('do_http(%s, %s) -> %s, %s' % (method, url, resp.status, resp.reason))
    if resp.status not in (200, 201):
        raise RuntimeError('http response (%s, %s)' % (resp.status, resp.reason))
    else:
        return resp.read()

def flag2mode(flags):
    log('flag2mode(%r)' % (flags,))
    #md = {os.O_RDONLY: 'r', os.O_WRONLY: 'w', os.O_RDWR: 'w+'}
    md = {os.O_RDONLY: 'rb', os.O_WRONLY: 'wb', os.O_RDWR: 'w+b'}
    m = md[flags & (os.O_RDONLY | os.O_WRONLY | os.O_RDWR)]

    if flags & os.O_APPEND:
        m = m.replace('w', 'a', 1)

    return m

class TFSIOError(IOError):
    pass

class ENOENT(TFSIOError):
    def __init__(self, msg):
        TFSIOError.__init__(self, errno.ENOENT, msg)

class EINVAL(TFSIOError):
    def __init__(self, msg):
        TFSIOError.__init__(self, errno.EINVAL, msg)

class EACCESS(TFSIOError):
    def __init__(self, msg):
        TFSIOError.__init__(self, errno.EACCESS, msg)

class EEXIST(TFSIOError):
    def __init__(self, msg):
        TFSIOError.__init__(self, errno.EEXIST, msg)

def logargsretexc(meth):
    def inner_logargsretexc(self, *args, **kwargs):
        log("%s(%r, %r)" % (meth, args, kwargs))
        try:
            ret = meth(self, *args, **kwargs)
        except:
            log('exception:\n%s' % (traceback.format_exc(),))
            raise
        log("ret: %r" % (ret, ))
        return ret
    inner_logargsretexc.__name__ = '<logwrap(%s)>' % (meth,)
    return inner_logargsretexc

def logexc(meth):
    def inner_logexc(self, *args, **kwargs):
        try:
            ret = meth(self, *args, **kwargs)
        except TFSIOError, tie:
            log('error: %s' % (tie,))
            raise
        except:
            log('exception:\n%s' % (traceback.format_exc(),))
            raise
        return ret
    inner_logexc.__name__ = '<logwrap(%s)>' % (meth,)
    return inner_logexc

def log_exc():
    log('exception:\n%s' % (traceback.format_exc(),))

class TahoeFuseFile(object):

    #def __init__(self, path, flags, *mode):
    def __init__(self, tfs, path, flags, *mode):
        log("TFF: __init__(%r, %r, %r)" % (path, flags, mode))
        self.tfs = tfs

        self._path = path # for tahoe put
        try:
            self.parent, self.name, self.fnode = self.tfs.get_parent_name_and_child(path)
            m = flag2mode(flags)
            log('TFF: flags2(mode) -> %s' % (m,))
            if m[0] in 'wa':
                # write
                self.fname = self.tfs.cache.tmp_file(os.urandom(20))
                if self.fnode is None:
                    log('TFF: [%s] open() for write: no file node, creating new File %s' % (self.name, self.fname, ))
                    self.fnode = File(0, 'URI:LIT:')
                    self.fnode.tmp_fname = self.fname # XXX kill this
                    self.parent.add_child(self.name, self.fnode, {})
                elif hasattr(self.fnode, 'tmp_fname'):
                    self.fname = self.fnode.tmp_fname
                    log('TFF: [%s] open() for write: existing file node lists %s' % (self.name, self.fname, ))
                else:
                    log('TFF: [%s] open() for write: existing file node lists no tmp_file, using new %s' % (self.name, self.fname, ))
                self.file = os.fdopen(os.open(self.fname, flags|os.O_CREAT, *mode), m)
                self.fd = self.file.fileno()
                log('TFF: opened(%s) for write' % self.fname)
                self.open_for_write = True
            else:
                # read
                assert self.fnode is not None
                uri = self.fnode.get_uri()

                # XXX make this go away
                if hasattr(self.fnode, 'tmp_fname'):
                    self.fname = self.fnode.tmp_fname
                    log('TFF: reopening(%s) for reading' % self.fname)
                else:
                    log('TFF: fetching file from cache for reading')
                    self.fname = self.tfs.cache.get_file(uri)

                self.file = os.fdopen(os.open(self.fname, flags, *mode), m)
                self.fd = self.file.fileno()
                self.open_for_write = False
                log('TFF: opened(%s) for read' % self.fname)
        except:
            log_exc()
            raise

    def log(self, msg):
        log("<TFF(%s:%s)> %s" % (os.path.basename(self.fname), self.name, msg))

    @logexc
    def read(self, size, offset):
        self.log('read(%r, %r)' % (size, offset, ))
        self.file.seek(offset)
        return self.file.read(size)

    @logexc
    def write(self, buf, offset):
        self.log("write(-%s-, %r)" % (len(buf), offset))
        if not self.open_for_write:
            return -errno.EACCES
        self.file.seek(offset)
        self.file.write(buf)
        return len(buf)

    @logexc
    def release(self, flags):
        self.log("release(%r)" % (flags,))
        self.file.close()
        if self.open_for_write:
            size = os.path.getsize(self.fname)
            self.fnode.size = size
            file_cap = self.tfs.upload(self.fname)
            self.fnode.ro_uri = file_cap
            # XXX [ ] TODO: set metadata
            # write new uri into parent dir entry
            self.parent.add_child(self.name, self.fnode, {})
            self.log("uploaded: %s" % (file_cap,))

        # dbg
        print_tree()

    def _fflush(self):
        if 'w' in self.file.mode or 'a' in self.file.mode:
            self.file.flush()

    @logexc
    def fsync(self, isfsyncfile):
        self.log("fsync(%r)" % (isfsyncfile,))
        self._fflush()
        if isfsyncfile and hasattr(os, 'fdatasync'):
            os.fdatasync(self.fd)
        else:
            os.fsync(self.fd)

    @logexc
    def flush(self):
        self.log("flush()")
        self._fflush()
        # cf. xmp_flush() in fusexmp_fh.c
        os.close(os.dup(self.fd))

    @logexc
    def fgetattr(self):
        self.log("fgetattr()")
        s = os.fstat(self.fd)
        self.log("fgetattr() -> %r" % (s,))
        return stat_to_dict(s)

    @logexc
    def ftruncate(self, len):
        self.log("ftruncate(%r)" % (len,))
        self.file.truncate(len)

class TahoeFuseBase(object):

    def __init__(self, tfs):
        log("TFB: __init__()")
        self.tfs = tfs
        self.files = {}

    def log(self, msg):
        log("<TFB> %s" % (msg, ))

    @logexc
    def readlink(self, path):
        self.log("readlink(%r)" % (path,))
        node = self.tfs.get_path(path)
        if node:
            raise EINVAL('Not a symlink') # nothing in tahoe is a symlink
        else:
            raise ENOENT('Invalid argument')

    @logexc
    def unlink(self, path):
        self.log("unlink(%r)" % (path,))
        self.tfs.unlink(path)

    @logexc
    def rmdir(self, path):
        self.log("rmdir(%r)" % (path,))
        self.tfs.unlink(path)

    @logexc
    def symlink(self, path, path1):
        self.log("symlink(%r, %r)" % (path, path1))
        self.tfs.link(path, path1)

    @logexc
    def rename(self, path, path1):
        self.log("rename(%r, %r)" % (path, path1))
        self.tfs.rename(path, path1)

    @logexc
    def link(self, path, path1):
        self.log("link(%r, %r)" % (path, path1))
        self.tfs.link(path, path1)

    @logexc
    def chmod(self, path, mode):
        self.log("XX chmod(%r, %r)" % (path, mode))
        #return -errno.EOPNOTSUPP

    @logexc
    def chown(self, path, user, group):
        self.log("XX chown(%r, %r, %r)" % (path, user, group))
        #return -errno.EOPNOTSUPP

    @logexc
    def truncate(self, path, len):
        self.log("XX truncate(%r, %r)" % (path, len))
        #return -errno.EOPNOTSUPP

    @logexc
    def utime(self, path, times):
        self.log("XX utime(%r, %r)" % (path, times))
        #return -errno.EOPNOTSUPP

    @logexc
    def statfs(self):
        self.log("statfs()")
        """
        Should return an object with statvfs attributes (f_bsize, f_frsize...).
        Eg., the return value of os.statvfs() is such a thing (since py 2.2).
        If you are not reusing an existing statvfs object, start with
        fuse.StatVFS(), and define the attributes.

        To provide usable information (ie., you want sensible df(1)
        output, you are suggested to specify the following attributes:

            - f_bsize - preferred size of file blocks, in bytes
            - f_frsize - fundamental size of file blcoks, in bytes
                [if you have no idea, use the same as blocksize]
            - f_blocks - total number of blocks in the filesystem
            - f_bfree - number of free blocks
            - f_files - total number of file inodes
            - f_ffree - nunber of free file inodes
        """

        block_size = 4096 # 4k
        preferred_block_size = 131072 # 128k, c.f. seg_size
        fs_size = 8*2**40 # 8Tb
        fs_free = 2*2**40 # 2Tb

        #s = fuse.StatVfs(f_bsize = preferred_block_size,
        s = dict(f_bsize = preferred_block_size,
                         f_frsize = block_size,
                         f_blocks = fs_size / block_size,
                         f_bfree = fs_free / block_size,
                         f_bavail = fs_free / block_size,
                         f_files = 2**30, # total files
                         f_ffree = 2**20, # available files
                         f_favail = 2**20, # available files (root)
                         f_flag = 2, # no suid
                         f_namemax = 255) # max name length
        #self.log('statfs(): %r' % (s,))
        return s

    def fsinit(self):
        self.log("fsinit()")

    ##################################################################

    @logexc
    def readdir(self, path, offset):
        self.log('readdir(%r, %r)' % (path, offset))
        node = self.tfs.get_path(path)
        if node is None:
            return -errno.ENOENT
        dirlist = ['.', '..'] + node.children.keys()
        self.log('dirlist = %r' % (dirlist,))
        #return [fuse.Direntry(d) for d in dirlist]
        return dirlist

    @logexc
    def getattr(self, path):
        self.log('getattr(%r)' % (path,))

        if path == '/':
            # we don't have any metadata for the root (no edge leading to it)
            mode = (stat.S_IFDIR | 755)
            mtime = self.tfs.root.mtime
            s = TStat({}, st_mode=mode, st_nlink=1, st_mtime=mtime)
            self.log('getattr(%r) -> %r' % (path, s))
            #return s
            return stat_to_dict(s)
            
        parent, name, child = self.tfs.get_parent_name_and_child(path)
        if not child: # implicitly 'or not parent'
            raise ENOENT('No such file or directory')
        return stat_to_dict(parent.get_stat(name))

    @logexc
    def access(self, path, mode):
        self.log("access(%r, %r)" % (path, mode))
        node = self.tfs.get_path(path)
        if not node:
            return -errno.ENOENT
        accmode = os.O_RDONLY | os.O_WRONLY | os.O_RDWR
        if (mode & 0222):
            if not node.writable():
                log('write access denied for %s (req:%o)' % (path, mode, ))
                return -errno.EACCES
        #else:
            #log('access granted for %s' % (path, ))

    @logexc
    def mkdir(self, path, mode):
        self.log("mkdir(%r, %r)" % (path, mode))
        self.tfs.mkdir(path)

    ##################################################################
    # file methods

    def open(self, path, flags):
        self.log('open(%r, %r)' % (path, flags, ))
        if path in self.files:
            # XXX todo [ ] should consider concurrent open files of differing modes
            return
        else:
            tffobj = TahoeFuseFile(self.tfs, path, flags)
            self.files[path] = tffobj

    def create(self, path, flags, mode):
        self.log('create(%r, %r, %r)' % (path, flags, mode))
        if path in self.files:
            # XXX todo [ ] should consider concurrent open files of differing modes
            return
        else:
            tffobj = TahoeFuseFile(self.tfs, path, flags, mode)
            self.files[path] = tffobj

    def _get_file(self, path):
        if not path in self.files:
            raise ENOENT('No such file or directory: %s' % (path,))
        return self.files[path]

    ##

    def read(self, path, size, offset):
        self.log('read(%r, %r, %r)' % (path, size, offset, ))
        return self._get_file(path).read(size, offset)

    @logexc
    def write(self, path, buf, offset):
        self.log("write(%r, -%s-, %r)" % (path, len(buf), offset))
        return self._get_file(path).write(buf, offset)

    @logexc
    def release(self, path, flags):
        self.log("release(%r, %r)" % (path, flags,))
        self._get_file(path).release(flags)
        del self.files[path]

    @logexc
    def fsync(self, path, isfsyncfile):
        self.log("fsync(%r, %r)" % (path, isfsyncfile,))
        return self._get_file(path).fsync(isfsyncfile)

    @logexc
    def flush(self, path):
        self.log("flush(%r)" % (path,))
        return self._get_file(path).flush()

    @logexc
    def fgetattr(self, path):
        self.log("fgetattr(%r)" % (path,))
        return self._get_file(path).fgetattr()

    @logexc
    def ftruncate(self, path, len):
        self.log("ftruncate(%r, %r)" % (path, len,))
        return self._get_file(path).ftruncate(len)

class TahoeFuseLocal(TahoeFuseBase, fuse.Fuse):
    def __init__(self, tfs, *args, **kw):
        log("TFL: __init__(%r, %r)" % (args, kw))
        TahoeFuseBase.__init__(self, tfs)
        fuse.Fuse.__init__(self, *args, **kw)

    def log(self, msg):
        log("<TFL> %s" % (msg, ))

    def main(self, *a, **kw):
        self.log("main(%r, %r)" % (a, kw))
        return fuse.Fuse.main(self, *a, **kw)

    # overrides for those methods which return objects not marshalled
    def fgetattr(self, path):
        return TStat({}, **(TahoeFuseBase.fgetattr(self, path)))

    def getattr(self, path):
        return TStat({}, **(TahoeFuseBase.getattr(self, path)))

    def statfs(self):
        return fuse.StatVfs(**(TahoeFuseBase.statfs(self)))
        #self.log('statfs()')
        #ret = fuse.StatVfs(**(TahoeFuseBase.statfs(self)))
        #self.log('statfs(): %r' % (ret,))
        #return ret

    @logexc
    def readdir(self, path, offset):
        return [ fuse.Direntry(d) for d in TahoeFuseBase.readdir(self, path, offset) ]

class TahoeFuseShim(fuse.Fuse):
    def __init__(self, trpc, *args, **kw):
        log("TF: __init__(%r, %r)" % (args, kw))
        self.trpc = trpc
        fuse.Fuse.__init__(self, *args, **kw)

    def log(self, msg):
        log("<TFs> %s" % (msg, ))

    @logexc
    def readlink(self, path):
        self.log("readlink(%r)" % (path,))
        return self.trpc.call('readlink', path)

    @logexc
    def unlink(self, path):
        self.log("unlink(%r)" % (path,))
        return self.trpc.call('unlink', path)

    @logexc
    def rmdir(self, path):
        self.log("rmdir(%r)" % (path,))
        return self.trpc.call('unlink', path)

    @logexc
    def symlink(self, path, path1):
        self.log("symlink(%r, %r)" % (path, path1))
        return self.trpc.call('link', path, path1)

    @logexc
    def rename(self, path, path1):
        self.log("rename(%r, %r)" % (path, path1))
        return self.trpc.call('rename', path, path1)

    @logexc
    def link(self, path, path1):
        self.log("link(%r, %r)" % (path, path1))
        return self.trpc.call('link', path, path1)

    @logexc
    def chmod(self, path, mode):
        self.log("XX chmod(%r, %r)" % (path, mode))
        return self.trpc.call('chmod', path, mode)

    @logexc
    def chown(self, path, user, group):
        self.log("XX chown(%r, %r, %r)" % (path, user, group))
        return self.trpc.call('chown', path, user, group)

    @logexc
    def truncate(self, path, len):
        self.log("XX truncate(%r, %r)" % (path, len))
        return self.trpc.call('truncate', path, len)

    @logexc
    def utime(self, path, times):
        self.log("XX utime(%r, %r)" % (path, times))
        return self.trpc.call('utime', path, times)

    @logexc
    def statfs(self):
        self.log("statfs()")
        response = self.trpc.call('statfs')
        #self.log("statfs(): %r" % (response,))
        kwargs = dict([ (str(k),v) for k,v in response.items() ])
        return fuse.StatVfs(**kwargs)

    def fsinit(self):
        self.log("fsinit()")

    def main(self, *a, **kw):
        self.log("main(%r, %r)" % (a, kw))

        return fuse.Fuse.main(self, *a, **kw)

    ##################################################################

    @logexc
    def readdir(self, path, offset):
        self.log('readdir(%r, %r)' % (path, offset))
        return [ fuse.Direntry(d) for d in self.trpc.call('readdir', path, offset) ]

    @logexc
    def getattr(self, path):
        self.log('getattr(%r)' % (path,))
        response = self.trpc.call('getattr', path)
        kwargs = dict([ (str(k),v) for k,v in response.items() ])
        s = TStat({}, **kwargs)
        self.log('getattr(%r) -> %r' % (path, s))
        return s

    @logexc
    def access(self, path, mode):
        self.log("access(%r, %r)" % (path, mode))
        return self.trpc.call('access', path, mode)

    @logexc
    def mkdir(self, path, mode):
        self.log("mkdir(%r, %r)" % (path, mode))
        return self.trpc.call('mkdir', path, mode)

    ##################################################################
    # file methods

    def open(self, path, flags):
        self.log('open(%r, %r)' % (path, flags, ))
        return self.trpc.call('open', path, flags)

    def create(self, path, flags, mode):
        self.log('create(%r, %r, %r)' % (path, flags, mode))
        return self.trpc.call('create', path, flags, mode)

    ##

    def read(self, path, size, offset):
        self.log('read(%r, %r, %r)' % (path, size, offset, ))
        return self.trpc.call('read', path, size, offset)

    @logexc
    def write(self, path, buf, offset):
        self.log("write(%r, -%s-, %r)" % (path, len(buf), offset))
        return self.trpc.call('write', path, buf, offset)

    @logexc
    def release(self, path, flags):
        self.log("release(%r, %r)" % (path, flags,))
        return self.trpc.call('release', path, flags)

    @logexc
    def fsync(self, path, isfsyncfile):
        self.log("fsync(%r, %r)" % (path, isfsyncfile,))
        return self.trpc.call('fsync', path, isfsyncfile)

    @logexc
    def flush(self, path):
        self.log("flush(%r)" % (path,))
        return self.trpc.call('flush', path)

    @logexc
    def fgetattr(self, path):
        self.log("fgetattr(%r)" % (path,))
        #return self.trpc.call('fgetattr', path)
        response = self.trpc.call('fgetattr', path)
        kwargs = dict([ (str(k),v) for k,v in response.items() ])
        s = TStat({}, **kwargs)
        self.log('getattr(%r) -> %r' % (path, s))
        return s

    @logexc
    def ftruncate(self, path, len):
        self.log("ftruncate(%r, %r)" % (path, len,))
        return self.trpc.call('ftruncate', path, len)


def launch_tahoe_fuse(tf_class, tobj, argv):
    sys.argv = ['tahoe fuse'] + list(argv)
    log('setting sys.argv=%r' % (sys.argv,))
    config = TahoeFuseOptions()
    version = "%prog " +VERSIONSTR+", fuse "+ fuse.__version__
    server = tf_class(tobj, version=version, usage=config.getSynopsis(), dash_s_do='setsingle')
    server.parse(errex=1)
    server.main()

def getnodeurl(nodedir):
    f = file(os.path.expanduser(os.path.join(nodedir, "node.url")), 'rb')
    nu = f.read().strip()
    f.close()
    if nu[-1] != "/":
        nu += "/"
    return nu

def fingerprint(uri):
    if uri is None:
        return None
    return base64.b32encode(sha.new(uri).digest()).lower()[:6]

stat_fields = [ 'st_mode', 'st_ino', 'st_dev', 'st_nlink', 'st_uid', 'st_gid', 'st_size',
                'st_atime', 'st_mtime', 'st_ctime', ]
def stat_to_dict(statobj, fields=None):
    if fields is None:
        fields = stat_fields
    d = {}
    for f in fields:
        d[f] = getattr(statobj, f, None)
    return d

class TStat(fuse.Stat):
    # in fuse 0.2, these are set by fuse.Stat.__init__
    # in fuse 0.2-pre3 (hardy) they are not. badness unsues if they're missing
    st_mode  = None
    st_ino   = 0
    st_dev   = 0
    st_nlink = None
    st_uid   = 0
    st_gid   = 0
    st_size  = 0
    st_atime = 0
    st_mtime = 0
    st_ctime = 0

    fields = [ 'st_mode', 'st_ino', 'st_dev', 'st_nlink', 'st_uid', 'st_gid', 'st_size',
               'st_atime', 'st_mtime', 'st_ctime', ]
    def __init__(self, metadata, **kwargs):
        # first load any stat fields present in 'metadata'
        for st in [ 'mtime', 'ctime' ]:
            if st in metadata:
                setattr(self, "st_%s" % st, metadata[st])
        for st in self.fields:
            if st in metadata:
                setattr(self, st, metadata[st])

        # then set any values passed in as kwargs
        fuse.Stat.__init__(self, **kwargs)

    def __repr__(self):
        return "<Stat%r>" % (stat_to_dict(self),)

class Directory(object):
    def __init__(self, tfs, ro_uri, rw_uri):
        self.tfs = tfs
        self.ro_uri = ro_uri
        self.rw_uri = rw_uri
        assert (rw_uri or ro_uri)
        self.children = {}
        self.last_load = None
        self.last_data = None
        self.mtime = 0

    def __repr__(self):
        return "<Directory %s>" % (fingerprint(self.get_uri()),)

    def maybe_refresh(self, name=None):
        """
        if the previously cached data was retrieved within the cache
        validity period, does nothing. otherwise refetches the data
        for this directory and reloads itself
        """
        now = time.time()
        if self.last_load is None or (now - self.last_load) > self.tfs.cache_validity:
            self.load(name)

    def load(self, name=None):
        now = time.time()
        log('%s.loading(%s)' % (self, name))
        url = self.tfs.compose_url("uri/%s?t=json", self.get_uri())
        data = urllib.urlopen(url).read()
        h = tagged_hash('cache_hash', data)
        if h == self.last_data:
            self.last_load = now
            log('%s.load() : no change h(data)=%s' % (self, base32.b2a(h), ))
            return
        try:
            parsed = simplejson.loads(data)
        except ValueError:
            log('%s.load(): unable to parse json data for dir:\n%r' % (self, data))
            return
        nodetype, d = parsed
        assert nodetype == 'dirnode'
        self.children.clear()
        for cname,details in d['children'].items():
            cname = unicode_to_utf8(cname)
            ctype, cattrs = details
            metadata = cattrs.get('metadata', {})
            if ctype == 'dirnode':
                cobj = self.tfs.dir_for(cname, cattrs.get('ro_uri'), cattrs.get('rw_uri'))
            else:
                assert ctype == "filenode"
                cobj = File(cattrs.get('size'), cattrs.get('ro_uri'))
            self.children[cname] = cobj, metadata
        self.last_load = now
        self.last_data = h
        self.mtime = now
        log('%s.load() loaded: \n%s' % (self, self.pprint(),))

    def get_children(self):
        return self.children.keys()

    def get_child(self, name):
        return self.children[name][0]

    def add_child(self, name, child, metadata):
        log('%s.add_child(%r, %r, %r)' % (self, name, child, metadata, ))
        self.children[name] = child, metadata
        url = self.tfs.compose_url("uri/%s/%s?t=uri", self.get_uri(), name)
        child_cap = do_http('PUT', url, child.get_uri())
        # XXX [ ] TODO: push metadata to tahoe node
        assert child_cap == child.get_uri()
        self.mtime = time.time()
        log('added child %r with %r to %r' % (name, child_cap, self))

    def remove_child(self, name):
        log('%s.remove_child(%r)' % (self, name, ))
        del self.children[name]
        url = self.tfs.compose_url("uri/%s/%s", self.get_uri(), name)
        resp = do_http('DELETE', url)
        self.mtime = time.time()
        log('child (%s) removal yielded %r' % (name, resp,))

    def get_uri(self):
        return self.rw_uri or self.ro_uri

    def writable(self):
        return self.rw_uri and self.rw_uri != self.ro_uri

    def pprint(self, prefix='', printed=None, suffix=''):
        ret = []
        if printed is None:
            printed = set()
        writable = self.writable() and '+' or ' '
        if self in printed:
            ret.append("         %s/%s ... <%s> : %s" % (prefix, writable, fingerprint(self.get_uri()), suffix, ))
        else:
            ret.append("[%s] %s/%s : %s" % (fingerprint(self.get_uri()), prefix, writable, suffix, ))
            printed.add(self)
            for name,(child,metadata) in sorted(self.children.items()):
                ret.append(child.pprint(' ' * (len(prefix)+1)+name, printed, repr(metadata)))
        return '\n'.join(ret)

    def get_metadata(self, name):
        return self.children[name][1]

    def get_stat(self, name):
        child,metadata = self.children[name]
        log("%s.get_stat(%s) md: %r" % (self, name, metadata))

        if isinstance(child, Directory):
            child.maybe_refresh(name)
            mode = metadata.get('st_mode') or (stat.S_IFDIR | 0755)
            s = TStat(metadata, st_mode=mode, st_nlink=1, st_mtime=child.mtime)
        else:
            if hasattr(child, 'tmp_fname'):
                s = os.stat(child.tmp_fname)
                log("%s.get_stat(%s) returning local stat of tmp file" % (self, name, ))
            else:
                s = TStat(metadata,
                          st_nlink = 1,
                          st_size = child.size,
                          st_mode = metadata.get('st_mode') or (stat.S_IFREG | 0444),
                          st_mtime = metadata.get('mtime') or self.mtime,
                          )
            return s

        log("%s.get_stat(%s)->%s" % (self, name, s))
        return s

class File(object):
    def __init__(self, size, ro_uri):
        self.size = size
        if ro_uri:
            ro_uri = str(ro_uri)
        self.ro_uri = ro_uri

    def __repr__(self):
        return "<File %s>" % (fingerprint(self.ro_uri) or [self.tmp_fname],)

    def pprint(self, prefix='', printed=None, suffix=''):
        return "         %s (%s) : %s" % (prefix, self.size, suffix, )

    def get_uri(self):
        return self.ro_uri

    def writable(self):
        return True

class TFS(object):
    def __init__(self, nodedir, nodeurl, root_uri, 
                       cache_validity_period=DEFAULT_DIRECTORY_VALIDITY):
        self.cache_validity = cache_validity_period
        self.nodeurl = nodeurl
        self.root_uri = root_uri
        self.dirs = {}

        cachedir = os.path.expanduser(os.path.join(nodedir, '_cache'))
        self.cache = FileCache(nodeurl, cachedir)
        ro_uri = NewDirectoryURI.init_from_string(self.root_uri).get_readonly()
        self.root = Directory(self, ro_uri, self.root_uri)
        self.root.maybe_refresh('<root>')

    def log(self, msg):
        log("<TFS> %s" % (msg, ))

    def pprint(self):
        return self.root.pprint()

    def compose_url(self, fmt, *args):
        return self.nodeurl + (fmt % tuple(map(urllib.quote, args)))

    def get_parent_name_and_child(self, path):
        """
        find the parent dir node, name of child relative to that parent, and
        child node within the TFS object space.
        @returns: (parent, name, child) if the child is found
                  (parent, name, None) if the child is missing from the parent
                  (None, name, None) if the parent is not found
        """
        if path == '/':
            return 
        dirname, name = os.path.split(path)
        parent = self.get_path(dirname)
        if parent:
            try:
                child = parent.get_child(name)
                return parent, name, child
            except KeyError:
                return parent, name, None
        else:
            return None, name, None

    def get_path(self, path):
        comps = path.strip('/').split('/')
        if comps == ['']:
            comps = []
        cursor = self.root
        c_name = '<root>'
        for comp in comps:
            if not isinstance(cursor, Directory):
                self.log('path "%s" is not a dir' % (path,))
                return None
            cursor.maybe_refresh(c_name)
            try:
                cursor = cursor.get_child(comp)
                c_name = comp
            except KeyError:
                self.log('path "%s" not found' % (path,))
                return None
        if isinstance(cursor, Directory):
            cursor.maybe_refresh(c_name)
        return cursor

    def dir_for(self, name, ro_uri, rw_uri):
        #self.log('dir_for(%s) [%s/%s]' % (name, fingerprint(ro_uri), fingerprint(rw_uri)))
        if ro_uri:
            ro_uri = str(ro_uri)
        if rw_uri:
            rw_uri = str(rw_uri)
        uri = rw_uri or ro_uri
        assert uri
        dirobj = self.dirs.get(uri)
        if not dirobj:
            self.log('dir_for(%s) creating new Directory' % (name, ))
            dirobj = Directory(self, ro_uri, rw_uri)
            self.dirs[uri] = dirobj
        return dirobj

    def upload(self, fname):
        self.log('upload(%r)' % (fname,))
        fh = file(fname, 'rb')
        url = self.compose_url("uri")
        file_cap = do_http('PUT', url, fh)
        self.log('uploaded to: %r' % (file_cap,))
        return file_cap

    def mkdir(self, path):
        self.log('mkdir(%r)' % (path,))
        parent, name, child = self.get_parent_name_and_child(path)

        if child:
            raise EEXIST('File exists: %s' % (name,))
        if not parent:
            raise ENOENT('No such file or directory: %s' % (path,))

        url = self.compose_url("uri?t=mkdir")
        new_dir_cap = do_http('PUT', url)

        ro_uri = NewDirectoryURI.init_from_string(new_dir_cap).get_readonly()
        child = Directory(self, ro_uri, new_dir_cap)
        parent.add_child(name, child, {})

    def rename(self, path, path1):
        self.log('rename(%s, %s)' % (path, path1))
        src_parent, src_name, src_child = self.get_parent_name_and_child(path)
        dst_parent, dst_name, dst_child = self.get_parent_name_and_child(path1)

        if not src_child or not dst_parent:
            raise ENOENT('No such file or directory')

        dst_parent.add_child(dst_name, src_child, {})
        src_parent.remove_child(src_name)

    def unlink(self, path):
        parent, name, child = self.get_parent_name_and_child(path)

        if child is None: # parent or child is missing
            raise ENOENT('No such file or directory')
        if not parent.writable():
            raise EACCESS('Permission denied')

        parent.remove_child(name)

    def link(self, path, path1):
        src = self.get_path(path)
        dst_parent, dst_name, dst_child = self.get_parent_name_and_child(path1)

        if not src:
            raise ENOENT('No such file or directory')
        if dst_parent is None:
            raise ENOENT('No such file or directory')
        if not dst_parent.writable():
            raise EACCESS('Permission denied')

        dst_parent.add_child(dst_name, src, {})

class FileCache(object):
    def __init__(self, nodeurl, cachedir):
        self.nodeurl = nodeurl
        self.cachedir = cachedir
        if not os.path.exists(self.cachedir):
            os.makedirs(self.cachedir)
        self.tmpdir = os.path.join(self.cachedir, 'tmp')
        if not os.path.exists(self.tmpdir):
            os.makedirs(self.tmpdir)

    def log(self, msg):
        log("<FC> %s" % (msg, ))

    def get_file(self, uri):
        self.log('get_file(%s)' % (uri,))
        if uri.startswith("URI:LIT"):
            return self.get_literal(uri)
        else:
            return self.get_chk(uri)

    def get_literal(self, uri):
        h = sha.new(uri).digest()
        u = LiteralFileURI.init_from_string(uri)
        fname = os.path.join(self.cachedir, '__'+base64.b32encode(h).lower())
        size = len(u.data)
        self.log('writing literal file %s (%s)' % (fname, size, ))
        fh = open(fname, 'wb')
        fh.write(u.data)
        fh.close()
        return fname

    def get_chk(self, uri):
        u = CHKFileURI.init_from_string(str(uri))
        storage_index = u.storage_index
        size = u.size
        fname = os.path.join(self.cachedir, base64.b32encode(storage_index).lower())
        if os.path.exists(fname):
            fsize = os.path.getsize(fname)
            if fsize == size:
                return fname
            else:
                self.log('warning file "%s" is too short %s < %s' % (fname, fsize, size))
        self.log('downloading file %s (%s)' % (fname, size, ))
        fh = open(fname, 'wb')
        url = "%suri/%s" % (self.nodeurl, uri)
        download = urllib.urlopen(''.join([ self.nodeurl, "uri/", uri ]))
        while True:
            chunk = download.read(4096)
            if not chunk:
                break
            fh.write(chunk)
        fh.close()
        return fname

    def tmp_file(self, id):
        fname = os.path.join(self.tmpdir, base64.b32encode(id).lower())
        return fname

_tfs = None # to appease pyflakes; is set in main()
def print_tree():
    log('tree:\n' + _tfs.pprint())


def unmarshal(obj):
    if obj is None or isinstance(obj, int) or isinstance(obj, long) or isinstance(obj, float):
        return obj
    elif isinstance(obj, unicode):
        #log('unmarshal(%r)' % (obj,))
        return base64.b64decode(obj)
    elif isinstance(obj, list):
        return map(unmarshal, obj)
    elif isinstance(obj, dict):
        return dict([ (k,unmarshal(v)) for k,v in obj.items() ])
    else:
        raise ValueError('object type not int,str,list,dict,none (%s) (%r)' % (type(obj), obj))

def marshal(obj):
    if obj is None or isinstance(obj, int) or isinstance(obj, long) or isinstance(obj, float):
        return obj
    elif isinstance(obj, str):
        return base64.b64encode(obj)
    elif isinstance(obj, list) or isinstance(obj, tuple):
        return map(marshal, obj)
    elif isinstance(obj, dict):
        return dict([ (k,marshal(v)) for k,v in obj.items() ])
    else:
        raise ValueError('object type not int,str,list,dict,none (%s)' % type(obj))


class TRPCProtocol(Protocol):
    compute_response_sha1 = True
    log_all_requests = False

    def connectionMade(self):
        self.buf = []

    def dataReceived(self, data):
        if data == 'keepalive\n':
            log('keepalive connection on %r' % (self.transport,))
            self.keepalive = True
            return

        if not data.endswith('\n'):
            self.buf.append(data)
            return
        if self.buf:
            self.buf.append(data)
            reqstr = ''.join(self.buf)
            self.buf = []
            self.dispatch_request(reqstr)
        else:
            self.dispatch_request(data)

    def dispatch_request(self, reqstr):
        try:
            req = simplejson.loads(reqstr)
        except ValueError, ve:
            log(ve)
            return

        d = defer.maybeDeferred(self.handle_request, req)
        d.addCallback(self.send_response)
        d.addErrback(self.send_error)

    def send_error(self, failure):
        log('failure: %s' % (failure,))
        if failure.check(TFSIOError):
            e = failure.value
            self.send_response(['error', 'errno', e.args[0], e.args[1]])
        else:
            self.send_response(['error', 'failure', str(failure)])

    def send_response(self, result):
        response = simplejson.dumps(result)
        header = { 'len': len(response), }
        if self.compute_response_sha1:
            header['sha1'] = base64.b64encode(sha.new(response).digest())
        hdr = simplejson.dumps(header)
        self.transport.write(hdr)
        self.transport.write('\n')
        self.transport.write(response)
        self.transport.loseConnection()

    def connectionLost(self, reason):
        if hasattr(self, 'keepalive'):
            log('keepalive connection %r lost, shutting down' % (self.transport,))
            reactor.callLater(0, reactor.stop)

    def handle_request(self, req):
        if type(req) is not list or not req or len(req) < 1:
            return ['error', 'malformed request']
        if req[0] == 'call':
            if len(req) < 3:
                return ['error', 'malformed request']
            methname = req[1]
            try:
                args = unmarshal(req[2])
            except ValueError, ve:
                return ['error', 'malformed arguments', str(ve)]

            try:
                meth = getattr(self.factory.server, methname)
            except AttributeError, ae:
                return ['error', 'no such method', str(ae)]

            if self.log_all_requests:
                log('call %s(%s)' % (methname, ', '.join(map(repr, args))))
            try:
                result = meth(*args)
            except TFSIOError, e:
                log('errno: %s; %s' % e.args)
                return ['error', 'errno', e.args[0], e.args[1]]
            except Exception, e:
                log('exception: ' + traceback.format_exc())
                return ['error', 'exception', str(e)]
            d = defer.succeed(None)
            d.addCallback(lambda junk: result) # result may be Deferred
            d.addCallback(lambda res: ['result', marshal(res)]) # only applies if not errback
            return d

class TFSServer(object):
    def __init__(self, socket_path, server=None):
        self.socket_path = socket_path
        log('TFSServer init socket: %s' % (socket_path,))

        self.factory = Factory()
        self.factory.protocol = TRPCProtocol
        if server:
            self.factory.server = server
        else:
            self.factory.server = self

    def get_service(self):
        if not hasattr(self, 'svc'):
            from twisted.application import strports
            self.svc = strports.service('unix:'+self.socket_path, self.factory)
        return self.svc

    def run(self):
        svc = self.get_service()
        def ss():
            try:
                svc.startService()
            except:
                reactor.callLater(0, reactor.stop)
                raise
        reactor.callLater(0, ss)
        reactor.run()

    def hello(self):
        return 'pleased to meet you'

    def echo(self, arg):
        return arg

    def failex(self):
        raise ValueError('expected')

    def fail(self):
        return defer.maybeDeferred(self.failex)

class RPCError(RuntimeError):
    pass

class TRPC(object):
    def __init__(self, socket_fname):
        self.socket_fname = socket_fname
        self.keepalive = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.keepalive.connect(self.socket_fname)
        self.keepalive.send('keepalive\n')
        log('requested keepalive on %s' % (self.keepalive,))

    def req(self, req):
        # open conenction to trpc server
        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        s.connect(self.socket_fname)
        # send request
        s.send(simplejson.dumps(req))
        s.send('\n')
        # read response header
        hdr_data = s.recv(8192)
        first_newline = hdr_data.index('\n')
        header = hdr_data[:first_newline]
        data = hdr_data[first_newline+1:]
        hdr = simplejson.loads(header)
        hdr_len = hdr['len']
        if hdr.has_key('sha1'):
            hdr_sha1 = base64.b64decode(hdr['sha1'])
            spool = [data]
            spool_sha = sha.new(data)
            # spool response
            while True:
                data = s.recv(8192)
                if data:
                    spool.append(data)
                    spool_sha.update(data)
                else:
                    break
        else:
            spool = [data]
            # spool response
            while True:
                data = s.recv(8192)
                if data:
                    spool.append(data)
                else:
                    break
        s.close()
        # decode response
        resp = ''.join(spool)
        spool = None
        assert hdr_len == len(resp), str((hdr_len, len(resp), repr(resp)))
        if hdr.has_key('sha1'):
            data_sha1 = spool_sha.digest()
            spool = spool_sha = None
            assert hdr_sha1 == data_sha1, str((base32.b2a(hdr_sha1), base32.b2a(data_sha1)))
        #else:
            #print 'warning, server provided no sha1 to check'
        return resp

    def call(self, methodname, *args):
        res = self.req(['call', methodname, marshal(args)])

        result = simplejson.loads(res)
        if not result or len(result) < 2:
            raise TypeError('malformed response %r' % (result,))
        if result[0] == 'error':
            if result[1] == 'errno':
                raise TFSIOError(result[2], result[3])
            else:
                raise RPCError(*(result[1:])) # error, exception / error, failure
        elif result[0] == 'result':
            return unmarshal(result[1])
        else:
            raise TypeError('unknown response type %r' % (result[0],))

    def shutdown(self):
        log('shutdown() closing keepalive %s' % (self.keepalive,))
        self.keepalive.close()

# (cut-n-pasted here due to an ImportError / some py2app linkage issues)
#from twisted.scripts._twistd_unix import daemonize
def daemonize():
    # See http://www.erlenstar.demon.co.uk/unix/faq_toc.html#TOC16
    if os.fork():   # launch child and...
        os._exit(0) # kill off parent
    os.setsid()
    if os.fork():   # launch child and...
        os._exit(0) # kill off parent again.
    os.umask(077)
    null=os.open('/dev/null', os.O_RDWR)
    for i in range(3):
        try:
            os.dup2(null, i)
        except OSError, e:
            if e.errno != errno.EBADF:
                raise
    os.close(null)

def main(argv):
    log("main(%s)" % (argv,))

    # check for version or help options (no args == help)
    if not argv:
        argv = ['--help']
    if len(argv) == 1 and argv[0] in ['-h', '--help']:
        config = TahoeFuseOptions()
        print >> sys.stderr, config
        print >> sys.stderr, 'fuse usage follows:'
    if len(argv) == 1 and argv[0] in ['-h', '--help', '--version']:
        launch_tahoe_fuse(TahoeFuseLocal, None, argv)
        return -2

    # parse command line options
    config = TahoeFuseOptions()
    try:
        #print 'parsing', argv
        config.parseOptions(argv)
    except usage.error, e:
        print config
        print e
        return -1

    # check for which alias or uri is specified
    if config['alias']:
        alias = config['alias']
        #print 'looking for aliases in', config['node-directory']
        aliases = get_aliases(os.path.expanduser(config['node-directory']))
        if alias not in aliases:
            raise usage.error('Alias %r not found' % (alias,))
        root_uri = aliases[alias]
        root_name = alias
    elif config['root-uri']:
        root_uri = config['root-uri']
        root_name = 'uri_' + base32.b2a(tagged_hash('root_name', root_uri))[:12]
        # test the uri for structural validity:
        try:
            NewDirectoryURI.init_from_string(root_uri)
        except:
            raise usage.error('root-uri must be a valid directory uri (not %r)' % (root_uri,))
    else:
        raise usage.error('At least one of --alias or --root-uri must be specified')

    nodedir = config['node-directory']
    nodeurl = config['node-url']
    if not nodeurl:
        nodeurl = getnodeurl(nodedir)

    # allocate socket
    socket_dir = os.path.join(os.path.expanduser(nodedir), "tfuse.sockets")
    socket_path = os.path.join(socket_dir, root_name)
    if len(socket_path) > 103:
        # try googling AF_UNIX and sun_len for some taste of why this oddity exists.
        raise OSError(errno.ENAMETOOLONG, 'socket path too long (%s)' % (socket_path,))

    fileutil.make_dirs(socket_dir, 0700)
    if os.path.exists(socket_path):
        log('socket exists')
        if config['server-shutdown']:
            log('calling shutdown')
            trpc = TRPC(socket_path)
            result = trpc.shutdown()
            log('result: %r' % (result,))
            log('called shutdown')
            return
        else:
            raise OSError(errno.EEXIST, 'fuse already running (%r exists)' % (socket_path,))
    elif config['server-shutdown']:
        raise OSError(errno.ENOTCONN, '--server-shutdown specified, but server not running')

    if not os.path.exists(config.mountpoint):
        raise OSError(errno.ENOENT, 'No such file or directory: "%s"' % (config.mountpoint,))

    global _tfs
    #
    # Standalone ("no-split")
    #
    if config['no-split']:
        reopen_logfile('tfuse.%s.unsplit.log' % (root_name,))
        log('\n'+(24*'_')+'init (unsplit)'+(24*'_')+'\n')

        cache_timeout = float(config['cache-timeout'])
        tfs = TFS(nodedir, nodeurl, root_uri, cache_timeout)
        #print tfs.pprint()

        # make tfs instance accesible to print_tree() for dbg
        _tfs = tfs

        args = [ '-o'+opt for opt in config.fuse_options ] + [config.mountpoint]
        launch_tahoe_fuse(TahoeFuseLocal, tfs, args)

    #
    # Server
    #
    elif config['server']:
        reopen_logfile('tfuse.%s.server.log' % (root_name,))
        log('\n'+(24*'_')+'init (server)'+(24*'_')+'\n')

        log('daemonizing')
        daemonize()

        cache_timeout = float(config['cache-timeout'])
        tfs = TFS(nodedir, nodeurl, root_uri, cache_timeout)
        #print tfs.pprint()

        # make tfs instance accesible to print_tree() for dbg
        _tfs = tfs

        log('launching tfs server')
        tfuse = TahoeFuseBase(tfs)
        tfs_server = TFSServer(socket_path, tfuse)
        tfs_server.run()
        log('tfs server ran, exiting')

    #
    # Client
    #
    else:
        reopen_logfile('tfuse.%s.client.log' % (root_name,))
        log('\n'+(24*'_')+'init (client)'+(24*'_')+'\n')

        server_args = [sys.executable, sys.argv[0], '--server'] + argv
        if 'Allmydata.app/Contents/MacOS' in sys.executable:
            # in this case blackmatch is the 'fuse' subcommand of the 'tahoe' executable
            # otherwise we assume blackmatch is being run from source
            server_args.insert(2, 'fuse')
        #print 'launching server:', server_args
        server = subprocess.Popen(server_args)
        waiting_since = time.time()
        wait_at_most = 8
        while not os.path.exists(socket_path):
            log('waiting for appearance of %r' % (socket_path,))
            time.sleep(1)
            if time.time() - waiting_since > wait_at_most:
                log('%r did not appear within %ss' % (socket_path, wait_at_most))
                raise IOError(2, 'no socket %s' % (socket_path,))
        #print 'launched server'
        trpc = TRPC(socket_path)


        args = [ '-o'+opt for opt in config.fuse_options ] + [config.mountpoint]
        launch_tahoe_fuse(TahoeFuseShim, trpc, args)

        
if __name__ == '__main__':
    sys.exit(main(sys.argv[1:]))