Merge pull request #612 from Ouranosinc/cli-test-register-perms-request

Author: fmigneault

CHANGES.rst

@@ -11,6 +11,8 @@ Changes
 
 Features / Changes
 ~~~~~~~~~~~~~~~~~~~~~
+* Add CLI helper ``batch_update_permissions`` that allows registering one or more `Permission` configuration files
+  against a running `Magpie` instance.
 * Security fix: bump Docker base ``python:3.11-alpine3.19``.
 
 Bug Fixes

magpie/cli/batch_update_permissions.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+"""
+Magpie helper to create or delete a set of permissions.
+
+When parsing permissions to create, any underlying user, group, or intermediate resource
+that are missing, but that can be resolved with reasonable defaults or with an explicit
+definition in the configuration file, will be dynamically created prior to setting the
+corresponding permission on it. All referenced services should exist beforehand. Consider
+using the 'register_providers' utility to register services beforehand as needed.
+
+See https://pavics-magpie.readthedocs.io/en/latest/configuration.html#file-permissions-cfg for more details.
+"""
+import argparse
+from typing import TYPE_CHECKING
+
+from magpie.cli.utils import make_logging_options, setup_logger_from_options
+from magpie.register import magpie_register_permissions_from_config
+from magpie.utils import get_logger
+
+if TYPE_CHECKING:
+    from typing import Any, Optional, Sequence
+
+LOGGER = get_logger(__name__,
+                    message_format="%(asctime)s - %(levelname)s - %(message)s",
+                    datetime_format="%d-%b-%y %H:%M:%S", force_stdout=False)
+
+ERROR_PARAMS = 2
+ERROR_EXEC = 1
+
+
+def make_parser():
+    # type: () -> argparse.ArgumentParser
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("-u", "--url", "--magpie-url", help=(
+        "URL used to access the magpie service (if omitted, will try using 'MAGPIE_URL' environment variable)."
+    ))
+    parser.add_argument("-U", "--username", "--magpie-admin-user", help=(
+        "Admin username for magpie login (if omitted, will try using 'MAGPIE_ADMIN_USER' environment variable)."
+    ))
+    parser.add_argument("-P", "--password", "--magpie-admin-password", help=(
+        "Admin password for magpie login (if omitted, will try using 'MAGPIE_ADMIN_PASSWORD' environment variable)."
+    ))
+    parser.add_argument("-c", "--config", required=True, nargs="+", action="append", help=(
+        "Path to a single configuration file or a directory containing configuration file that contains permissions. "
+        "The option can be specified multiple times to provide multiple lookup directories or specific files to load. "
+        "Configuration files must be in JSON or YAML format, with their respective extensions, or the '.cfg' extension."
+    ))
+    make_logging_options(parser)
+    return parser
+
+
+def main(args=None, parser=None, namespace=None):
+    # type: (Optional[Sequence[str]], Optional[argparse.ArgumentParser], Optional[argparse.Namespace]) -> Any
+    if not parser:
+        parser = make_parser()
+    ns = parser.parse_args(args=args, namespace=namespace)
+    setup_logger_from_options(LOGGER, ns)
+
+    all_configs = [cfg for cfg_args in ns.config for cfg in cfg_args]
+    try:
+        for config in all_configs:
+            LOGGER.info("Processing: [%s]", config)
+            magpie_register_permissions_from_config(config)
+    except Exception as exc:
+        LOGGER.error("Failed permissions parsing and update from specified configurations [%s].", str(exc))
+        return ERROR_EXEC
+    return 0
+
+
+if __name__ == "__main__":
+    main()

magpie/constants.py

@@ -206,7 +206,7 @@ def get_constant(constant_name,             # type: Str
       5. search in environment variables
 
     Parameter :paramref:`constant_name` is expected to have the format ``MAGPIE_[VARIABLE_NAME]`` although any value can
-    be passed to retrieve generic settings from all above mentioned search locations.
+    be passed to retrieve generic settings from all above-mentioned search locations.
 
     If :paramref:`settings_name` is provided as alternative name, it is used as is to search for results if
     :paramref:`constant_name` was not found. Otherwise, ``magpie.[variable_name]`` is used for additional search when

magpie/register.py

@@ -5,7 +5,7 @@
 import subprocess  # nosec
 import time
 from tempfile import NamedTemporaryFile
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, overload
 
 import requests
 import six
@@ -56,18 +56,21 @@
         AnyCookiesType,
         AnyResolvedSettings,
         AnyResponseType,
+        AnySettingsContainer,
         CombinedConfig,
         CookiesOrSessionType,
         GroupsConfig,
         GroupsSettings,
+        Literal,
         MultiConfigs,
         PermissionConfigItem,
         PermissionsConfig,
         ServicesConfig,
         ServicesSettings,
         Str,
         UsersConfig,
-        UsersSettings
+        UsersSettings,
+        WebhooksConfig
     )
 
 
