tighten metadata and plugin map parsing

Author: zouyonghe

scripts/validate_plugins/detect_changed_plugins.py

@@ -8,20 +8,18 @@
 import sys
 from pathlib import Path
 
+if __package__ in {None, ""}:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from scripts.validate_plugins.plugins_map import load_plugins_map_text
+
 
 DEFAULT_ASTRBOT_REF = "master"
 ASTRBOT_REMOTE_URL = "https://github.com/AstrBotDevs/AstrBot"
 
 
 def load_plugins_map(text: str, *, source_name: str) -> dict[str, dict]:
-    try:
-        data = json.loads(text)
-    except json.JSONDecodeError as exc:
-        raise ValueError(f"plugins.json is invalid on the {source_name}: {exc}") from exc
-
-    if not isinstance(data, dict):
-        raise ValueError("plugins.json must contain a JSON object")
-    return data
+    return load_plugins_map_text(text, source_name=source_name)
 
 
 def detect_changed_plugin_names(*, base: dict[str, dict], head: dict[str, dict]) -> list[str]:

scripts/validate_plugins/plugins_map.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+
+def validate_plugins_map(data: object, *, source_name: str) -> dict[str, dict]:
+    if not isinstance(data, dict):
+        raise ValueError("plugins.json must contain a JSON object")
+
+    for name, payload in data.items():
+        if not isinstance(name, str):
+            raise ValueError(
+                f"plugins.json on the {source_name} has a non-string key: {name!r}"
+            )
+        if not isinstance(payload, dict):
+            raise ValueError(
+                f"plugins.json entry {name!r} on the {source_name} must be a JSON object"
+            )
+
+    return data
+
+
+def load_plugins_map_text(text: str, *, source_name: str) -> dict[str, dict]:
+    try:
+        data = json.loads(text)
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"plugins.json is invalid on the {source_name}: {exc}") from exc
+
+    return validate_plugins_map(data, source_name=source_name)
+
+
+def load_plugins_map_file(path: Path, *, source_name: str) -> dict[str, dict]:
+    return load_plugins_map_text(path.read_text(encoding="utf-8"), source_name=source_name)

scripts/validate_plugins/run.py

@@ -16,6 +16,11 @@
 from pathlib import Path
 from urllib.parse import urlparse
 
+if __package__ in {None, ""}:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from scripts.validate_plugins.plugins_map import load_plugins_map_file
+
 try:
     import yaml
 except ImportError:  # pragma: no cover - optional in local unit tests
