[Downloader] Add schema validation to info.json file processing (#3533)

* schema v1

* set hidden to True for shared libs

* fix test data

* add warning about invalid top-level structure

* don't show full traceback for JSONDecodeError

redbot/cogs/downloader/info_schemas.py

@@ -0,0 +1,230 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Callable, Dict, Tuple, Union, cast
+
+from redbot import VersionInfo, version_info as red_version_info
+
+from . import installable
+from .log import log
+
+if TYPE_CHECKING:
+    from .json_mixins import RepoJSONMixin
+
+
+__all__ = ("REPO_SCHEMA", "INSTALLABLE_SCHEMA", "update_mixin")
+
+
+class UseDefault:
+    """To be used as sentinel."""
+
+
+# sentinel value
+USE_DEFAULT = UseDefault()
+
+
+def ensure_tuple_of_str(
+    info_file: Path, key_name: str, value: Union[Any, UseDefault]
+) -> Tuple[str, ...]:
+    default: Tuple[str, ...] = ()
+    if value is USE_DEFAULT:
+        return default
+    if not isinstance(value, list):
+        log.warning(
+            "Invalid value of '%s' key (expected list, got %s)"
+            " in JSON information file at path: %s",
+            key_name,
+            type(value).__name__,
+            info_file,
+        )
+        return default
+    for item in value:
+        if not isinstance(item, str):
+            log.warning(
+                "Invalid item in '%s' list (expected str, got %s)"
+                " in JSON information file at path: %s",
+                key_name,
+                type(item).__name__,
+                info_file,
+            )
+            return default
+    return tuple(value)
+
+
+def ensure_str(info_file: Path, key_name: str, value: Union[Any, UseDefault]) -> str:
+    default = ""
+    if value is USE_DEFAULT:
+        return default
+    if not isinstance(value, str):
+        log.warning(
+            "Invalid value of '%s' key (expected str, got %s)"
+            " in JSON information file at path: %s",
+            key_name,
+            type(value).__name__,
+            info_file,
+        )
+        return default
+    return value
+
+
+def ensure_red_version_info(
+    info_file: Path, key_name: str, value: Union[Any, UseDefault]
+) -> VersionInfo:
+    default = red_version_info
+    if value is USE_DEFAULT:
+        return default
+    if not isinstance(value, str):
+        log.warning(
+            "Invalid value of '%s' key (expected str, got %s)"
+            " in JSON information file at path: %s",
+            key_name,
+            type(value).__name__,
+            info_file,
+        )
+        return default
+    try:
+        version_info = VersionInfo.from_str(value)
+    except ValueError:
+        log.warning(
+            "Invalid value of '%s' key (given value isn't a valid version string)"
+            " in JSON information file at path: %s",
+            key_name,
+            info_file,
+        )
+        return default
+    return version_info
+
+
+def ensure_python_version_info(
+    info_file: Path, key_name: str, value: Union[Any, UseDefault]
+) -> Tuple[int, int, int]:
+    default = (3, 5, 1)
+    if value is USE_DEFAULT:
+        return default
+    if not isinstance(value, list):
+        log.warning(
+            "Invalid value of '%s' key (expected list, got %s)"
+            " in JSON information file at path: %s",
+            key_name,
+            type(value).__name__,
+            info_file,
+        )
+        return default
+    count = len(value)
+    if count != 3:
+        log.warning(
+            "Invalid value of '%s' key (expected list with 3 items, got %s items)"
+            " in JSON information file at path: %s",
+            key_name,
+            count,
+            info_file,
+        )
+        return default
+    for item in value:
+        if not isinstance(item, int):
+            log.warning(
+                "Invalid item in '%s' list (expected int, got %s)"
+                " in JSON information file at path: %s",
+                key_name,
+                type(item).__name__,
+                info_file,
+            )
+            return default
+    return cast(Tuple[int, int, int], tuple(value))
+
+
+def ensure_bool(
+    info_file: Path, key_name: str, value: Union[Any, UseDefault], *, default: bool = False
+) -> bool:
+    if value is USE_DEFAULT:
+        return default
+    if not isinstance(value, bool):
+        log.warning(
+            "Invalid value of '%s' key (expected bool, got %s)"
+            " in JSON information file at path: %s",
+            key_name,
+            type(value).__name__,
+            info_file,
+        )
+        return default
+    return value
+
+
+def ensure_required_cogs_mapping(
+    info_file: Path, key_name: str, value: Union[Any, UseDefault]
+) -> Dict[str, str]:
+    default: Dict[str, str] = {}
+    if value is USE_DEFAULT:
+        return default
+    if not isinstance(value, dict):
+        log.warning(
+            "Invalid value of '%s' key (expected dict, got %s)"
+            " in JSON information file at path: %s",
+            key_name,
+            type(value).__name__,
+            info_file,
+        )
+        return default
+    # keys in json dicts are always strings
+    for item in value.values():
+        if not isinstance(item, str):
+            log.warning(
+                "Invalid item in '%s' dict (expected str, got %s)"
+                " in JSON information file at path: %s",
+                key_name,
+                type(item).__name__,
+                info_file,
+            )
+            return default
+    return value
+
+
+def ensure_installable_type(
+    info_file: Path, key_name: str, value: Union[Any, UseDefault]
+) -> installable.InstallableType:
+    default = installable.InstallableType.COG
+    if value is USE_DEFAULT:
+        return default
+    if not isinstance(value, str):
+        log.warning(
+            "Invalid value of '%s' key (expected str, got %s)"
+            " in JSON information file at path: %s",
+            key_name,
+            type(value).__name__,
+            info_file,
+        )
+        return default  # NOTE: old behavior was to use InstallableType.UNKNOWN
+    if value in ("", "COG"):
+        return installable.InstallableType.COG
+    if value == "SHARED_LIBRARY":
+        return installable.InstallableType.SHARED_LIBRARY
+    return installable.InstallableType.UNKNOWN
+
+
+EnsureCallable = Callable[[Path, str, Union[Any, UseDefault]], Any]
+SchemaType = Dict[str, EnsureCallable]
+
+REPO_SCHEMA: SchemaType = {
+    "author": ensure_tuple_of_str,
+    "description": ensure_str,
+    "install_msg": ensure_str,
+    "short": ensure_str,
+}
+INSTALLABLE_SCHEMA: SchemaType = {
+    "min_bot_version": ensure_red_version_info,
+    "max_bot_version": ensure_red_version_info,
+    "min_python_version": ensure_python_version_info,
+    "hidden": ensure_bool,
+    "disabled": ensure_bool,
+    "required_cogs": ensure_required_cogs_mapping,
+    "requirements": ensure_tuple_of_str,
+    "tags": ensure_tuple_of_str,
+    "type": ensure_installable_type,
+}
+
+
+def update_mixin(repo_or_installable: RepoJSONMixin, schema: SchemaType) -> None:
+    info = repo_or_installable._info
+    info_file = repo_or_installable._info_file
+    for key, callback in schema.items():
+        setattr(repo_or_installable, key, callback(info_file, key, info.get(key, USE_DEFAULT)))

redbot/cogs/downloader/installable.py

@@ -1,16 +1,16 @@
 from __future__ import annotations
 
-import json
 import functools
 import shutil
 from enum import IntEnum
 from pathlib import Path
-from typing import MutableMapping, Any, TYPE_CHECKING, Optional, Dict, Union, Callable, Tuple, cast
+from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Tuple, Union, cast
 
 from .log import log
+from .info_schemas import INSTALLABLE_SCHEMA, update_mixin
 from .json_mixins import RepoJSONMixin
 
-from redbot.core import __version__, version_info as red_version_info, VersionInfo
+from redbot.core import VersionInfo
 
 if TYPE_CHECKING:
     from .repo_manager import RepoManager, Repo
@@ -41,14 +41,15 @@ class Installable(RepoJSONMixin):
         Repo object of the Installable, if repo is missing this will be `None`
     commit : `str`, optional
         Installable's commit. This is not the same as ``repo.commit``
-    author : `tuple` of `str`, optional
+    author : `tuple` of `str`
         Name(s) of the author(s).
-    bot_version : `tuple` of `int`
+    min_bot_version : `VersionInfo`
-        The minimum bot version required for this installation. Right now
+        The minimum bot version required for this Installable.
-        this is always :code:`3.0.0`.
+    max_bot_version : `VersionInfo`
+        The maximum bot version required for this Installable.
+        Ignored if `min_bot_version` is newer than `max_bot_version`.
     min_python_version : `tuple` of `int`
-        The minimum python version required for this cog. This field will not
+        The minimum python version required for this cog.
-        apply to repo info.json's.
     hidden : `bool`
         Whether or not this cog will be hidden from the user when they use
         `Downloader`'s commands.
@@ -78,29 +79,23 @@ class Installable(RepoJSONMixin):
             Installable's commit. This is not the same as ``repo.commit``
 
         """
-        super().__init__(location)
-
         self._location = location
 
         self.repo = repo
         self.repo_name = self._location.parent.stem
         self.commit = commit
 
-        self.min_bot_version = red_version_info
+        self.min_bot_version: VersionInfo
-        self.max_bot_version = red_version_info
+        self.max_bot_version: VersionInfo
-        self.min_python_version = (3, 5, 1)
+        self.min_python_version: Tuple[int, int, int]
-        self.hidden = False
+        self.hidden: bool
-        self.disabled = False
+        self.disabled: bool
-        self.required_cogs: Dict[str, str] = {}  # Cog name -> repo URL
+        self.required_cogs: Dict[str, str]  # Cog name -> repo URL
-        self.requirements: Tuple[str, ...] = ()
+        self.requirements: Tuple[str, ...]
-        self.tags: Tuple[str, ...] = ()
+        self.tags: Tuple[str, ...]
-        self.type = InstallableType.UNKNOWN
+        self.type: InstallableType
 
-        if self._info_file.exists():
+        super().__init__(location)
-            self._process_info_file(self._info_file)
-
-        if self._info == {}:
-            self.type = InstallableType.COG
 
     def __eq__(self, other: Any) -> bool:
         # noinspection PyProtectedMember
@@ -140,84 +135,9 @@ class Installable(RepoJSONMixin):
     def _read_info_file(self) -> None:
         super()._read_info_file()
 
-        if self._info_file.exists():
+        update_mixin(self, INSTALLABLE_SCHEMA)
-            self._process_info_file()
+        if self.type == InstallableType.SHARED_LIBRARY:
-
-    def _process_info_file(
-        self, info_file_path: Optional[Path] = None
-    ) -> MutableMapping[str, Any]:
-        """
-        Processes an information file. Loads dependencies among other
-        information into this object.
-
-        :type info_file_path:
-        :param info_file_path: Optional path to information file, defaults to `self.__info_file`
-        :return: Raw information dictionary
-        """
-        info_file_path = info_file_path or self._info_file
-        if info_file_path is None or not info_file_path.is_file():
-            raise ValueError("No valid information file path was found.")
-
-        info: Dict[str, Any] = {}
-        with info_file_path.open(encoding="utf-8") as f:
-            try:
-                info = json.load(f)
-            except json.JSONDecodeError:
-                info = {}
-                log.exception("Invalid JSON information file at path: {}".format(info_file_path))
-            else:
-                self._info = info
-
-        try:
-            min_bot_version = VersionInfo.from_str(str(info.get("min_bot_version", __version__)))
-        except ValueError:
-            min_bot_version = self.min_bot_version
-        self.min_bot_version = min_bot_version
-
-        try:
-            max_bot_version = VersionInfo.from_str(str(info.get("max_bot_version", __version__)))
-        except ValueError:
-            max_bot_version = self.max_bot_version
-        self.max_bot_version = max_bot_version
-
-        try:
-            min_python_version = tuple(info.get("min_python_version", (3, 5, 1)))
-        except ValueError:
-            min_python_version = self.min_python_version
-        self.min_python_version = min_python_version
-
-        try:
-            hidden = bool(info.get("hidden", False))
-        except ValueError:
-            hidden = False
-        self.hidden = hidden
-
-        try:
-            disabled = bool(info.get("disabled", False))
-        except ValueError:
-            disabled = False
-        self.disabled = disabled
-
-        self.required_cogs = info.get("required_cogs", {})
-
-        self.requirements = info.get("requirements", ())
-
-        try:
-            tags = tuple(info.get("tags", ()))
-        except ValueError:
-            tags = ()
-        self.tags = tags
-
-        installable_type = info.get("type", "")
-        if installable_type in ("", "COG"):
-            self.type = InstallableType.COG
-        elif installable_type == "SHARED_LIBRARY":
-            self.type = InstallableType.SHARED_LIBRARY
             self.hidden = True
-        else:
-            self.type = InstallableType.UNKNOWN
-
-        return info
 
 
 class InstalledModule(Installable):

redbot/cogs/downloader/json_mixins.py

@@ -1,6 +1,9 @@
 import json
 from pathlib import Path
-from typing import Optional, Tuple, Dict, Any
+from typing import Any, Dict, Tuple
+
+from .info_schemas import REPO_SCHEMA, update_mixin
+from .log import log
 
 
 class RepoJSONMixin:
@@ -9,35 +12,36 @@ class RepoJSONMixin:
     def __init__(self, repo_folder: Path):
         self._repo_folder = repo_folder
 
-        self.author: Tuple[str, ...] = ()
+        self.author: Tuple[str, ...]
-        self.install_msg: Optional[str] = None
+        self.install_msg: str
-        self.short: Optional[str] = None
+        self.short: str
-        self.description: Optional[str] = None
+        self.description: str
 
         self._info_file = repo_folder / self.INFO_FILE_NAME
-        if self._info_file.exists():
+        self._info: Dict[str, Any]
+
         self._read_info_file()
 
-        self._info: Dict[str, Any] = {}
-
     def _read_info_file(self) -> None:
-        if not (self._info_file.exists() or self._info_file.is_file()):
+        if self._info_file.exists():
-            return
-
             try:
                 with self._info_file.open(encoding="utf-8") as f:
                     info = json.load(f)
-        except json.JSONDecodeError:
+            except json.JSONDecodeError as e:
-            return
+                log.error(
+                    "Invalid JSON information file at path: %s\nError: %s", self._info_file, str(e)
+                )
+                info = {}
         else:
+            info = {}
+        if not isinstance(info, dict):
+            log.warning(
+                "Invalid top-level structure (expected dict, got %s)"
+                " in JSON information file at path: %s",
+                type(info).__name__,
+                self._info_file,
+            )
+            info = {}
         self._info = info
 
-        try:
+        update_mixin(self, REPO_SCHEMA)
-            author = tuple(info.get("author", []))
-        except ValueError:
-            author = ()
-        self.author = author
-
-        self.install_msg = info.get("install_msg")
-        self.short = info.get("short")
-        self.description = info.get("description")

redbot/pytest/downloader.py

@@ -88,7 +88,7 @@ INFO_JSON = {
     "hidden": False,
     "install_msg": "A post-installation message",
     "required_cogs": {},
-    "requirements": ("tabulate"),
+    "requirements": ("tabulate",),
     "short": "A short description",
     "tags": ("tag1", "tag2"),
     "type": "COG",
@@ -102,7 +102,7 @@ LIBRARY_INFO_JSON = {
     "hidden": False,  # libraries are always hidden, this tests it will be flipped
     "install_msg": "A library install message",
     "required_cogs": {},
-    "requirements": ("tabulate"),
+    "requirements": ("tabulate",),
     "short": "A short library description",
     "tags": ("libtag1", "libtag2"),
     "type": "SHARED_LIBRARY",