@@ -572,6 +575,36 @@ def _load_config(path_or_dict, section, allow_missing=False):
 CONFIG_KNOWN_EXTENSIONS = frozenset([".cfg", ".json", ".yml", ".yaml"])
 
 
+@overload
+def get_all_configs(path_or_dict, section, allow_missing=False):
+    # type: (Union[Str, CombinedConfig], Literal["groups"], bool) -> GroupsConfig
+    ...
+
+
+@overload
+def get_all_configs(path_or_dict, section, allow_missing=False):
+    # type: (Union[Str, CombinedConfig], Literal["users"], bool) -> UsersConfig
+    ...
+
+
+@overload
+def get_all_configs(path_or_dict, section, allow_missing=False):
+    # type: (Union[Str, CombinedConfig], Literal["permissions"], bool) -> PermissionsConfig
+    ...
+
+
+@overload
+def get_all_configs(path_or_dict, section, allow_missing=False):
+    # type: (Union[Str, CombinedConfig], Literal["services"], bool) -> ServicesConfig
+    ...
+
+
+@overload
+def get_all_configs(path_or_dict, section, allow_missing=False):
+    # type: (Union[Str, CombinedConfig], Literal["webhooks"], bool) -> WebhooksConfig
+    ...
+
+
 def get_all_configs(path_or_dict, section, allow_missing=False):
     # type: (Union[Str, CombinedConfig], Str, bool) -> MultiConfigs
     """
@@ -987,13 +1020,19 @@ def _validate_response(operation, is_create, item_type="Permission"):
         _validate_response(lambda: _apply_session(usr_name, None), is_create=create_perm)
 
 
-def magpie_register_permissions_from_config(permissions_config, magpie_url=None, db_session=None, raise_errors=False):
-    # type: (Union[Str, PermissionsConfig], Optional[Str], Optional[Session], bool) -> None
+def magpie_register_permissions_from_config(
+    permissions_config,     # type: Union[Str, PermissionsConfig]
+    settings=None,          # type: Optional[AnySettingsContainer]
+    db_session=None,        # type: Optional[Session]
+    raise_errors=False,     # type: bool
+):                          # type: (...) -> None
     """
     Applies `permissions` specified in configuration(s) defined as file, directory with files or literal configuration.
 
     :param permissions_config: file/dir path to `permissions` config or JSON/YAML equivalent pre-loaded.
