Add RBAC content guard support

fixes: pulp#352

CHANGES/352.feature

@@ -0,0 +1 @@
+Added commands for CRUD RBAC Content Guards.

pulpcore/cli/core/__init__.py

@@ -4,6 +4,7 @@
 from pulpcore.cli.core.access_policy import access_policy
 from pulpcore.cli.core.artifact import artifact
 from pulpcore.cli.core.content import content
+from pulpcore.cli.core.content_guard import content_guard
 from pulpcore.cli.core.export import export
 from pulpcore.cli.core.exporter import exporter
 from pulpcore.cli.core.group import group
@@ -28,6 +29,7 @@
 main.add_command(export)
 main.add_command(exporter)
 main.add_command(group)
+main.add_command(content_guard)
 main.add_command(importer)
 main.add_command(orphan)
 main.add_command(orphans)  # This one is deprecated

pulpcore/cli/core/content_guard.py

@@ -0,0 +1,109 @@
+import gettext
+from typing import List, Optional
+
+import click
+
+from pulpcore.cli.common.context import (
+    PluginRequirement,
+    PulpContext,
+    pass_entity_context,
+    pass_pulp_context,
+)
+from pulpcore.cli.common.generic import (
+    create_command,
+    destroy_command,
+    href_option,
+    list_command,
+    load_json_callback,
+    name_option,
+    show_command,
+    update_command,
+)
+from pulpcore.cli.core.context import PulpContentGuardContext, PulpRbacContentGuardContext
+
+_ = gettext.gettext
+
+
+@click.group()
+@pass_pulp_context
+@click.pass_context
+def content_guard(ctx: click.Context, pulp_ctx: PulpContext) -> None:
+    ctx.obj = PulpContentGuardContext(pulp_ctx)
+
+
+create_options = [click.option("--name", required=True), click.option("--description")]
+filter_options = [click.option("--name")]
+lookup_options = [name_option, href_option]
+
+content_guard.add_command(list_command(decorators=filter_options))
+
+
+@content_guard.group()
+@pass_pulp_context
+@click.pass_context
+def rbac(ctx: click.Context, pulp_ctx: PulpContext) -> None:
+    pulp_ctx.needs_plugin(PluginRequirement("core", "3.15.0.dev"))
+    ctx.obj = PulpRbacContentGuardContext(pulp_ctx)
+
+
+rbac.add_command(list_command(decorators=filter_options))
+rbac.add_command(create_command(decorators=create_options))
+rbac.add_command(show_command(decorators=lookup_options))
+rbac.add_command(update_command(decorators=lookup_options))
+rbac.add_command(destroy_command(decorators=lookup_options))
+
+
+@rbac.command()
+@name_option
+@href_option
+@click.option(
+    "--group",
+    "groups",
+    help=_("Groups to remove download permission from. Can be specified multiple times."),
+    multiple=True,
+)
+@click.option(
+    "--user",
+    "users",
+    help=_("Users to remove download permission from. Can be specified multiple times."),
+    multiple=True,
+)
+@pass_entity_context
+@pass_pulp_context
+def assign(
+    pulp_ctx: PulpContext,
+    guard_ctx: PulpRbacContentGuardContext,
+    users: Optional[List[str]],
+    groups: Optional[List[str]],
+) -> None:
+    href = guard_ctx.entity["pulp_href"]
+    result = guard_ctx.assign(href=href, users=users, groups=groups)
+    pulp_ctx.output_result(result)
+
+
+@rbac.command()
+@name_option
+@href_option
+@click.option(
+    "--group",
+    "groups",
+    help=_("Groups to remove download permission from. Can be specified multiple times."),
+    multiple=True,
+)
+@click.option(
+    "--user",
+    "users",
+    help=_("Users to remove download permission from. Can be specified multiple times."),
+    multiple=True,
+)
+@pass_entity_context
+@pass_pulp_context
+def remove(
+    pulp_ctx: PulpContext,
+    guard_ctx: PulpRbacContentGuardContext,
+    users: Optional[List[str]],
+    groups: Optional[List[str]],
+) -> None:
+    href = guard_ctx.entity["pulp_href"]
+    result = guard_ctx.remove(href=href, users=users, groups=groups)
+    pulp_ctx.output_result(result)

