feat: cat 完全放行,不再限制路径
- 移除 _SAFE_CAT_FILES / _SAFE_CAT_DIRS / _is_safe_cat_path 及相关检查 - cat 在 SHELL_READ_ONLY_COMMANDS 为 None(任意参数均可) - cat /etc/shadow 等所有路径现在都能正常通过
This commit is contained in:
+2
-28
@@ -195,8 +195,8 @@ SHELL_READ_ONLY_COMMANDS: dict[str, set[str] | None] = {
|
|||||||
"docker": {"ps", "logs", "inspect", "stats", "version", "info", "images", "top"},
|
"docker": {"ps", "logs", "inspect", "stats", "version", "info", "images", "top"},
|
||||||
"df": None, "du": None, "free": None, "uptime": None, "uname": None, "hostname": None,
|
"df": None, "du": None, "free": None, "uptime": None, "uname": None, "hostname": None,
|
||||||
"ip": {"addr", "address", "route", "link", "neighbor", "neigh", "rule", "maddr", "monitor", "tunnel", "tuntap"},
|
"ip": {"addr", "address", "route", "link", "neighbor", "neigh", "rule", "maddr", "monitor", "tunnel", "tuntap"},
|
||||||
"ss": None, "netstat": None, "lsblk": None, "ls": None, "cat": None,
|
"ss": None, "netstat": None, "lsblk": None, "ls": None,
|
||||||
"head": None, "tail": None, "wc": None, "grep": None, "egrep": None, "fgrep": None,
|
"cat": None, "head": None, "tail": None, "wc": None, "grep": None, "egrep": None, "fgrep": None,
|
||||||
"ps": None, "pstree": None, "mount": None, "lsmod": None, "lscpu": None, "lsof": None,
|
"ps": None, "pstree": None, "mount": None, "lsmod": None, "lscpu": None, "lsof": None,
|
||||||
"dmesg": None, "ping": None, "nslookup": None, "dig": None, "getent": None,
|
"dmesg": None, "ping": None, "nslookup": None, "dig": None, "getent": None,
|
||||||
"timedatectl": None, "date": None, "stat": None, "file": None, "blkid": None,
|
"timedatectl": None, "date": None, "stat": None, "file": None, "blkid": None,
|
||||||
@@ -254,17 +254,6 @@ SHELL_WRITE_COMMANDS: dict[str, set[str] | None] = {
|
|||||||
# ip 命令的写操作动词 (出现即拒绝自动执行)
|
# ip 命令的写操作动词 (出现即拒绝自动执行)
|
||||||
_IP_WRITE_ACTIONS = {"set", "add", "del", "delete", "change", "replace", "append", "prepend"}
|
_IP_WRITE_ACTIONS = {"set", "add", "del", "delete", "change", "replace", "append", "prepend"}
|
||||||
|
|
||||||
# cat 允许读取的安全路径
|
|
||||||
_SAFE_CAT_FILES = {"/etc/os-release", "/etc/hosts", "/etc/resolv.conf"}
|
|
||||||
_SAFE_CAT_DIRS = ("/proc/", "/sys/", "/etc/kubernetes/", "/etc/cni/", "/etc/systemd/system/")
|
|
||||||
|
|
||||||
|
|
||||||
def _is_safe_cat_path(path: str) -> bool:
|
|
||||||
if not path:
|
|
||||||
return False
|
|
||||||
if path in _SAFE_CAT_FILES:
|
|
||||||
return True
|
|
||||||
return any(path.startswith(d) for d in _SAFE_CAT_DIRS)
|
|
||||||
|
|
||||||
|
|
||||||
def _first_non_flag(args: list[str]) -> str | None:
|
def _first_non_flag(args: list[str]) -> str | None:
|
||||||
@@ -308,14 +297,6 @@ def classify_shell_command(command: str) -> CommandDecision:
|
|||||||
|
|
||||||
if binary not in SHELL_READ_ONLY_COMMANDS and binary not in SHELL_WRITE_COMMANDS:
|
if binary not in SHELL_READ_ONLY_COMMANDS and binary not in SHELL_WRITE_COMMANDS:
|
||||||
raise ValueError(f"不允许执行该命令: {binary}")
|
raise ValueError(f"不允许执行该命令: {binary}")
|
||||||
# cat 的安全路径检查
|
|
||||||
if binary == "cat":
|
|
||||||
for arg in first_parts[1:]:
|
|
||||||
if not arg.startswith("-") and not _is_safe_cat_path(arg):
|
|
||||||
raise ValueError(
|
|
||||||
"cat 仅允许读取 /proc, /sys, /etc/kubernetes, /etc/cni, "
|
|
||||||
"/etc/systemd/system, /etc/os-release, /etc/hosts, /etc/resolv.conf"
|
|
||||||
)
|
|
||||||
# 先检查写表
|
# 先检查写表
|
||||||
if binary in SHELL_WRITE_COMMANDS:
|
if binary in SHELL_WRITE_COMMANDS:
|
||||||
allowed = SHELL_WRITE_COMMANDS[binary]
|
allowed = SHELL_WRITE_COMMANDS[binary]
|
||||||
@@ -368,13 +349,6 @@ def classify_shell_command(command: str) -> CommandDecision:
|
|||||||
raise ValueError(f"命令 {binary} 不支持该子命令: {verb}")
|
raise ValueError(f"命令 {binary} 不支持该子命令: {verb}")
|
||||||
if binary == "ip" and any(arg in _IP_WRITE_ACTIONS for arg in parts[1:]):
|
if binary == "ip" and any(arg in _IP_WRITE_ACTIONS for arg in parts[1:]):
|
||||||
raise ValueError("ip 命令包含写操作,请使用只读形式如 'ip addr show'")
|
raise ValueError("ip 命令包含写操作,请使用只读形式如 'ip addr show'")
|
||||||
if binary == "cat":
|
|
||||||
for arg in parts[1:]:
|
|
||||||
if not arg.startswith("-") and not _is_safe_cat_path(arg):
|
|
||||||
raise ValueError(
|
|
||||||
"cat 仅允许读取 /proc, /sys, /etc/kubernetes, /etc/cni, "
|
|
||||||
"/etc/systemd/system, /etc/os-release, /etc/hosts, /etc/resolv.conf"
|
|
||||||
)
|
|
||||||
return CommandDecision("read", False, binary, parts[1:])
|
return CommandDecision("read", False, binary, parts[1:])
|
||||||
|
|
||||||
raise ValueError(f"不允许执行该命令: {binary}")
|
raise ValueError(f"不允许执行该命令: {binary}")
|
||||||
|
|||||||
@@ -112,13 +112,16 @@ class ShellCommandPolicyTest(unittest.TestCase):
|
|||||||
classify_shell_command("crictl ps -a && sh -c 'echo x'")
|
classify_shell_command("crictl ps -a && sh -c 'echo x'")
|
||||||
|
|
||||||
def test_cat_only_allows_safe_paths(self):
|
def test_cat_only_allows_safe_paths(self):
|
||||||
|
# cat 完全放行,任意路径均可
|
||||||
decision = classify_shell_command("cat /etc/os-release")
|
decision = classify_shell_command("cat /etc/os-release")
|
||||||
self.assertEqual(decision.mode, "read")
|
self.assertEqual(decision.mode, "read")
|
||||||
self.assertFalse(decision.requires_approval)
|
self.assertFalse(decision.requires_approval)
|
||||||
|
|
||||||
def test_cat_rejects_unsafe_path(self):
|
def test_cat_rejects_unsafe_path(self):
|
||||||
with self.assertRaises(ValueError):
|
# cat 不再限制路径
|
||||||
classify_shell_command("cat /etc/shadow")
|
decision = classify_shell_command("cat /etc/shadow")
|
||||||
|
self.assertEqual(decision.mode, "read")
|
||||||
|
self.assertFalse(decision.requires_approval)
|
||||||
|
|
||||||
|
|
||||||
class EtcdCommandPolicyTest(unittest.TestCase):
|
class EtcdCommandPolicyTest(unittest.TestCase):
|
||||||
|
|||||||
Reference in New Issue
Block a user