@@ -98,6 +103,15 @@ def select_plugins(
 
 
 def _parse_simple_yaml(path: Path) -> dict:
+    """Very small YAML subset parser used as a fallback when PyYAML is unavailable.
+
+    Supported format:
+    - Flat mapping of `key: value` pairs
+    - No indentation (no nested objects or multiline continuations)
+    - No lists (`- item` syntax)
+    - `#` starts a comment when preceded by whitespace (or at line start)
+    """
+
     def parse_value(raw_value: str) -> str:
         value = raw_value.strip()
         if not value:
@@ -112,13 +126,36 @@ def parse_value(raw_value: str) -> str:
         value = re.split(r"\s+#", value, maxsplit=1)[0].rstrip()
         return value.strip("\"'")
 
-    result = {}
-    for raw_line in path.read_text(encoding="utf-8").splitlines():
-        line = raw_line.strip()
-        if not line or line.startswith("#") or ":" not in line:
+    result: dict[str, str] = {}
+    for lineno, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
+        stripped = raw_line.strip()
+        if not stripped:
+            continue
+        if stripped.startswith("#"):
             continue
+        if raw_line[0].isspace():
+            raise ValueError(
+                f"Unsupported YAML indentation in {path} at line {lineno}: {raw_line!r}"
+            )
+
+        line = stripped
+        if line.startswith("-"):
+            raise ValueError(
+                f"Unsupported YAML list syntax in {path} at line {lineno}: {raw_line!r}"
+            )
+        if ":" not in line:
+            raise ValueError(
+                f"Unsupported YAML content (expected 'key: value') in {path} at line {lineno}: {raw_line!r}"
+            )
+
         key, value = line.split(":", 1)
-        result[key.strip()] = parse_value(value)
+        key = key.strip()
+        if not key:
+            raise ValueError(f"Empty key is not allowed in {path} at line {lineno}: {raw_line!r}")
+        if key in result:
+            raise ValueError(f"Duplicate key '{key}' in {path} at line {lineno}")
+
+        result[key] = parse_value(value)
     return result
 
 
@@ -139,7 +176,11 @@ def load_metadata(path: Path) -> dict:
         if not isinstance(loaded, dict):
             raise MetadataLoadError("metadata.yaml must contain a mapping at the top level")
         return loaded
-    return _parse_simple_yaml(path)
+
+    try:
+        return _parse_simple_yaml(path)
+    except ValueError as exc:
+        raise MetadataLoadError(str(exc)) from exc
 
 
 def precheck_plugin_directory(plugin_dir: Path) -> dict:
@@ -232,20 +273,7 @@ def build_report(results: list[dict]) -> dict:
 
 
 def load_plugins_index(path: Path) -> dict[str, dict]:
-    data = json.loads(path.read_text(encoding="utf-8"))
-    if not isinstance(data, dict):
-        raise ValueError("plugins.json must contain a JSON object")
-
-    for key, value in data.items():
-        if not isinstance(key, str):
-            raise ValueError(
-                f"plugins.json keys must be strings, got {type(key).__name__!r}: {key!r}"
-            )
-        if not isinstance(value, dict):
-            raise ValueError(
-                f"plugins.json values must be objects/dicts; key {key!r} has {type(value).__name__!r}"
-            )
-    return data
+    return load_plugins_map_file(path, source_name="plugins.json")
 
 
 def combine_requested_names(

tests/test_detect_changed_plugins.py

@@ -36,6 +36,12 @@ def test_load_plugins_map_returns_dict_for_valid_json(self):
 
         self.assertEqual(plugins, {"plugin-a": {"repo": "https://github.com/example/a"}})
 
+    def test_load_plugins_map_rejects_non_dict_entries(self):
+        module = load_detection_module()
+
+        with self.assertRaisesRegex(ValueError, "plugins.json entry 'plugin-a' on the PR head must be a JSON object"):
+            module.load_plugins_map('{"plugin-a": "bad"}', source_name="PR head")
+
 
 class ChangedPluginDetectionTests(unittest.TestCase):
     def test_detect_changed_plugin_names_returns_only_modified_entries(self):

tests/test_validate_plugins.py

@@ -203,14 +203,11 @@ def test_parse_simple_yaml_handles_comments_quotes_and_whitespace(self):
 
         with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False) as handle:
             handle.write(
-                """
-                # leading comment
-
-                key1: value1      # trailing comment
-                key2: \" spaced value \"
-                key3: 'another value'
-                key4: value-with-#-hash
-                """
+                "# leading comment\n\n"
+                "key1: value1      # trailing comment\n"
+                'key2: " spaced value "\n'
+                "key3: 'another value'\n"
+                "key4: value-with-#-hash\n"
             )
             metadata_path = Path(handle.name)
 
@@ -224,6 +221,45 @@ def test_parse_simple_yaml_handles_comments_quotes_and_whitespace(self):
         self.assertEqual(parsed["key3"], "another value")
         self.assertEqual(parsed["key4"], "value-with-#-hash")
 
+    def test_parse_simple_yaml_rejects_indented_lines(self):
+        module = load_validator_module()
+
+        with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False) as handle:
+            handle.write("name: demo\n  nested: nope\n")
+            metadata_path = Path(handle.name)
+
+        try:
+            with self.assertRaisesRegex(ValueError, "Unsupported YAML indentation"):
+                module._parse_simple_yaml(metadata_path)
+        finally:
+            os.remove(metadata_path)
+
+    def test_parse_simple_yaml_rejects_list_syntax(self):
+        module = load_validator_module()
+
+        with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False) as handle:
+            handle.write("- item\n")
+            metadata_path = Path(handle.name)
+
+        try:
+            with self.assertRaisesRegex(ValueError, "Unsupported YAML list syntax"):
+                module._parse_simple_yaml(metadata_path)
+        finally:
+            os.remove(metadata_path)
+
+    def test_parse_simple_yaml_rejects_duplicate_keys(self):
+        module = load_validator_module()
+
+        with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False) as handle:
+            handle.write("name: first\nname: second\n")
+            metadata_path = Path(handle.name)
+
+        try:
+            with self.assertRaisesRegex(ValueError, "Duplicate key 'name'"):
+                module._parse_simple_yaml(metadata_path)
+        finally:
+            os.remove(metadata_path)
+
     def test_load_metadata_uses_yaml_safe_load_when_available(self):
         module = load_validator_module()
 
@@ -268,12 +304,7 @@ def test_load_metadata_uses_simple_parser_when_yaml_unavailable(self):
         module = load_validator_module()
 
         with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False) as handle:
-            handle.write(
-                """
-                name: demo-plugin
-                version: \"0.2.3\"
-                """
-            )
+            handle.write('name: demo-plugin\nversion: "0.2.3"\n')
             metadata_path = Path(handle.name)
 
         yaml_backup = getattr(module, "yaml", None)
@@ -287,6 +318,22 @@ def test_load_metadata_uses_simple_parser_when_yaml_unavailable(self):
         self.assertEqual(metadata.get("name"), "demo-plugin")
         self.assertEqual(metadata.get("version"), "0.2.3")
 
+    def test_load_metadata_wraps_fallback_parse_errors(self):
+        module = load_validator_module()
+
+        with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False) as handle:
+            handle.write("name: demo\n  nested: nope\n")
+            metadata_path = Path(handle.name)
+
+        yaml_backup = getattr(module, "yaml", None)
+        try:
+            module.yaml = None
+            with self.assertRaisesRegex(module.MetadataLoadError, "Unsupported YAML indentation"):
+                module.load_metadata(metadata_path)
+        finally:
+            module.yaml = yaml_backup
+            os.remove(metadata_path)
+
     def test_load_plugins_index_accepts_valid_object(self):
         module = load_validator_module()
 
@@ -332,7 +379,7 @@ def test_load_plugins_index_rejects_non_dict_values(self):
             index_path = Path(handle.name)
 
         try:
-            with self.assertRaisesRegex(ValueError, "plugins.json values must be objects/dicts"):
+            with self.assertRaisesRegex(ValueError, "plugins.json entry 'not-a-dict'.*must be a JSON object"):
                 module.load_plugins_index(index_path)
         finally:
             os.remove(index_path)