AdamWen95
90 lines
4.8 KiB
Python
90 lines
4.8 KiB
Python
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from console.settings import Settings
|
|
from console.native_plans import build_plan, profile_for, validate_options, public_execution
|
|
from console.native_io import SecretFilter
|
|
|
|
|
|
class NativePlanTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name)
|
|
source = self.root / "original"
|
|
source.mkdir()
|
|
(source / "logs").mkdir()
|
|
self.entry = source / "test.py"
|
|
self.entry.write_text("print('original')\n")
|
|
self.settings = Settings(self.root / "runtime", [{"id": "original", "path": str(source)}])
|
|
self.profile = {"id": "test", "source_group": "original", "entry": "test.py", "cwd": str(source),
|
|
"interpreter": sys.executable, "user": "root", "arguments": ["{concurrency}", "{output_file}"],
|
|
"parameters": [{"key": "concurrency", "label": "并发", "type": "integer", "default": 2, "min": 1, "max": 8},
|
|
{"key": "configuration", "type": "select", "default": "known", "options": [{"value": "known", "label": "已登记配置"}]}],
|
|
"impact": {"level": "service", "title": "切换服务", "details": ["可能停止旧服务"], "requires_confirmation": True},
|
|
"environment": {}, "credential_files": [], "interactive": False, "output_paths": [str(source / "logs" / "console-{run_id}*")]}
|
|
self.settings.native_catalog = {"profiles": [self.profile], "targets": []}
|
|
self.script = {"id": "a" * 32, "kind": "external", "source_group": "original", "source_path": str(self.entry),
|
|
"sha256": hashlib.sha256(self.entry.read_bytes()).hexdigest(), "config": {"native_options": {}, "timeout_seconds": 300}, "language": "python"}
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def test_plan_uses_original_file_and_typed_argv_without_shell_join(self):
|
|
before = self.entry.read_bytes()
|
|
plan = build_plan(self.settings, self.script, run_id="b" * 32, check_host=False)
|
|
self.assertEqual(plan["argv"][:2], [sys.executable, str(self.entry)])
|
|
self.assertEqual(plan["argv"][2], "2")
|
|
self.assertIn("console-" + "b" * 32, plan["argv"][3])
|
|
self.assertEqual(plan["cwd"], str(self.entry.parent))
|
|
self.assertEqual(self.entry.read_bytes(), before)
|
|
|
|
def test_unregistered_uploaded_and_replaced_entries_cannot_run_natively(self):
|
|
for change in ({"kind": "managed"}, {"source_path": str(self.root / "unknown.py")}, {"source_group": "other"}):
|
|
self.assertIsNone(profile_for(self.settings, {**self.script, **change}))
|
|
|
|
def test_parameters_reject_injection_unknown_keys_bool_and_out_of_range(self):
|
|
for value in ({"configuration": "../../etc/profile"}, {"configuration": "known;id"},
|
|
{"concurrency": True}, {"concurrency": 9}, {"command": "id"}):
|
|
with self.assertRaises(ValueError):
|
|
validate_options(self.profile, value)
|
|
|
|
def test_confirmation_changes_with_options_source_or_recipe(self):
|
|
first = build_plan(self.settings, self.script, check_host=False)["plan_token"]
|
|
self.script["config"]["native_options"] = {"concurrency": 3}
|
|
second = build_plan(self.settings, self.script, check_host=False)["plan_token"]
|
|
self.assertNotEqual(first, second)
|
|
self.profile["impact"]["details"].append("使用全部 GPU")
|
|
self.assertNotEqual(second, build_plan(self.settings, self.script, check_host=False)["plan_token"])
|
|
self.entry.write_text("print('changed')\n")
|
|
with self.assertRaises(ValueError):
|
|
build_plan(self.settings, self.script, check_host=False)
|
|
|
|
def test_credentials_not_in_public_execution(self):
|
|
self.profile["credential_files"] = [str(self.root / "private.env")]
|
|
visible = public_execution(self.settings, self.script)
|
|
self.assertEqual(visible["mode"], "native")
|
|
self.assertNotIn("credential_files", visible)
|
|
self.assertNotIn("private.env", json.dumps(visible))
|
|
|
|
def test_stream_filter_redacts_secret_split_across_reads(self):
|
|
secret = "dummy-credential-abcdefghijk"
|
|
filt = SecretFilter([secret])
|
|
output = filt.feed(b"hello\nkey=dummy-cred")
|
|
output += filt.feed(b"ential-abcdefghijk\nworld\n")
|
|
output += filt.finish()
|
|
self.assertNotIn(secret, output.decode())
|
|
self.assertIn("[REDACTED]", output.decode())
|
|
self.assertIn("world", output.decode())
|
|
|
|
def test_ansi_split_inside_secret_cannot_reconstruct_plaintext(self):
|
|
secret = "dummy-credential-abcdefghijk"
|
|
filt = SecretFilter([secret])
|
|
result = filt.feed(b"dummy-cred\x1b[")
|
|
result += filt.feed(b"31mential-abcdefghijk\x1b[0m\n")
|
|
result += filt.finish()
|
|
self.assertNotIn(secret.encode(), result)
|
|
self.assertIn(b"[REDACTED]", result)
|