fix(cssc): 31250186 address partial comments in public repo PR 20250415

src/acrcssc/README.rst

@@ -15,6 +15,7 @@ Continuous Patching is currently in preview. The following limitations apply:
 - Windows-based container images aren’t supported
 - Only "OS-level" vulnerabilities will be patched. This includes packages in the image managed by a package manager such as “apt” and “yum”. Vulnerabilities at the “application level” are unable to be patched, such as compiled languages like Go, Python, NodeJS
 - Patching is only supported in Public regions, not in Sovereign regions
+- CSSC patching is not supported for registries or in regions where Tasks are unavailable.
 
 Features
 ========

src/acrcssc/azext_acrcssc/_params.py

@@ -15,7 +15,7 @@ def load_arguments(self: AzCommandsLoader, _):
     from .helper._workflow_status import WorkflowTaskState
 
     with self.argument_context("acr supply-chain workflow") as c:
-        c.argument('resource_group', options_list=['--resource-group', '-g'], help='Name of resource group.You can configure the default group using `az configure --defaults group=<name>`', completer=get_resource_name_completion_list(REGISTRY_RESOURCE_TYPE), configured_default='acr', validator=validate_registry_name)
+        c.argument('resource_group', options_list=['--resource-group', '-g'], help='Name of resource group. You can configure the default group using `az configure --defaults group=<name>`', completer=get_resource_name_completion_list(REGISTRY_RESOURCE_TYPE))
         c.argument('registry_name', options_list=['--registry', '-r'], help='The name of the container registry. It should be specified in lower case. You can configure the default registry name using `az configure --defaults acr=<registry name>`', completer=get_resource_name_completion_list(REGISTRY_RESOURCE_TYPE), configured_default='acr', validator=validate_registry_name)
         c.argument("workflow_type", arg_type=get_enum_type(CSSCTaskTypes), options_list=['--type', '-t'], help="Type of workflow task.", required=True)
 

src/acrcssc/azext_acrcssc/_validators.py

@@ -75,28 +75,29 @@ def _validate_continuouspatch_config(config):
                 raise InvalidArgumentValueError("Configuration error: Tag '*' is not allowed with other tags in the same repository. Use '*' as the only tag in the repository to avoid overlaps.")
 
 
-# to save on API calls, we the list of tasks found in the registry
+# to save on API calls, we use task_list to return a list of CSSC tasks found in the registry
 def check_continuous_task_exists(cmd, registry):
     task_list = []
     missing_tasks = []
-    try:
-        acrtask_client = cf_acr_tasks(cmd.cli_ctx)
-        for task_name in CONTINUOUSPATCH_ALL_TASK_NAMES:
+
+    acrtask_client = cf_acr_tasks(cmd.cli_ctx)
+    for task_name in CONTINUOUSPATCH_ALL_TASK_NAMES:
+        try:
             task = get_task(cmd, registry, task_name, acrtask_client)
             if task is None:
                 missing_tasks.append(task_name)
             else:
                 task_list.append(task)
+        except Exception as exception:
+            logger.debug(f"Failed to find tasks from registry {registry.name} : {exception}")
+            missing_tasks.append(task_name)
 
-        if len(missing_tasks) > 0:
-            logger.debug(f"Failed to find tasks {', '.join(missing_tasks)} from registry {registry.name}")
-            return False, task_list
-
-        return True, task_list
-    except Exception as exception:
-        logger.debug(f"Failed to find tasks from registry {registry.name} : {exception}")
+    if len(missing_tasks) > 0:
+        logger.debug(f"Failed to find tasks {', '.join(missing_tasks)} from registry {registry.name}")
         return False, task_list
 
+    return True, task_list
+
 
 def check_continuous_task_config_exists(cmd, registry):
     # A client cannot be used in this situation because the 'show registry/image'
@@ -136,7 +137,7 @@ def _validate_schedule(schedule):
 
 def validate_inputs(schedule, config_file_path=None, dryrun=False, run_immediately=False):
     _validate_schedule(schedule)
