strict file validation

5 files changed, 46 insertions, 21 deletions

diff --git a/kvmd/apps/__init__.py b/kvmd/apps/__init__.py
index 04c1b621..c4b0b0bc 100644
--- a/kvmd/apps/__init__.py
+++ b/kvmd/apps/__init__.py
@@ -58,7 +58,8 @@ from ..validators.auth import valid_user
 from ..validators.auth import valid_users_list
 
 from ..validators.os import valid_abs_path
-from ..validators.os import valid_abs_path_exists
+from ..validators.os import valid_abs_file
+from ..validators.os import valid_abs_dir
 from ..validators.os import valid_unix_mode
 from ..validators.os import valid_command
 
@@ -87,7 +88,7 @@ def init(
 
     args_parser = argparse.ArgumentParser(prog=(prog or argv[0]), description=description, add_help=add_help)
     args_parser.add_argument("-c", "--config", dest="config_path", default="/etc/kvmd/main.yaml", metavar="<file>",
-                             type=valid_abs_path_exists, help="Set config file path")
+                             type=valid_abs_file, help="Set config file path")
     args_parser.add_argument("-o", "--set-options", dest="set_options", default=[], nargs="+",
                              help="Override config options list (like sec/sub/opt=value)")
     args_parser.add_argument("-m", "--dump-config", dest="dump_config", action="store_true",
@@ -205,8 +206,8 @@ def _get_config_scheme() -> Dict:
             },
 
             "info": {
-                "meta":   Option("/etc/kvmd/meta.yaml",    type=valid_abs_path_exists, unpack_as="meta_path"),
-                "extras": Option("/usr/share/kvmd/extras", type=valid_abs_path_exists, unpack_as="extras_path"),
+                "meta":   Option("/etc/kvmd/meta.yaml",    type=valid_abs_file, unpack_as="meta_path"),
+                "extras": Option("/usr/share/kvmd/extras", type=valid_abs_dir, unpack_as="extras_path"),
             },
 
             "hid": {
@@ -303,7 +304,7 @@ def _get_config_scheme() -> Dict:
             },
 
             "auth": {
-                "file": Option("/etc/kvmd/ipmipasswd", type=valid_abs_path_exists, unpack_as="path"),
+                "file": Option("/etc/kvmd/ipmipasswd", type=valid_abs_file, unpack_as="path"),
             },
         },
     }
diff --git a/kvmd/apps/otgmsd/__init__.py b/kvmd/apps/otgmsd/__init__.py
index b22c0dfb..2fe566ba 100644
--- a/kvmd/apps/otgmsd/__init__.py
+++ b/kvmd/apps/otgmsd/__init__.py
@@ -33,7 +33,7 @@ import psutil
 from ...validators.basic import valid_bool
 from ...validators.basic import valid_number
 
-from ...validators.os import valid_abs_path_exists
+from ...validators.os import valid_abs_file
 
 from .. import init
 
@@ -98,7 +98,7 @@ def main(argv: Optional[List[str]]=None) -> None:
                         metavar="<1|0|yes|no>", help="Set CD-ROM flag")
     parser.add_argument("--set-rw", default=None, type=valid_bool,
                         metavar="<1|0|yes|no>", help="Set RW flag")
