1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
|
#!/usr/bin/env python3
# Copyright (c) 2018, George Tokmaji
# Permission to use, copy, modify, and/or distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
from base import *
import sys
sys.path.append("../c4group")
import c4group
class C4GroupOperations(LCOperations):
def __init__(self, path):
super().__init__()
self._grp = c4group.C4GroupDirectory(path)
self._grp.load(recursive=False)
self._add_grp_content(llfuse.ROOT_INODE, self._grp)
def _add_grp_content(self, inode, grp):
for i in grp.content:
i["title"] = os.fsdecode(i.filename)
self._add_entry(inode, i, isinstance(i, c4group.C4GroupDirectory))
def _is_dir(self, inode):
return isinstance(self._inode_list[inode], c4group.C4GroupDirectory) or super()._is_dir(inode)
def getattr(self, inode, ctx=None):
entry = self._inode_list[inode]
attr = super().getattr(inode, ctx)
if entry:
attr.st_size = entry.size
return attr
def opendir(self, inode, ctx=None):
if self._is_dir(inode):
try:
self._get_inode_by_parent(inode)
except IndexError:
grp = self._inode_list[inode]
grp.load(recursive=False)
self._add_grp_content(inode, grp)
return inode
def open(self, inode, flags, ctx=None):
entry = self._inode_list[inode]
if self._is_dir(inode):
raise llfuse.FUSEError(errno.EISDIR)
if flags & os.O_APPEND:
raise llfuse.FUSEError(errno.EROFS)
return inode
def read(self, inode, off, size):
entry = self._inode_list[inode]
return entry.content[off:off+size]
if __name__ == "__main__":
# Usage: ./c4groupfs.py <file> <mountpoint>
fuse_options = set(llfuse.default_options)
fuse_options.add("fsname=c4groupfs")
fuse_options.add("debug")
fuse_options.discard("default_permissions")
op = C4GroupOperations(sys.argv[1])
llfuse.init(op, sys.argv[2], fuse_options)
try:
llfuse.main()
finally:
llfuse.close(unmount=True)
|