-    if config_file_path is not None:
+    if config_file_path:
         validate_continuouspatch_config_v1(config_file_path)
     validate_run_type(dryrun, run_immediately)
 

src/acrcssc/azext_acrcssc/azext_metadata.json

@@ -1,4 +1,4 @@
 {
     "azext.isPreview": true,
-    "azext.minCliCoreVersion": "2.15.0"
+    "azext.minCliCoreVersion": "2.60.0"
 }

src/acrcssc/azext_acrcssc/helper/_constants.py

@@ -11,8 +11,6 @@
 class CSSCTaskTypes(Enum):
     """Enum for the task type."""
     ContinuousPatchV1 = 'continuouspatchv1'
-    # CopaV1 = "CopaV1"
-    # TrivyV1 = "TrivyV1"
 
 
 class TaskRunStatus(Enum):
@@ -36,6 +34,7 @@ class TaskRunStatus(Enum):
 RESOURCE_GROUP = "resource_group"
 SUBSCRIPTION = "subscription"
 TMP_DRY_RUN_FILE_NAME = "tmp_dry_run_template.yaml"
+REGISTRY_BASIC_SKU = "basic"
 
 
 # Continuous Patch Constants

src/acrcssc/azext_acrcssc/helper/_deployment.py

@@ -44,13 +44,12 @@ def validate_and_deploy_template(cmd_ctx,
         mode=DeploymentMode.incremental)
     try:
         validate_template(cmd_ctx, resource_group, deployment_name, template)
-        if (dryrun):
+        if dryrun:
             logger.debug("Dry run, skipping deployment")
             return None
 
         return deploy_template(cmd_ctx, resource_group, deployment_name, template)
     except Exception as exception:
-        logger.debug(f'Failed to validate and deploy template: {exception}')
         raise AzCLIError(f'Failed to validate and deploy template: {exception}')
 
 
@@ -61,10 +60,7 @@ def validate_template(cmd_ctx, resource_group, deployment_name, template):
     api_clients = cf_resources(cmd_ctx)
     validation_res = None
     deployment = Deployment(
-        properties=template,
-        # tags = { "test": CSSC_TAGS }, #we need to know if tagging
-        # is something that will help ust, tasks are proxy resources,
-        # so not sure how that would work
+        properties=template
     )
 
     for validation_attempt in range(2):
@@ -80,7 +76,8 @@ def validate_template(cmd_ctx, resource_group, deployment_name, template):
                 cmd_ctx, "Validating ARM template..."
             )(validation)
             break
-        except Exception:  # pylint: disable=broad-except
+        except Exception as exception:  # pylint: disable=broad-except
+            logger.debug(f"Validation attempt {validation_attempt + 1} failed for template {template}, exception: {exception}")
             if validation_attempt == 1:
                 raise
 
@@ -138,7 +135,7 @@ def deploy_template(cmd_ctx, resource_group, deployment_name, template):
     logger.debug(f"Deployed: {deployment.name} {deployment.id} {depl_props}")
 
     if depl_props.provisioning_state != "Succeeded":
