107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
SDK_SRC = PROJECT_ROOT / "packages" / "jwebot-sdk" / "src"
|
|
KERNEL_SRC = PROJECT_ROOT / "packages" / "jwebot-kernel" / "src"
|
|
sys.path[:0] = [str(SDK_SRC), str(KERNEL_SRC)]
|
|
|
|
from jwebot_kernel import ( # noqa: E402
|
|
SourcePolicyError,
|
|
parse_manifest,
|
|
validate_source_tree,
|
|
)
|
|
|
|
|
|
def manifest():
|
|
return parse_manifest(
|
|
{
|
|
"id": "test-command",
|
|
"author": "tester",
|
|
"description": "Run a test command",
|
|
"help": ["/test"],
|
|
"version": "1.0.0",
|
|
"api_version": "1",
|
|
"entrypoint": "test_command:plugin",
|
|
"type": "command",
|
|
"capabilities": {
|
|
"commands": ["/test"],
|
|
"storage": False,
|
|
"network_profiles": [],
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
class SourcePolicyTests(unittest.TestCase):
|
|
def validate(self, source: str):
|
|
temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
root = Path(temporary.name)
|
|
source_root = root / "src"
|
|
source_root.mkdir()
|
|
(source_root / "test_command.py").write_text(source, encoding="utf-8")
|
|
return validate_source_tree(root, manifest())
|
|
|
|
def test_sdk_and_pure_stdlib_imports_are_allowed(self):
|
|
paths = self.validate(
|
|
"\n".join(
|
|
(
|
|
"from dataclasses import dataclass",
|
|
"from jwebot_sdk import ReplyPlan",
|
|
"",
|
|
"@dataclass",
|
|
"class Value:",
|
|
" number: int",
|
|
"",
|
|
"def reply():",
|
|
" return ReplyPlan.text(str(Value(1).number))",
|
|
)
|
|
)
|
|
)
|
|
self.assertEqual(len(paths), 1)
|
|
|
|
def test_subprocess_import_is_rejected(self):
|
|
with self.assertRaisesRegex(SourcePolicyError, "JWP001"):
|
|
self.validate("import subprocess\n")
|
|
|
|
def test_socket_import_is_rejected(self):
|
|
with self.assertRaisesRegex(SourcePolicyError, "JWP001"):
|
|
self.validate("import socket\n")
|
|
|
|
def test_open_is_rejected(self):
|
|
with self.assertRaisesRegex(SourcePolicyError, "JWP002"):
|
|
self.validate('open("/tmp/value")\n')
|
|
|
|
def test_dynamic_import_is_rejected(self):
|
|
with self.assertRaisesRegex(SourcePolicyError, "JWP002"):
|
|
self.validate('__import__("os")\n')
|
|
|
|
def test_dunder_attribute_access_is_rejected(self):
|
|
with self.assertRaisesRegex(SourcePolicyError, "JWP003"):
|
|
self.validate("value.__class__\n")
|
|
|
|
def test_relative_local_import_is_allowed(self):
|
|
paths = self.validate("from .helper import value\n")
|
|
self.assertEqual(len(paths), 1)
|
|
|
|
def test_symlinked_source_is_rejected(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source_root = root / "src"
|
|
source_root.mkdir()
|
|
target = root / "outside.py"
|
|
target.write_text("value = 1\n", encoding="utf-8")
|
|
(source_root / "test_command.py").symlink_to(target)
|
|
with self.assertRaisesRegex(SourcePolicyError, "JWP006"):
|
|
validate_source_tree(root, manifest())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|