-    parser.add_argument("--set-image", default=None, type=valid_abs_path_exists,
+    parser.add_argument("--set-image", default=None, type=valid_abs_file,
                         metavar="<path>", help="Set the image file")
     parser.add_argument("--eject", action="store_true",
                         help="Eject the image")
diff --git a/kvmd/plugins/auth/htpasswd.py b/kvmd/plugins/auth/htpasswd.py
index 2f4e2df8..94e29711 100644
--- a/kvmd/plugins/auth/htpasswd.py
+++ b/kvmd/plugins/auth/htpasswd.py
@@ -26,7 +26,7 @@ import passlib.apache
 
 from ...yamlconf import Option
 
-from ...validators.os import valid_abs_path_exists
+from ...validators.os import valid_abs_file
 
 from . import BaseAuthService
 
@@ -39,7 +39,7 @@ class Plugin(BaseAuthService):
     @classmethod
     def get_plugin_options(cls) -> Dict:
         return {
-            "file": Option("/etc/kvmd/htpasswd", type=valid_abs_path_exists, unpack_as="path"),
+            "file": Option("/etc/kvmd/htpasswd", type=valid_abs_file, unpack_as="path"),
         }
 
     async def authorize(self, user: str, passwd: str) -> bool:
diff --git a/kvmd/plugins/msd/otg.py b/kvmd/plugins/msd/otg.py
index ca57a932..839fbc1f 100644
--- a/kvmd/plugins/msd/otg.py
+++ b/kvmd/plugins/msd/otg.py
@@ -29,7 +29,7 @@ from typing import AsyncGenerator
 
 from ...yamlconf import Option
 
-from ...validators.os import valid_abs_path_exists
+from ...validators.os import valid_abs_dir
 from ...validators.os import valid_command
 
 from . import MsdOperationError
@@ -48,7 +48,7 @@ class Plugin(BaseMsd):
     def get_plugin_options(cls) -> Dict:
         sudo = ["/usr/bin/sudo", "--non-interactive"]
         return {
-            "storage":     Option("/var/lib/kvmd/msd", type=valid_abs_path_exists, unpack_as="storage_path"),
+            "storage":     Option("/var/lib/kvmd/msd", type=valid_abs_dir, unpack_as="storage_path"),
             "remount_cmd": Option([*sudo, "/usr/bin/kvmd-helper-otgmsd-remount", "{mode}"], type=valid_command),
             "unlock_cmd":  Option([*sudo, "/usr/bin/kvmd-helper-otgmsd-unlock", "unlock"], type=valid_command),
         }
diff --git a/kvmd/validators/os.py b/kvmd/validators/os.py
index d37ae5d4..3fcca5c1 100644
--- a/kvmd/validators/os.py
+++ b/kvmd/validators/os.py
@@ -21,6 +21,7 @@
 
 
 import os
+import stat
 
 from typing import List
 from typing import Any
@@ -33,22 +34,45 @@ from .basic import valid_string_list
 
 
 # =====
-def valid_abs_path(arg: Any, exists: bool=False, name: str="") -> str:
-    if not name:
-        name = ("existent absolute path" if exists else "absolute path")
+def valid_abs_path(arg: Any, type: str="", name: str="") -> str:  # pylint: disable=redefined-builtin
+    if type:
+        if not name:
+            name = f"absolute path to existent {type}"
+        type = {
+            "file": "reg",
+            "dir": "dir",
+            "link": "lnk",
+            "sock": "sock",
+            "fifo": "fifo",
+            "char": "chr",
+            "block": "blk",
+        }[type]
+    else:
+        if not name:
+            name = "absolute path"
 
     if len(str(arg).strip()) == 0:
         arg = None
-    arg = check_not_none_string(arg, name)
+    arg = os.path.abspath(check_not_none_string(arg, name))
+
+    if type:
+        try:
+            st = os.stat(arg)
+        except Exception as err:
+            raise_error(arg, f"{name}: {err}")
+        else:
+            if not getattr(stat, f"S_IS{type.upper()}")(st.st_mode):
+                raise_error(arg, name)
 
-    arg = os.path.abspath(arg)
-    if exists and not os.access(arg, os.F_OK):
-        raise_error(arg, name)
     return arg
 
 
-def valid_abs_path_exists(arg: Any, name: str="") -> str:
-    return valid_abs_path(arg, exists=True, name=name)
+def valid_abs_file(arg: Any, name: str="") -> str:
+    return valid_abs_path(arg, type="file", name=name)
+
+
+def valid_abs_dir(arg: Any, name: str="") -> str:
+    return valid_abs_path(arg, type="dir", name=name)
 
 
 def valid_printable_filename(arg: Any, name: str="") -> str:
@@ -78,5 +102,5 @@ def valid_command(arg: Any) -> List[str]:
     cmd = valid_string_list(arg, delim=r"[,\t]+", name="command")
     if len(cmd) == 0:
         raise_error(arg, "command")
-    cmd[0] = valid_abs_path_exists(cmd[0], name="command entry point")
+    cmd[0] = valid_abs_file(cmd[0], name="command entry point")
     return cmd