-        logger.debug(f"Failed to provision: {depl_props}")
+        logger.error(f"Failed to provision: {depl_props}")
         raise RuntimeError(
             "Deploy of template to resource group"
             f" {resource_group} proceeded but the provisioning"

src/acrcssc/azext_acrcssc/helper/_ociartifactoperations.py

@@ -58,6 +58,7 @@ def create_oci_artifact_continuous_patch(registry, cssc_config_file, dryrun):
             else:
                 oci_target_name = f"{CSSC_WORKFLOW_POLICY_REPOSITORY}/{CONTINUOUSPATCH_OCI_ARTIFACT_CONFIG}:{CONTINUOUSPATCH_OCI_ARTIFACT_CONFIG_TAG_V1}"
 
+            logger.debug(f"Publish OCI artifact to: {oci_target_name}")
             oras_client.push(
                 target=oci_target_name,
                 files=[temp_artifact_name])
@@ -72,21 +73,28 @@ def get_oci_artifact_continuous_patch(cmd, registry):
     logger.debug("Entering get_oci_artifact_continuous_patch with parameter: %s", registry.login_server)
     config = None
     file_name = None
+    oras_client = None
     try:
         oras_client = _oras_client(registry)
 
         oci_target_name = f"{CSSC_WORKFLOW_POLICY_REPOSITORY}/{CONTINUOUSPATCH_OCI_ARTIFACT_CONFIG}:{CONTINUOUSPATCH_OCI_ARTIFACT_CONFIG_TAG_V1}"
 
+        logger.debug(f"Pull OCI artifact from: {oci_target_name}")
         oci_artifacts = oras_client.pull(target=oci_target_name,
                                          overwrite=True)
+
+        if not oci_artifacts:
+            raise AzCLIError(f"Failed to pull OCI artifact from ACR: {oci_target_name}")
+
         trigger_task = get_task(cmd, registry, CONTINUOUSPATCH_TASK_SCANREGISTRY_NAME)
         file_name = oci_artifacts[0]
         logger.debug(f"OCI artifact file name: {file_name}, trigger task: {trigger_task}")
         config = ContinuousPatchConfig().from_file(file_name, trigger_task)
     except Exception as exception:
         raise AzCLIError(f"Failed to get OCI artifact from ACR: {exception}")
     finally:
-        oras_client.logout(hostname=str.lower(registry.login_server))
+        if oras_client:
+            oras_client.logout(hostname=str.lower(registry.login_server))
 
     return config, file_name
 
@@ -98,19 +106,19 @@ def delete_oci_artifact_continuous_patch(cmd, registry):
 
     try:
         token = _get_acr_token(registry.name, subscription)
-
+        oci_target_name = f"{CSSC_WORKFLOW_POLICY_REPOSITORY}/{CONTINUOUSPATCH_OCI_ARTIFACT_CONFIG}"
         # Delete repository, removing only image isn't deleting the repository always (Bug)
         acr_repository_delete(
             cmd=cmd,
             registry_name=registry.name,
-            repository=f"{CSSC_WORKFLOW_POLICY_REPOSITORY}/{CONTINUOUSPATCH_OCI_ARTIFACT_CONFIG}",
+            repository=oci_target_name,
             username=BEARER_TOKEN_USERNAME,
             password=token,
             yes=True)
         logger.debug("Call to acr_repository_delete completed successfully")
     except Exception as exception:
         logger.debug(exception)
-        logger.error(f"{CSSC_WORKFLOW_POLICY_REPOSITORY}/{CONTINUOUSPATCH_OCI_ARTIFACT_CONFIG}:{CONTINUOUSPATCH_OCI_ARTIFACT_CONFIG_TAG_V1} might not exist or attempt to delete failed.")
+        logger.error(f"{oci_target_name}:{CONTINUOUSPATCH_OCI_ARTIFACT_CONFIG_TAG_V1} might not exist or attempt to delete failed.")
         raise
 
 
@@ -191,9 +199,10 @@ def from_json(self, json_str, trigger_task=None):
         try:
             json_config = json.loads(json_str)
             validate(json_config, CONTINUOUSPATCH_CONFIG_SCHEMA_V1)
+        except json.JSONDecodeError as e:
+            raise AzCLIError(f"Failed to parse JSON: {e}", e)
         except ValidationError as e:
-            logger.error("Error validating the continuous patch config file: %s", e)
-            return None
+            raise AzCLIError(f"Error validating the continuous patch config file: {e}", e)
 
         self.version = json_config.get("version", "")
         repositories = json_config.get("repositories", [])

src/acrcssc/azext_acrcssc/helper/_taskoperations.py

@@ -63,24 +63,23 @@ def create_update_continuous_patch_v1(cmd,
     schedule_cron_expression = None
     cssc_tasks_exists, task_list = check_continuous_task_exists(cmd, registry)
 
-    if schedule is not None:
+    if schedule:
         schedule_cron_expression = convert_timespan_to_cron(schedule)
         logger.debug(f"converted schedule to cron expression: {schedule_cron_expression}")
 
     if is_create_workflow:
         if cssc_tasks_exists:
             raise AzCLIError(f"{ERROR_MESSAGE_WORKFLOW_TASKS_ALREADY_EXISTS}")
 
-        if cssc_config_file is not None:
-            create_oci_artifact_continuous_patch(registry, cssc_config_file, dryrun)
-            logger.debug(f"Uploading of {cssc_config_file} for create completed successfully.")
+        create_oci_artifact_continuous_patch(registry, cssc_config_file, dryrun)
+        logger.debug(f"Uploading of {cssc_config_file} for create completed successfully.")
 
         _create_cssc_workflow(cmd, registry, schedule_cron_expression, resource_group, dryrun)
     else:
         if not cssc_tasks_exists:
             raise AzCLIError(f"{ERROR_MESSAGE_WORKFLOW_TASKS_DOES_NOT_EXIST}")
 
-        if cssc_config_file is not None:
+        if cssc_config_file:
             create_oci_artifact_continuous_patch(registry, cssc_config_file, dryrun)
             logger.debug(f"Uploading of {cssc_config_file} for update completed successfully.")
 
@@ -136,7 +135,7 @@ def _update_cssc_workflow(cmd, registry, schedule_cron_expression, resource_grou
             logger.debug(f"Task {task.name} is different from the extension task, updating the task")
             _update_task_yaml(acr_task_client, registry, resource_group, task, extension_task)
 
-    if schedule_cron_expression is not None:
+    if schedule_cron_expression:
         _update_task_schedule(acr_task_client, registry, resource_group, schedule_cron_expression, dry_run)
 
 
@@ -476,23 +475,26 @@ def _delete_task_role_assignment(cli_ctx, acrtask_client, registry, resource_gro
 
     identity = task.identity
 
-    if identity:
-        assigned_roles = role_client.role_assignments.list_for_scope(
-            registry.id,
-            filter=f"principalId eq '{identity.principal_id}'"
-        )
+    if not identity or not identity.principal_id:
+        logger.debug(f"Task {task_name} has no associated managed identity. Skipping role assignment deletion.")
+        return None
+
+    assigned_roles = role_client.role_assignments.list_for_scope(
+        registry.id,
+        filter=f"principalId eq '{identity.principal_id}'"
+    )
 
-        for role in assigned_roles:
-            try:
-                logger.debug(f"Deleting role assignments of task {task_name} from the registry")
-                role_client.role_assignments.delete(
-                    scope=registry.id,
-                    role_assignment_name=role.name
-                )
-            except ResourceNotFoundError:
-                logger.debug(f"Role assignment {role.name} does not exist in registry {registry.name}")
-            except AzCLIError as exception:
-                logger.error(f"Failed to delete role assignment {role.name} from registry {registry.name} : {exception}")
+    for role in assigned_roles:
+        try:
+            logger.debug(f"Deleting role assignments of task {task_name} from the registry")
+            role_client.role_assignments.delete(
+                scope=registry.id,
+                role_assignment_name=role.name
+            )
+        except ResourceNotFoundError:
+            logger.debug(f"Role assignment {role.name} does not exist in registry {registry.name}")
+        except AzCLIError as exception:
+            logger.error(f"Failed to delete role assignment {role.name} from registry {registry.name} : {exception}")
 
 
 def _transform_task_list(tasks):
@@ -531,17 +533,6 @@ def _get_custom_registry_credentials(cmd):
     )
 
 
-def _is_vault_secret(cmd, credential):
-    keyvault_dns = None
-    try:
-        keyvault_dns = cmd.cli_ctx.cloud.suffixes.keyvault_dns
-    except Exception:
-        return False
-    if credential is not None:
-        return keyvault_dns.upper() in credential.upper()
-    return False
-
-
 def get_next_date(cron_expression):
     from croniter import croniter
     now = datetime.now(timezone.utc)