config file parser

copyparty/__init__.py

@@ -11,3 +11,18 @@ PY2 = sys.version_info[0] == 2
 if PY2:
     sys.dont_write_bytecode = True
 
+
+class EnvParams(object):
+    def __init__(self):
+        if sys.platform == "win32":
+            self.cfg = os.path.normpath(os.environ["APPDATA"] + "/copyparty")
+        elif sys.platform == "darwin":
+            self.cfg = os.path.expanduser("~/Library/Preferences/copyparty")
+        else:
+            self.cfg = os.path.normpath(
+                os.getenv("XDG_CONFIG_HOME", os.path.expanduser("~/.config"))
+                + "/copyparty"
+            )
+
+
+E = EnvParams()

copyparty/authsrv.py

@@ -136,6 +136,54 @@ class AuthSrv(object):
         else:
             return {v: k for k, v in orig.items()}
 
+    def laggy_iter(self, iterable):
+        """returns [value,isFinalValue]"""
+        it = iter(iterable)
+        prev = next(it)
+        for x in it:
+            yield prev, False
+            prev = x
+
+        yield prev, True
+
+    def _parse_config_file(self, fd, user, mread, mwrite, mount):
+        vol_src = None
+        vol_dst = None
+        for ln in [x.decode("utf-8").strip() for x in fd]:
+            if not ln and vol_src is not None:
+                vol_src = None
+                vol_dst = None
+
+            if not ln or ln.startswith("#"):
+                continue
+
+            if vol_src is None:
+                if ln.startswith("u "):
+                    u, p = ln[2:].split(":", 1)
+                    user[u] = p
+                else:
+                    vol_src = ln
+                continue
+
+            if vol_src and vol_dst is None:
+                vol_dst = ln
+                if not vol_dst.startswith("/"):
+                    raise Exception('invalid mountpoint "{}"'.format(vol_dst))
+
+                # cfg files override arguments and previous files
+                vol_src = os.path.abspath(vol_src)
+                vol_dst = vol_dst.strip("/")
+                mount[vol_dst] = vol_src
+                mread[vol_dst] = []
+                mwrite[vol_dst] = []
+                continue
+
+            lvl, uname = ln.split(" ")
+            if lvl in "ra":
+                mread[vol_dst].append(uname)
+            if lvl in "wa":
+                mwrite[vol_dst].append(uname)
+
     def reload(self):
         """
         construct a flat list of mountpoints and usernames
@@ -174,11 +222,9 @@ class AuthSrv(object):
                         mwrite[dst].append(uname)
 
         if self.args.c:
-            for logfile in self.args.c:
-                with open(logfile, "rb") as f:
-                    for ln in [x.decode("utf-8").rstrip() for x in f]:
-                        # self.log(ln)
-                        pass
+            for cfg_fn in self.args.c:
+                with open(cfg_fn, "rb") as f:
+                    self._parse_config_file(f, user, mread, mwrite, mount)
 
         # -h says our defaults are CWD at root and read/write for everyone
         vfs = VFS(os.path.abspath("."), "", ["*"], ["*"])

tests/test_vfs.py

@@ -7,6 +7,8 @@ import json
 import shutil
 import unittest
 
+from io import StringIO
+from textwrap import dedent
 from argparse import Namespace
 from copyparty.authsrv import *
 
@@ -173,4 +175,39 @@ class TestVFS(unittest.TestCase):
         self.assertEqual(r1, r2)
         self.assertEqual(v1, v2)
 
+        # config file parser
+        cfg_path = "/dev/shm/test.cfg"
+        with open(cfg_path, "wb") as f:
+            f.write(
+                dedent(
+                    """
+                    u a:123
+                    u asd:fgh:jkl
+                    
+                    ./src
+                    /dst
+                    r a
+                    a asd
+                    """
+                ).encode("utf-8")
+            )
+
+        au = AuthSrv(Namespace(c=[cfg_path], a=[], v=[]), None)
+        self.assertEqual(au.user["a"], "123")
+        self.assertEqual(au.user["asd"], "fgh:jkl")
+        n = au.vfs
+        # root was not defined, so PWD with everyone r/w
+        self.assertEqual(n.vpath, "")
+        self.assertEqual(n.realpath, td)
+        self.assertEqual(n.uread, ["*"])
+        self.assertEqual(n.uwrite, ["*"])
+        self.assertEqual(len(n.nodes), 1)
+        n = n.nodes["dst"]
+        self.assertEqual(n.vpath, "dst")
+        self.assertEqual(n.realpath, td + "/src")
+        self.assertEqual(n.uread, ["a", "asd"])
+        self.assertEqual(n.uwrite, ["asd"])
+        self.assertEqual(len(n.nodes), 0)
+
         shutil.rmtree(td)
+        os.unlink(cfg_path)