implement user alias management (#355)

* implement user alias management

* add exception handling for too long alias and for nonexistent user

* update error handling to correctly report on too many aliases, and also to prevent adding an alias that is too long.

* adding cmd 'list-aliases' and moving error handling into py42

* add empty list test for list-aliases

Co-authored-by: Tora Kozic <tora.kozic@code42.com>

Author: ceciliastevens

CHANGELOG.md

@@ -8,6 +8,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 The intended audience of this file is for py42 consumers -- as such, changes that don't affect
 how a consumer would use the library (e.g. adding unit tests, updating documentation, etc) are not captured here.
 
+## Unreleased
+
+### Added
+- `users add-alias`, `users remove-alias`, `users bulk add-alias`, and `users bulk remove-alias` for managing cloud aliases for users.
+
 ## 1.12.1 - 2022-01-21
 
 ### Fixed

setup.py

@@ -39,7 +39,7 @@
         "keyrings.alt==3.2.0",
         "ipython==7.16.3",
         "pandas>=1.1.3",
-        "py42>=1.19.3",
+        "py42>=1.21.1",
     ],
     extras_require={
         "dev": [

src/code42cli/click_ext/groups.py

@@ -7,6 +7,8 @@
 from py42.exceptions import Py42ActiveLegalHoldError
 from py42.exceptions import Py42CaseAlreadyHasEventError
 from py42.exceptions import Py42CaseNameExistsError
+from py42.exceptions import Py42CloudAliasCharacterLimitExceededError
+from py42.exceptions import Py42CloudAliasLimitExceededError
 from py42.exceptions import Py42DescriptionLimitExceededError
 from py42.exceptions import Py42ForbiddenError
 from py42.exceptions import Py42HTTPError
@@ -86,6 +88,8 @@ def invoke(self, ctx):
             Py42TrustedActivityConflictError,
             Py42TrustedActivityInvalidCharacterError,
             Py42TrustedActivityIdNotFound,
+            Py42CloudAliasLimitExceededError,
+            Py42CloudAliasCharacterLimitExceededError,
         ) as err:
             msg = err.args[0]
             self.logger.log_error(msg)

src/code42cli/cmds/users.py

@@ -3,6 +3,7 @@
 import click
 from pandas import DataFrame
 from pandas import json_normalize
+from py42.exceptions import Py42BadRequestError
 from py42.exceptions import Py42NotFoundError
 
 from code42cli.bulk import generate_template_cmd_factory
@@ -215,6 +216,8 @@ def reactivate(state, username):
 
 _bulk_user_roles_headers = ["username", "role_name"]
 
+_bulk_user_alias_headers = ["username", "alias"]
+
 
 @users.command(name="move")
 @username_option("The username of the user to move.", required=True)
@@ -226,6 +229,37 @@ def change_organization(state, username, org_id):
     _change_organization(state.sdk, username, org_id)
 
 
+@users.command()
+@click.argument("username")
+@click.argument("alias")
+@sdk_options()
+def add_alias(state, username, alias):
+    """Add a cloud alias for a given user."""
+    _add_cloud_alias(state.sdk, username, alias)
+
+
+@users.command()
+@click.argument("username")
+@click.argument("alias")
+@sdk_options()
+def remove_alias(state, username, alias):
+    """Remove a cloud alias for a given user."""
+    _remove_cloud_alias(state.sdk, username, alias)
+
+
+@users.command()
+@click.argument("username")
+@sdk_options()
+def list_aliases(state, username):
+    """List the cloud aliases for a given user."""
+    user = _get_user(state.sdk, username)
+    aliases = user["cloudUsernames"]
+    if aliases:
+        click.echo(aliases)
+    else:
+        click.echo(f"No cloud aliases for user '{username}' found.")
+
+
 @users.group(cls=OrderedGroup)
 @sdk_options(hidden=True)
 def orgs(state):
@@ -292,6 +326,8 @@ def bulk(state):
     commands_dict={
         "update": _bulk_user_update_headers,
         "move": _bulk_user_move_headers,
+        "add-alias": _bulk_user_alias_headers,
+        "remove-alias": _bulk_user_alias_headers,
     },
     help_message="Generate the CSV template needed for bulk user commands.",
 )
