230 lines
9.5 KiB
Python
230 lines
9.5 KiB
Python
"""Coverage batch v2: host.py shell scanner + classify branches.
|
|
|
|
Baseline left 48 lines of _scan_shell_command uncovered (NUL,
|
|
dquote escapes/backtick, bare $, brace, #-in-word, [-glob, !,
|
|
&> / &>>, pending-redirect errors on && / | / ;, >> / >| / <)
|
|
plus three classify_verify_command lines (cwd missing, cwd escape
|
|
via symlink, script path unvalidated without cwd). Each case below
|
|
asserts the real segments/features/denial reason, not just that the
|
|
scanner did not crash.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
import host
|
|
from host import _scan_shell_command, classify_verify_command
|
|
|
|
|
|
def _scan(cmd):
|
|
return _scan_shell_command(cmd)
|
|
|
|
|
|
def _denied(cmd):
|
|
segments, features, denied = _scan(cmd)
|
|
return denied
|
|
|
|
|
|
class ScanDenialsTestCase(unittest.TestCase):
|
|
def test_nul_and_newlines(self):
|
|
self.assertEqual(_denied("echo a\x00b"), "命令含 NUL 字节")
|
|
self.assertIn("换行", _denied("echo a\nb"))
|
|
self.assertIn("换行", _denied("echo a\rb"))
|
|
|
|
def test_unclosed_and_command_substitution(self):
|
|
self.assertEqual(_denied("echo 'abc"), "引号未闭合")
|
|
self.assertEqual(_denied('echo "abc'), "引号未闭合")
|
|
self.assertIn("$(...)", _denied("echo $(whoami)"))
|
|
self.assertIn("反引号", _denied("echo `whoami`"))
|
|
self.assertIn("反引号", _denied('echo "a`whoami`"'))
|
|
self.assertIn("${ 未闭合", _denied("echo ${FOO"))
|
|
self.assertIn("子 shell", _denied("echo (foo)"))
|
|
self.assertIn("花括号", _denied("echo {a,b}"))
|
|
self.assertIn("花括号", _denied("echo }"))
|
|
self.assertIn("注释", _denied("# echo hi"))
|
|
self.assertIn("取反", _denied("! echo hi"))
|
|
self.assertIn("后台", _denied("pytest &"))
|
|
self.assertIn("here-doc", _denied("cat << EOF"))
|
|
|
|
def test_pending_redirect_errors(self):
|
|
self.assertEqual(_denied("pytest >"), "重定向后缺目标")
|
|
self.assertEqual(_denied("pytest > && pytest"), "重定向后缺目标")
|
|
self.assertEqual(_denied("pytest > | pytest"), "重定向后缺目标")
|
|
self.assertEqual(_denied("pytest > ; pytest"), "重定向后缺目标")
|
|
self.assertEqual(_denied("pytest > || pytest"), "重定向后缺目标")
|
|
|
|
|
|
class ScanWordsTestCase(unittest.TestCase):
|
|
def test_squote_is_literal(self):
|
|
segments, features, denied = _scan("echo 'a$b`c'")
|
|
self.assertIsNone(denied)
|
|
self.assertEqual(segments[0]["words"][1]["text"], "a$b`c")
|
|
self.assertFalse(segments[0]["words"][1]["expand"])
|
|
self.assertNotIn("env", features)
|
|
|
|
def test_dquote_escape_and_dollar(self):
|
|
segments, features, denied = _scan('echo "a\\$b"')
|
|
self.assertIsNone(denied)
|
|
self.assertEqual(segments[0]["words"][1]["text"], "a$b")
|
|
segments, features, denied = _scan('echo "$FOO"')
|
|
self.assertIsNone(denied)
|
|
self.assertIn("env", features)
|
|
self.assertTrue(segments[0]["words"][1]["expand"])
|
|
segments, features, denied = _scan('echo "${FOO}"')
|
|
self.assertIsNone(denied)
|
|
self.assertEqual(segments[0]["words"][1]["text"], "${FOO}")
|
|
self.assertIn("env", features)
|
|
|
|
def test_normal_backslash_escape(self):
|
|
segments, _features, denied = _scan("echo a\\ b")
|
|
self.assertIsNone(denied)
|
|
self.assertEqual([w["text"] for w in segments[0]["words"]],
|
|
["echo", "a b"])
|
|
|
|
def test_bare_dollar_is_literal(self):
|
|
segments, _features, denied = _scan("echo $%")
|
|
self.assertIsNone(denied)
|
|
self.assertEqual(segments[0]["words"][1]["text"], "$%")
|
|
segments, _features, denied = _scan("echo $")
|
|
self.assertIsNone(denied)
|
|
self.assertEqual(segments[0]["words"][1]["text"], "$")
|
|
|
|
def test_hash_inside_word_is_literal(self):
|
|
segments, _features, denied = _scan("echo foo#bar")
|
|
self.assertIsNone(denied)
|
|
self.assertEqual(segments[0]["words"][1]["text"], "foo#bar")
|
|
|
|
def test_bracket_glob_vs_test_command(self):
|
|
segments, features, denied = _scan("echo foo[ab]")
|
|
self.assertIsNone(denied)
|
|
self.assertIn("glob", features)
|
|
self.assertTrue(segments[0]["words"][1]["glob"])
|
|
self.assertEqual(segments[0]["words"][1]["text"], "foo[ab]")
|
|
segments, features, denied = _scan("[ -f x ]")
|
|
self.assertIsNone(denied)
|
|
self.assertNotIn("glob", features)
|
|
self.assertEqual(segments[0]["words"][0]["text"], "[")
|
|
|
|
def test_bang_inside_word_is_literal(self):
|
|
segments, _features, denied = _scan("echo foo!bar")
|
|
self.assertIsNone(denied)
|
|
self.assertEqual(segments[0]["words"][1]["text"], "foo!bar")
|
|
|
|
|
|
class ScanOperatorsTestCase(unittest.TestCase):
|
|
def test_dollar_error_inside_dquote(self):
|
|
self.assertIn("$(...)", _denied('echo "$(whoami)"'))
|
|
|
|
def test_plain_and_pipe_and_semicolon(self):
|
|
segments, features, denied = _scan("pytest && pytest")
|
|
self.assertIsNone(denied)
|
|
self.assertIn("compound", features)
|
|
self.assertEqual(len(segments), 2)
|
|
segments, features, denied = _scan("pytest | pytest")
|
|
self.assertIsNone(denied)
|
|
self.assertIn("pipe", features)
|
|
self.assertEqual(len(segments), 2)
|
|
segments, features, denied = _scan("pytest ; pytest")
|
|
self.assertIsNone(denied)
|
|
self.assertIn("compound", features)
|
|
self.assertEqual(len(segments), 2)
|
|
|
|
def test_compound_or(self):
|
|
segments, features, denied = _scan("pytest || pytest")
|
|
self.assertIsNone(denied)
|
|
self.assertIn("compound", features)
|
|
self.assertEqual(len(segments), 2)
|
|
|
|
def test_and_redirect_and_append_and_clobber(self):
|
|
segments, features, denied = _scan("echo hi &> out.txt")
|
|
self.assertIsNone(denied)
|
|
self.assertIn("redirect", features)
|
|
self.assertEqual(segments[0]["redirects"][0]["target"], "out.txt")
|
|
self.assertFalse(segments[0]["redirects"][0]["fd"])
|
|
segments, features, denied = _scan("echo hi &>> out.txt")
|
|
self.assertIsNone(denied)
|
|
self.assertIn("redirect", features)
|
|
self.assertEqual(segments[0]["redirects"][0]["target"], "out.txt")
|
|
segments, features, denied = _scan("pytest >> out.txt")
|
|
self.assertIsNone(denied)
|
|
self.assertEqual(segments[0]["redirects"][0]["target"], "out.txt")
|
|
segments, features, denied = _scan("pytest >| out.txt")
|
|
self.assertIsNone(denied)
|
|
self.assertEqual(segments[0]["redirects"][0]["target"], "out.txt")
|
|
segments, features, denied = _scan("cat < in.txt")
|
|
self.assertIsNone(denied)
|
|
self.assertEqual(segments[0]["redirects"][0]["target"], "in.txt")
|
|
self.assertFalse(segments[0]["redirects"][0]["fd"])
|
|
|
|
def test_fd_redirect(self):
|
|
segments, features, denied = _scan("pytest 2>&1")
|
|
self.assertIsNone(denied)
|
|
self.assertIn("redirect", features)
|
|
self.assertEqual(segments[0]["redirects"][0]["target"], "1")
|
|
self.assertTrue(segments[0]["redirects"][0]["fd"])
|
|
segments, features, denied = _scan("pytest 2> err.txt")
|
|
self.assertIsNone(denied)
|
|
self.assertEqual(segments[0]["redirects"][0]["target"], "err.txt")
|
|
self.assertFalse(segments[0]["redirects"][0]["fd"])
|
|
|
|
def test_tilde_and_glob_features(self):
|
|
_segments, features, denied = _scan("echo ~/x")
|
|
self.assertIsNone(denied)
|
|
self.assertIn("env", features)
|
|
_segments, features, denied = _scan("echo *.py")
|
|
self.assertIsNone(denied)
|
|
self.assertIn("glob", features)
|
|
|
|
|
|
class ClassifyCwdTestCase(unittest.TestCase):
|
|
def test_script_path_unvalidated_without_cwd(self):
|
|
result = classify_verify_command("./run.sh", cwd=None)
|
|
self.assertTrue(any("未校验" in note for note in result["notes"]),
|
|
result)
|
|
|
|
def test_script_path_cwd_missing(self):
|
|
result = classify_verify_command("./run.sh",
|
|
cwd="/nonexistent-xyz-abc")
|
|
self.assertEqual(result["kind"], "denied")
|
|
self.assertIn("cwd 不存在", result["reason"])
|
|
|
|
def test_script_path_escapes_cwd_via_symlink(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
outside = os.path.join(tmp, "outside")
|
|
os.mkdir(outside)
|
|
target = os.path.join(outside, "x.sh")
|
|
Path(target).write_text("#!/bin/sh\n")
|
|
os.chmod(target, 0o755)
|
|
work = os.path.join(tmp, "work")
|
|
os.mkdir(work)
|
|
os.symlink(outside, os.path.join(work, "link"))
|
|
result = classify_verify_command("link/x.sh", cwd=work)
|
|
self.assertEqual(result["kind"], "denied")
|
|
self.assertIn("越出 cwd", result["reason"])
|
|
|
|
def test_script_path_inside_cwd_executable_passes(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
script = os.path.join(tmp, "run.sh")
|
|
Path(script).write_text("#!/bin/sh\n")
|
|
os.chmod(script, 0o755)
|
|
result = classify_verify_command("./run.sh", cwd=tmp)
|
|
self.assertNotEqual(result["kind"], "denied")
|
|
|
|
def test_absolute_and_dotdot_paths_denied(self):
|
|
self.assertEqual(classify_verify_command("/bin/pytest")["kind"],
|
|
"denied")
|
|
self.assertIn("绝对路径",
|
|
classify_verify_command("/bin/pytest")["reason"])
|
|
self.assertEqual(classify_verify_command("../run.sh")["kind"],
|
|
"denied")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|