pulpcore/cli/core/context.py

@@ -283,6 +283,13 @@ def scope(self) -> Dict[str, Any]:
         return {PulpGroupContext.HREF: self.group_ctx.pulp_href}
 
 
+class PulpContentGuardContext(PulpEntityContext):
+    ENTITY = "content guard"
+    ENTITIES = "content guards"
+    HREF_PATTERN = r"^/pulp/api/v3/contentguards/(?P<plugin>\w+)/(?P<resource_type>\w+)/"
+    LIST_ID = "contentguards_list"
+
+
 class PulpImporterContext(PulpEntityContext):
     ENTITY = "PulpImporter"
     HREF = "pulp_importer_href"
@@ -293,6 +300,27 @@ class PulpImporterContext(PulpEntityContext):
     LIST_ID = "importers_core_pulp_list"
 
 
+class PulpRbacContentGuardContext(PulpContentGuardContext):
+    ENTITY = "RBAC content guard"
+    ENTITIES = "RBAC content guards"
+    HREF = "r_b_a_c_content_guard_href"
+    LIST_ID = "contentguards_core_rbac_list"
+    CREATE_ID = "contentguards_core_rbac_create"
+    UPDATE_ID = "contentguards_core_rbac_partial_update"
+    DELETE_ID = "contentguards_core_rbac_delete"
+    READ_ID = "contentguards_core_rbac_read"
+    ASSIGN_ID: ClassVar[str] = "contentguards_core_rbac_assign_permission"
+    REMOVE_ID: ClassVar[str] = "contentguards_core_rbac_remove_permission"
+
+    def assign(self, href: str, users: Optional[List[str]], groups: Optional[List[str]]) -> Any:
+        body = self.preprocess_body({"usernames": users, "groupnames": groups})
+        return self.pulp_ctx.call(self.ASSIGN_ID, parameters={self.HREF: href}, body=body)
+
+    def remove(self, href: str, users: Optional[List[str]], groups: Optional[List[str]]) -> Any:
+        body = self.preprocess_body({"usernames": users, "groupnames": groups})
+        return self.pulp_ctx.call(self.REMOVE_ID, parameters={self.HREF: href}, body=body)
+
+
 class PulpSigningServiceContext(PulpEntityContext):
     ENTITY = "signing service"
     ENTITIES = "signing services"

tests/scripts/pulpcore/test_content_guards.sh

@@ -0,0 +1,26 @@
+#!/bin/bash
+
+# shellcheck source=tests/scripts/config.source
+. "$(dirname "$(dirname "$(realpath "$0")")")"/config.source
+
+pulp debug has-plugin --name "core" --min-version "3.15.1" || exit 3
+
+cleanup() {
+  pulp content-guard rbac destroy --name "cli_test_guard" || true
+  pulp group destroy --name "cli_test_group" || true
+}
+trap cleanup EXIT
+
+expect_succ pulp content-guard rbac create --name "cli_test_guard"
+expect_succ pulp content-guard list
+test "$(echo "$OUTPUT" | jq -r length)" -gt "0"
+expect_succ pulp content-guard rbac list --name "cli_test_guard"
+test "$(echo "$OUTPUT" | jq -r length)" -eq "1"
+expect_succ pulp content-guard rbac show --name "cli_test_guard"
+
+expect_succ pulp group create --name "cli_test_group"
+expect_succ pulp content-guard rbac assign --name "cli_test_guard" --group "cli_test_group"
+test "$(echo "$OUTPUT" | jq -r '.groups' | jq -r length)" -eq "1"
+expect_succ pulp content-guard rbac remove --name "cli_test_guard" --user "admin" --group "cli_test_group"
+test "$(echo "$OUTPUT" | jq -r '.users' | jq -r length)" -eq "0"
+test "$(echo "$OUTPUT" | jq -r '.groups' | jq -r length)" -eq "0"