mirror of
https://github.com/mofeng-git/One-KVM.git
synced 2025-12-12 01:00:29 +08:00
strict file validation
This commit is contained in:
parent
372bf2a4af
commit
97e2d9128c
@ -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"),
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
@ -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")
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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),
|
||||
}
|
||||
|
||||
@ -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
|
||||
|
||||
@ -29,7 +29,6 @@ import pytest
|
||||
|
||||
from kvmd.validators import ValidatorError
|
||||
from kvmd.validators.os import valid_abs_path
|
||||
from kvmd.validators.os import valid_abs_path_exists
|
||||
from kvmd.validators.os import valid_printable_filename
|
||||
from kvmd.validators.os import valid_unix_mode
|
||||
from kvmd.validators.os import valid_command
|
||||
@ -63,21 +62,22 @@ def test_fail__valid_abs_path(arg: Any) -> None:
|
||||
("/root", "/root"),
|
||||
(".", os.path.abspath(".")),
|
||||
])
|
||||
def test_ok__valid_abs_path_exists(arg: Any, retval: str) -> None:
|
||||
assert valid_abs_path_exists(arg) == retval
|
||||
def test_ok__valid_abs_path__dir(arg: Any, retval: str) -> None:
|
||||
assert valid_abs_path(arg, type="dir") == retval
|
||||
|
||||
|
||||
@pytest.mark.parametrize("arg", [
|
||||
"/f/o/o/b/a/r",
|
||||
"/etc/passwd",
|
||||
"/etc/passwd/",
|
||||
"~",
|
||||
"/foo~",
|
||||
"/foo/~",
|
||||
"",
|
||||
None,
|
||||
])
|
||||
def test_fail__valid_abs_path_exists(arg: Any) -> None:
|
||||
def test_fail__valid_abs_path__dir(arg: Any) -> None:
|
||||
with pytest.raises(ValidatorError):
|
||||
print(valid_abs_path_exists(arg))
|
||||
print(valid_abs_path(arg, type="dir"))
|
||||
|
||||
|
||||
# =====
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user