-    :param magpie_url: URL to magpie instance (when using requests; default: `magpie.url` from this app's config).
+    :param settings: Magpie settings to resolve an instance session when using requests instead of DB session.
+        Will look for ``magpie.url``, ``magpie.admin_user`` and ``magpie.admin_password`` by default, or any
+        corresponding environment variable resolution if omitted in the settings.
     :param db_session: db session to use instead of requests to directly create/remove permissions with config.
     :param raise_errors: raises errors related to permissions, instead of just logging the info.
 
@@ -1002,9 +1041,9 @@ def magpie_register_permissions_from_config(permissions_config, magpie_url=None,
     """
     LOGGER.info("Starting permissions processing.")
 
+    magpie_url = None
     if _use_request(db_session):
-        magpie_url = magpie_url or get_magpie_url()
-        settings = {"magpie.url": magpie_url}
+        magpie_url = get_magpie_url(settings)
         LOGGER.debug("Editing permissions using requests to [%s]...", magpie_url)
         err_msg = "Invalid credentials to register Magpie permissions."
         cookies_or_session = get_admin_cookies(settings, raise_message=err_msg)
@@ -1013,19 +1052,36 @@ def magpie_register_permissions_from_config(permissions_config, magpie_url=None,
         cookies_or_session = db_session
 
     LOGGER.debug("Loading configurations.")
-    permissions = get_all_configs(permissions_config, "permissions")  # type: List[PermissionsConfig]
+    if isinstance(permissions_config, list):
+        permissions = [permissions_config]
+    else:
+        permissions = get_all_configs(permissions_config, "permissions")
     perms_cfg_count = len(permissions)
     LOGGER.log(logging.INFO if perms_cfg_count else logging.WARNING,
                "Found %s permissions configurations.", perms_cfg_count)
     users_settings = groups_settings = None
     if perms_cfg_count:
-        users = get_all_configs(permissions_config, "users", allow_missing=True)    # type: List[UsersConfig]
-        groups = get_all_configs(permissions_config, "groups", allow_missing=True)  # type: List[GroupsConfig]
+        if isinstance(permissions_config, str):
+            users = get_all_configs(permissions_config, "users", allow_missing=True)
+        else:
+            users = []
+        if isinstance(permissions_config, str):
+            groups = get_all_configs(permissions_config, "groups", allow_missing=True)
+        else:
+            groups = []
         users_settings = _resolve_config_registry(users, "username") or {}
         groups_settings = _resolve_config_registry(groups, "name") or {}
     for i, perms in enumerate(permissions):
         LOGGER.info("Processing permissions from configuration (%s/%s).", i + 1, perms_cfg_count)
-        _process_permissions(perms, magpie_url, cookies_or_session, users_settings, groups_settings, raise_errors)
+        _process_permissions(
+            perms,
+            magpie_url,
+            cookies_or_session,
+            users_settings,
+            groups_settings,
+            settings,
+            raise_errors,
+        )
     LOGGER.info("All permissions processed.")
 
 
@@ -1054,16 +1110,23 @@ def _resolve_config_registry(config_files, key):
     return config_map
 
 
-def _process_permissions(permissions, magpie_url, cookies_or_session, users=None, groups=None, raise_errors=False):
-    # type: (PermissionsConfig, Str, Session, Optional[UsersSettings], Optional[GroupsSettings], bool) -> None
+def _process_permissions(
+    permissions,            # type: PermissionsConfig
+    magpie_url,             # type: Str
+    cookies_or_session,     # type: Session
+    users=None,             # type: Optional[UsersSettings]
+    groups=None,            # type: Optional[GroupsSettings]
+    settings=None,          # type: Optional[AnySettingsContainer]
+    raise_errors=False,     # type: bool
+):                          # type: (...) -> None
     """
     Processes a single `permissions` configuration.
     """
     if not permissions:
         LOGGER.warning("Permissions configuration are empty.")
         return
 
-    anon_user = get_constant("MAGPIE_ANONYMOUS_USER")
+    anon_user = get_constant("MAGPIE_ANONYMOUS_USER", settings)
     perm_count = len(permissions)
     LOGGER.log(logging.INFO if perm_count else logging.WARNING,
                "Found %s permissions to evaluate from configuration.", perm_count)

setup.cfg

@@ -95,6 +95,8 @@ exclude_lines =
 	LOGGER.error
 	LOGGER.exception
 	LOGGER.log
+	@overload
+	...
 
 [tool:pytest]
 addopts = 

setup.py

@@ -247,6 +247,7 @@ def _extra_requirements(base_requirements, other_requirements):
         "console_scripts": [
             "magpie_cli = magpie.cli:magpie_helper_cli",     # redirect to others below
             "magpie_helper = magpie.cli:magpie_helper_cli",  # alias to helper
+            "magpie_batch_update_permissions = magpie.cli.batch_update_permissions:main",
             "magpie_batch_update_users = magpie.cli.batch_update_users:main",
             "magpie_register_defaults = magpie.cli.register_defaults:main",
             "magpie_register_providers = magpie.cli.register_providers:main",

tests/interfaces.py

@@ -2780,7 +2780,7 @@ def test_GetUserResourcePermissions_MultipleGroupPermissions(self):
             [(effect_perm1_deny, grp1_reason), (effect_perm2_deny, PERMISSION_REASON_DEFAULT)]
         )
 
-        # apply allow user permission on parent service (on level above, not same resource as previous tests)
+        # apply 'allow' user permission on parent service (on level above, not same resource as previous tests)
         # allow user permission takes priority over deny from second group, but only during effective resolution
         # even if second group deny still exists, the user allow permission takes priority as it is more specific
         # during check of local inherited permissions (no recursive considered), second group deny remains the result