Added guest collection creation

changelog.d/20231127_161505_derek_guest_collection_creation_sc_14410.md

@@ -0,0 +1,11 @@
+
+### Enhancements
+
+* Added a new command for non-admins to create GCSv5 Guest Collections.
+
+    ```
+    globus collection create guest <mapped_collection_id> <root_path> <display_name>
+    ```
+
+* Updated the login-manager's "assert_logins" to verify that non-static dependent scopes
+  are met before executing the command (by evaluating a user's current consents).

src/globus_cli/commands/collection/__init__.py

@@ -4,6 +4,7 @@
 @group(
     "collection",
     lazy_subcommands={
+        "create": (".create", "collection_create"),
         "delete": (".delete", "collection_delete"),
         "list": (".list", "collection_list"),
         "show": (".show", "collection_show"),

src/globus_cli/commands/collection/_common.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import globus_sdk
+
+from globus_cli.termio import Field, formatters
+from globus_cli.types import DATA_CONTAINER_T
+
+
+def filter_fields(check_fields: list[Field], data: DATA_CONTAINER_T) -> list[Field]:
+    return [f for f in check_fields if f.get_value(data) is not None]
+
+
+def standard_collection_fields(auth_client: globus_sdk.AuthClient) -> list[Field]:
+    from globus_cli.services.gcs import ConnectorIdFormatter
+
+    return [
+        Field("Display Name", "display_name"),
+        Field(
+            "Owner",
+            "identity_id",
+            formatter=formatters.auth.IdentityIDFormatter(auth_client),
+        ),
+        Field("ID", "id"),
+        Field("Collection Type", "collection_type"),
+        Field("Mapped Collection ID", "mapped_collection_id"),
+        Field("User Credential ID", "user_credential_id"),
+        Field("Storage Gateway ID", "storage_gateway_id"),
+        Field("Connector", "connector_id", formatter=ConnectorIdFormatter()),
+        Field("Allow Guest Collections", "allow_guest_collections"),
+        Field("Disable Anonymous Writes", "disable_anonymous_writes"),
+        Field("High Assurance", "high_assurance"),
+        Field("Authentication Timeout (Minutes)", "authentication_timeout_mins"),
+        Field("Multi-factor Authentication", "require_mfa"),
+        Field("Manager URL", "manager_url"),
+        Field("HTTPS URL", "https_url"),
+        Field("TLSFTP URL", "tlsftp_url"),
+        Field("Force Encryption", "force_encryption"),
+        Field("Public", "public"),
+        Field("Organization", "organization"),
+        Field("Department", "department"),
+        Field("Keywords", "keywords"),
+        Field("Description", "description"),
+        Field("Contact E-mail", "contact_email"),
+        Field("Contact Info", "contact_info"),
+        Field("Collection Info Link", "info_link"),
+        Field("User Message", "user_message"),
+        Field("User Message Link", "user_message_link"),
+    ]

src/globus_cli/commands/collection/create/__init__.py

@@ -0,0 +1,11 @@
+from globus_cli.parsing import group
+
+
+@group(
+    "create",
+    lazy_subcommands={
+        "guest": (".guest", "collection_create_guest"),
+    },
+)
+def collection_create() -> None:
+    """Create a new Collection"""

src/globus_cli/commands/collection/create/guest.py

@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+import typing as t
+import uuid
+
+import click
+import globus_sdk
+
+from globus_cli.commands.collection._common import (
+    filter_fields,
+    standard_collection_fields,
+)
+from globus_cli.constants import ExplicitNullType
+from globus_cli.endpointish import EntityType
+from globus_cli.login_manager import LoginManager, read_well_known_config
+from globus_cli.parsing import command, endpointish_params, mutex_option_group
+from globus_cli.services.gcs import CustomGCSClient
+from globus_cli.termio import TextMode, display
+
+
+@command("guest", short_help="Create a GCSv5 Guest Collection")
+@click.argument("mapped-collection-id", type=click.UUID)
+@click.argument("collection-base-path", type=str)
+@click.option(
+    "--user-credential-id",
+    type=click.UUID,
+    default=None,
+    help="ID identifying a registered local user to associate with the new collection",
+)
+@click.option(
+    "--local-username",
+    type=str,
+    default=None,
+    help=(
+        "[Alternative to --user-credential-id] Local username to associate with the new"
+        " collection (must match exactly one pre-registered User Credential ID)"
+    ),
+)
+@mutex_option_group("--user-credential-id", "--local-username")
+@endpointish_params.create(name="collection")
+@click.option(
+    "--identity-id",
+    default=None,
+    help="User who should own the collection (defaults to the current user)",
+)
+@click.option(
+    "--public/--private",
+    "public",
+    default=True,
+    help="Set the collection to be public or private",
+)
+@click.option(
+    "--enable-https/--disable-https",
+    "enable_https",
+    default=None,
+    help=(
+        "Explicitly enable or disable  HTTPS support (requires a managed endpoint "
+        "with API v1.1.0)"
+    ),
+)
+@mutex_option_group("--enable-https", "--disable-https")
+@LoginManager.requires_login("auth", "transfer")
+def collection_create_guest(
+    login_manager: LoginManager,
+    *,
+    mapped_collection_id: uuid.UUID,
+    collection_base_path: str,
+    user_credential_id: uuid.UUID | None,
+    local_username: str | None,
+    contact_info: str | None | ExplicitNullType,
+    contact_email: str | None | ExplicitNullType,
+    default_directory: str | None | ExplicitNullType,
+    department: str | None | ExplicitNullType,
+    description: str | None | ExplicitNullType,
+    display_name: str,
+    enable_https: bool | None,
+    force_encryption: bool | None,
+    identity_id: str | None,
+    info_link: str | None | ExplicitNullType,
+    keywords: list[str] | None,
+    public: bool,
+    organization: str | None | ExplicitNullType,
+    user_message: str | None | ExplicitNullType,
+    user_message_link: str | None | ExplicitNullType,
+    verify: dict[str, bool],
+) -> None:
+    """
+    Create a GCSv5 Guest Collection
+    """
+    gcs_client = login_manager.get_gcs_client(
+        collection_id=mapped_collection_id,
+        include_data_access=True,
+        assert_entity_type=(EntityType.GCSV5_MAPPED,),
+    )
+
+    if not user_credential_id:
+        user_credential_id = _select_user_credential_id(
+            gcs_client, mapped_collection_id, local_username, identity_id
+        )
+
+    converted_kwargs: dict[str, t.Any] = ExplicitNullType.nullify_dict(
+        {
+            "collection_base_path": collection_base_path,
+            "contact_info": contact_info,
+            "contact_email": contact_email,
+            "default_directory": default_directory,
+            "department": department,
+            "description": description,
+            "display_name": display_name,
+            "enable_https": enable_https,
+            "force_encryption": force_encryption,
+            "identity_id": identity_id,
+            "info_link": info_link,
+            "keywords": keywords,
+            "mapped_collection_id": mapped_collection_id,
+            "public": public,
+            "organization": organization,
+            "user_credential_id": user_credential_id,
+            "user_message": user_message,
+            "user_message_link": user_message_link,
+        }
+    )
+    converted_kwargs.update(verify)
+
+    res = gcs_client.create_collection(
+        globus_sdk.GuestCollectionDocument(**converted_kwargs)
+    )
+
+    fields = standard_collection_fields(login_manager.get_auth_client())
+    display(res, text_mode=TextMode.text_record, fields=filter_fields(fields, res))
+
+
+def _select_user_credential_id(
+    gcs_client: CustomGCSClient,
+    mapped_collection_id: uuid.UUID,
+    local_username: str | None,
+    identity_id: str | None,
+) -> uuid.UUID:
+    """
+    In the case that the user didn't specify a user credential id, see if we can select
+      one automatically.
+
+    A User Credential is only eligible if it is the only candidate matching the given
+      request parameters (which may be omitted).
+    """
+    mapped_collection = gcs_client.get_collection(mapped_collection_id)
+    storage_gateway_id = mapped_collection["storage_gateway_id"]
+
+    if not identity_id:
+        user_data = read_well_known_config("auth_user_data", allow_null=False)
+        identity_id = user_data["sub"]
+
+    # Grab the list of user credentials which match the endpoint, storage gateway,
+    #   identity id, and local username (if specified)
+    user_creds = [
+        user_cred
+        for user_cred in gcs_client.get_user_credential_list(
+            storage_gateway=storage_gateway_id
+        )
+        if (
+            user_cred["identity_id"] == identity_id
+            and (local_username is None or user_cred.get("username") == local_username)
+        )
+    ]
+
+    if len(user_creds) > 1:
+        # Only instruct them to include --local-username if they didn't already
+        local_username_or = "either --local-username or " if not local_username else ""
+        raise ValueError(
+            "Multiple User Credentials sufficient for Guest Collection Creation. "
+            f"Please try again supplying {local_username_or}--user-credential-id."
+        )
+    if len(user_creds) == 0:
+        endpoint_id = gcs_client.source_epish.get_collection_endpoint_id()
+        raise ValueError(
+            "No User Credentials sufficient for Guest Collection creation.\n\n"
+            "To create a Guest Collection, you must first register at least one User "
+            "Credential on the requisite 'Endpoint':'Storage Gateway' combo "
+            f"('{endpoint_id}':'{storage_gateway_id}').\n"
+            "This can be done with: `globus endpoint user-credential create ...`\n"
+        )
+
+    return uuid.UUID(user_creds[0]["id"])

src/globus_cli/commands/collection/show.py

@@ -3,51 +3,14 @@
 import uuid
 
 import click
-import globus_sdk
 
+from globus_cli.commands.collection._common import (
+    filter_fields,
+    standard_collection_fields,
+)
 from globus_cli.login_manager import LoginManager
 from globus_cli.parsing import collection_id_arg, command
 from globus_cli.termio import Field, TextMode, display, formatters
-from globus_cli.types import DATA_CONTAINER_T
-
-
-def _filter_fields(check_fields: list[Field], data: DATA_CONTAINER_T) -> list[Field]:
-    return [f for f in check_fields if f.get_value(data) is not None]
-
-
-def _get_standard_fields(auth_client: globus_sdk.AuthClient) -> list[Field]:
-    from globus_cli.services.gcs import ConnectorIdFormatter
-
-    return [
-        Field("Display Name", "display_name"),
-        Field(
-            "Owner",
-            "identity_id",
-            formatter=formatters.auth.IdentityIDFormatter(auth_client),
-        ),
-        Field("ID", "id"),
-        Field("Collection Type", "collection_type"),
-        Field("Storage Gateway ID", "storage_gateway_id"),
-        Field("Connector", "connector_id", formatter=ConnectorIdFormatter()),
-        Field("Allow Guest Collections", "allow_guest_collections"),
-        Field("Disable Anonymous Writes", "disable_anonymous_writes"),
-        Field("High Assurance", "high_assurance"),
-        Field("Authentication Timeout", "authentication_timeout_mins"),
-        Field("Multi-factor Authentication", "require_mfa"),
-        Field("Manager URL", "manager_url"),
-        Field("HTTPS URL", "https_url"),
-        Field("TLSFTP URL", "tlsftp_url"),
-        Field("Force Encryption", "force_encryption"),
-        Field("Public", "public"),
-        Field("Organization", "organization"),
-        Field("Department", "department"),
-        Field("Keywords", "keywords"),
-        Field("Description", "description"),
-        Field("Contact E-mail", "contact_email"),
-        Field("Contact Info", "contact_info"),
-        Field("Collection Info Link", "info_link"),
-    ]
-
 
 PRIVATE_FIELDS: list[Field] = [
     Field("Root Path", "root_path"),
@@ -87,7 +50,7 @@ def collection_show(
     gcs_client = login_manager.get_gcs_client(collection_id=collection_id)
 
     query_params = {}
-    fields: list[Field] = _get_standard_fields(login_manager.get_auth_client())
+    fields: list[Field] = standard_collection_fields(login_manager.get_auth_client())
 
     if include_private_policies:
         query_params["include"] = "private_policies"
@@ -97,7 +60,7 @@ def collection_show(
 
     # walk the list of all known fields and reduce the rendering to only look
     # for fields which are actually present
-    real_fields = _filter_fields(fields, res)
+    real_fields = filter_fields(fields, res)
 
     display(
         res,