@@ -539,6 +575,86 @@ def handle_row(**row):
     formatter.echo_formatted_list(result_rows)
 
 
+@bulk.command(
+    name="add-alias",
+    help=f"Add aliases to a list of users from the provided CSV in format: {','.join(_bulk_user_alias_headers)}",
+)
+@read_csv_arg(headers=_bulk_user_alias_headers)
+@format_option
+@sdk_options()
+def bulk_add_alias(state, csv_rows, format):
+    """Bulk add aliases to users"""
+
+    # Initialize the SDK before starting any bulk processes
+    # to prevent multiple instances and having to enter 2fa multiple times.
+    sdk = state.sdk
+    success_header = "alias added"
+
+    csv_rows[0][success_header] = "False"
+    formatter = OutputFormatter(format, {key: key for key in csv_rows[0].keys()})
+    stats = create_worker_stats(len(csv_rows))
+
+    def handle_row(**row):
+        try:
+            _add_cloud_alias(
+                sdk, **{key: row[key] for key in row.keys() if key != success_header}
+            )
+            row[success_header] = "True"
+        except Exception as err:
+            row[success_header] = f"False: {err}"
+            stats.increment_total_errors()
+        return row
+
+    result_rows = run_bulk_process(
+        handle_row,
+        csv_rows,
+        progress_label="Adding aliases to users:",
+        stats=stats,
+        raise_global_error=False,
+    )
+    formatter.echo_formatted_list(result_rows)
+
+
+@bulk.command(
+    name="remove-alias",
+    help=f"Remove aliases from a list of users from the provided CSV in format: {','.join(_bulk_user_alias_headers)}",
+)
+@read_csv_arg(headers=_bulk_user_alias_headers)
+@format_option
+@sdk_options()
+def bulk_remove_alias(state, csv_rows, format):
+    """Bulk remove aliases from users"""
+
+    # Initialize the SDK before starting any bulk processes
+    # to prevent multiple instances and having to enter 2fa multiple times.
+    sdk = state.sdk
+    success_header = "alias removed"
+
+    csv_rows[0][success_header] = "False"
+    formatter = OutputFormatter(format, {key: key for key in csv_rows[0].keys()})
+    stats = create_worker_stats(len(csv_rows))
+
+    def handle_row(**row):
+        try:
+            _remove_cloud_alias(
+                sdk, **{key: row[key] for key in row.keys() if key != success_header}
+            )
+            row[success_header] = "True"
+        except Exception as err:
+            row[success_header] = f"False: {err}"
+            stats.increment_total_errors()
+        return row
+
+    result_rows = run_bulk_process(
+        handle_row,
+        csv_rows,
+        progress_label="Removing aliases from users:",
+        stats=stats,
+        raise_global_error=False,
+    )
+    formatter.echo_formatted_list(result_rows)
+
+
 def _add_user_role(sdk, username, role_name):
     user_id = _get_legacy_user_id(sdk, username)
     _get_role_id(sdk, role_name)  # function provides role name validation
@@ -666,3 +782,21 @@ def _deactivate_user(sdk, username):
 def _reactivate_user(sdk, username):
     user_id = _get_legacy_user_id(sdk, username)
     sdk.users.reactivate(user_id)
+
+
+def _get_user(sdk, username):
+    # use when retrieving the user information from the detectionlists module
+    try:
+        return sdk.detectionlists.get_user(username).data
+    except Py42BadRequestError:
+        raise UserDoesNotExistError(username)
+
+
+def _add_cloud_alias(sdk, username, alias):
+    user = _get_user(sdk, username)
+    sdk.detectionlists.add_user_cloud_alias(user["userId"], alias)
+
+
+def _remove_cloud_alias(sdk, username, alias):
+    user = _get_user(sdk, username)
+    sdk.detectionlists.remove_user_cloud_alias(user["userId"], alias)