AZ ACR CSSC Extension

src/acrcssc/HISTORY.rst

@@ -0,0 +1,8 @@
+.. :changelog:
+
+Release History
+===============
+
+0.1.0
+++++++
+* Initial release.

src/acrcssc/README.rst

@@ -0,0 +1,5 @@
+Microsoft Azure CLI 'acrcssc' Extension
+==========================================
+
+This package is for the 'acrcssc' extension.
+i.e. 'az acrcssc'

src/acrcssc/azext_acrcssc/__init__.py

@@ -0,0 +1,32 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from azure.cli.core import AzCommandsLoader
+
+from azext_acrcssc._help import helps  # pylint: disable=unused-import
+
+
+class AcrcsscCommandsLoader(AzCommandsLoader):
+
+    def __init__(self, cli_ctx=None):
+        from azure.cli.core.commands import CliCommandType
+        from azext_acrcssc._client_factory import cf_acr
+        acrcssc_custom = CliCommandType(
+            operations_tmpl='azext_acrcssc.custom#{}',
+            client_factory=cf_acr)
+        super(AcrcsscCommandsLoader, self).__init__(cli_ctx=cli_ctx,
+                                                    custom_command_type=acrcssc_custom)
+
+    def load_command_table(self, args):
+        from azext_acrcssc.commands import load_command_table
+        load_command_table(self, args)
+        return self.command_table
+
+    def load_arguments(self, command):
+        from azext_acrcssc._params import load_arguments
+        load_arguments(self, command)
+
+
+COMMAND_LOADER_CLS = AcrcsscCommandsLoader

src/acrcssc/azext_acrcssc/_client_factory.py

@@ -0,0 +1,61 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+from azure.cli.core.commands.client_factory import get_mgmt_service_client
+from azure.cli.core.profiles import ResourceType
+from azure.mgmt.containerregistry import ContainerRegistryManagementClient
+from .helper._constants import (
+    ACR_API_VERSION_2023_01_01_PREVIEW,
+    ACR_API_VERSION_2019_06_01_PREVIEW
+)
+
+from azure.mgmt.authorization import AuthorizationManagementClient
+
+
+def cf_acr(cli_ctx, *_) -> ContainerRegistryManagementClient:
+    return get_mgmt_service_client(cli_ctx,
+                                   ResourceType.MGMT_CONTAINERREGISTRY,
+                                   api_version=ACR_API_VERSION_2023_01_01_PREVIEW)
+
+
+def cf_acr_registries(cli_ctx, *_) -> ContainerRegistryManagementClient:
+    return get_mgmt_service_client(cli_ctx,
+                                   ResourceType.MGMT_CONTAINERREGISTRY,
+                                   api_version=ACR_API_VERSION_2023_01_01_PREVIEW).registries
+
+
+def cf_acr_tasks(cli_ctx, *_):
+    return get_mgmt_service_client(cli_ctx,
+                                   ResourceType.MGMT_CONTAINERREGISTRY,
+                                   api_version=ACR_API_VERSION_2019_06_01_PREVIEW).tasks
+
+
+def cf_acr_registries_tasks(cli_ctx, *_):
+    return get_mgmt_service_client(cli_ctx,
+                                   ResourceType.MGMT_CONTAINERREGISTRY,
+                                   api_version=ACR_API_VERSION_2019_06_01_PREVIEW).registries
+
+
+def cf_acr_taskruns(cli_ctx, *_):
+    return get_mgmt_service_client(cli_ctx,
+                                   ResourceType.MGMT_CONTAINERREGISTRY,
+                                   api_version=ACR_API_VERSION_2019_06_01_PREVIEW).task_runs
+
+
+def cf_acr_runs(cli_ctx, *_):
+    return get_mgmt_service_client(cli_ctx,
+                                   ResourceType.MGMT_CONTAINERREGISTRY,
+                                   api_version=ACR_API_VERSION_2019_06_01_PREVIEW).runs
+
+
+def cf_resources(cli_ctx, subscription_id=None):
+    return get_mgmt_service_client(cli_ctx,
+                                   ResourceType.MGMT_RESOURCE_RESOURCES,
+                                   subscription_id=subscription_id)
+
+
+def cf_authorization(cli_ctx, subscription_id=None) -> AuthorizationManagementClient:
+    return get_mgmt_service_client(cli_ctx,
+                                   ResourceType.MGMT_AUTHORIZATION,
+                                   subscription_id=subscription_id, api_version="2022-04-01")

src/acrcssc/azext_acrcssc/_help.py

@@ -0,0 +1,50 @@
+# coding=utf-8
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from knack.help_files import helps  # pylint: disable=unused-import
+
+helps['acr supply-chain'] = """
+    type: group
+    short-summary: Commands to manage acr supply chain resources.
+"""
+
+helps['acr supply-chain workflow'] = """
+    type: group
+    short-summary: Commands to manage acr supply chain workflows.
+"""
+
+helps['acr supply-chain workflow create'] = """
+    type: command
+    short-summary: Create acr supply chain workflow.
+    examples:
+        - name: Create acr supply chain workflow
+          text: az acr supply-chain workflow create -r $MyRegistry -g $MyResourceGroup \
+                --type continuouspatchv1 --cadence 1d --config path-to-config-file
+"""
+helps['acr supply-chain workflow update'] = """
+    type: command
+    short-summary: Update acr supply chain workflow.
+    examples:
+        - name: Updates acr supply chain workflow
+          text: az acr supply-chain workflow update -r $MyRegistry -g $MyResourceGroup --type \
+                continuouspatchv1 --cadence 1d --config path-to-config-file
+"""
+
+helps['acr supply-chain workflow show'] = """
+     type: command
+     short-summary: Show acr supply chain workflow tasks.
+     examples:
+        - name: Show all acr supply chain workflow
+          text: az acr supply-chain workflow show -r $MyRegistry -g $MyResourceGroup --type continuouspatchv1
+"""
+
+helps['acr supply-chain workflow delete'] = """
+    type: command
+    short-summary: Delete acr supply chain workflow.
+    examples:
+        - name: Delete acr supply chain workflow and associated configuration files
+          text: az acr supply-chain workflow delete -r $MyRegistry -g $MyResourceGroup --type continuouspatchv1
+"""

src/acrcssc/azext_acrcssc/_params.py

@@ -0,0 +1,32 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+# pylint: disable=line-too-long
+from azure.cli.core import AzCommandsLoader
+from azure.cli.core.commands.parameters import (get_resource_name_completion_list, get_three_state_flag, get_enum_type)
+
+from azure.cli.command_modules.acr._constants import REGISTRY_RESOURCE_TYPE
+
+from azure.cli.command_modules.acr._validators import validate_registry_name
+
+
+def load_arguments(self: AzCommandsLoader, _):
+    from .helper._constants import CSSCTaskTypes
+
+    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('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)
+
+    with self.argument_context("acr supply-chain workflow create") as c:
+        c.argument("config", options_list=["--config"], help="Configuration file path containing the json schema for the list of repositories and tags to filter within the registry. Schema example:{\"repositories\":[{\"repository\":\"alpine\",\"tags\":[\"tag1\",\"tag2\"],\"enabled\":true},{\"repository\":\"python\",\"tags\":[\"*\"],\"enabled\":false}], \"version\": \"v1\"}", required=True)
+        c.argument("cadence", options_list=["--cadence"], help="Cadence to run the scan and patching task. E.g. `<n>d` where <n> is the number of days between each run. Max value is 30d.", required=True)
+        c.argument("defer_immediate_run", options_list=["--defer-immediate-run"], help="Use this flag to defer immediately running of selected workflow task. Default value: false.", arg_type=get_three_state_flag(), required=False)
+        c.argument("dryrun", options_list=["--dry-run"], help="Use this flag to see the qualifying repositories and tags that would be affected by the workflow. Default value: false. 'config' parameter is mandatory to provide with dry-run", arg_type=get_three_state_flag(), required=False)
+
+    with self.argument_context("acr supply-chain workflow update") as c:
+        c.argument("config", options_list=["--config"], help="Configuration file path containing the json schema for the list of repositories and tags to filter within the registry. Schema example:{\"repositories\":[{\"repository\":\"alpine\",\"tags\":[\"tag1\",\"tag2\"],\"enabled\":true},{\"repository\":\"python\",\"tags\":[\"*\"],\"enabled\":false}], \"version\": \"v1\"}", required=False)
+        c.argument("cadence", options_list=["--cadence"], help="Cadence to run the scan and patching task. E.g. `<n>d` where n is the number of days between each run. Max value is 30d.", required=True)
+        c.argument("defer_immediate_run", options_list=["--defer-immediate-run"], help="Use this flag to defer immediately running of selected workflow task. Default value: false.", arg_type=get_three_state_flag(), required=False)
+        c.argument("dryrun", options_list=["--dry-run"], help="Use this flag to see the qualifying repositories and tags that would be affected by the workflow. Default value: false. 'config' parameter is mandatory to provide with dry-run", arg_type=get_three_state_flag(), required=False)

src/acrcssc/azext_acrcssc/_validators.py

@@ -0,0 +1,135 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+# pylint: disable=line-too-long
+# pylint: disable=broad-exception-caught
+# pylint: disable=logging-fstring-interpolation
+import json
+import os
+import re
+from knack.log import get_logger
+from azure.cli.command_modules.acr.repository import acr_repository_show
+from .helper._constants import (
+    BEARER_TOKEN_USERNAME,
+    CSSC_WORKFLOW_POLICY_REPOSITORY,
+    CONTINUOSPATCH_OCI_ARTIFACT_CONFIG,
+    CONTINUOUSPATCH_CONFIG_SCHEMA_V1,
+    CONTINUOUSPATCH_CONFIG_SCHEMA_SIZE_LIMIT,
+    CONTINUOSPATCH_ALL_TASK_NAMES,
+    ERROR_MESSAGE_INVALID_TIMESPAN_FORMAT,
+    ERROR_MESSAGE_INVALID_TIMESPAN_VALUE,
+    RESOURCE_GROUP,
+    SUBSCRIPTION)
+from .helper._constants import CSSCTaskTypes, ERROR_MESSAGE_INVALID_TASK, RECOMMENDATION_CADENCE
+from .helper._ociartifactoperations import _get_acr_token
+from azure.mgmt.core.tools import (parse_resource_id)
+from azure.cli.core.azclierror import InvalidArgumentValueError
+from ._client_factory import cf_acr_tasks
+logger = get_logger(__name__)
+
+
+def validate_continuouspatch_config_v1(config_path):
+    _validate_continuouspatch_file(config_path)
+    _validate_continuouspatch_json(config_path)
+
+
+def _validate_continuouspatch_file(config_path):
+    if not os.path.exists(config_path):
+        raise InvalidArgumentValueError(f"Config path file: {config_path} does not exist in the path specified")
+    if not os.path.isfile(config_path):
+        raise InvalidArgumentValueError(f"Config path file: {config_path} is not a valid file")
+    if os.path.getsize(config_path) > CONTINUOUSPATCH_CONFIG_SCHEMA_SIZE_LIMIT:
+        raise InvalidArgumentValueError(f"Config path file: {config_path} is too large. Max size limit is 10 MB")
+    if os.path.getsize(config_path) == 0:
+        raise InvalidArgumentValueError(f"Config path file: {config_path} is empty")
+    if not os.access(config_path, os.R_OK):
+        raise InvalidArgumentValueError(f"Config path file: '{config_path}' is not readable")
+
+
+def _validate_continuouspatch_json(config_path):
+    from jsonschema import validate
+    try:
+        with open(config_path, 'r') as f:
+            config = json.load(f)
+            validate(config, CONTINUOUSPATCH_CONFIG_SCHEMA_V1)
+    except Exception as e:
+        logger.debug(f"Error validating the continuous patch config file: {e}")
+        raise InvalidArgumentValueError("File used for --config is not a valid config JSON file. Use --help to see the schema of the config file.")
+    finally:
+        f.close()
+
+
+def check_continuous_task_exists(cmd, registry):
+    exists = False
+    for task_name in CONTINUOSPATCH_ALL_TASK_NAMES:
+        exists = exists or _check_task_exists(cmd, registry, task_name)
+    return exists
+
+
+def check_continuous_task_config_exists(cmd, registry):
+    # A client cannot be used in this situation because the 'show registry/image'
+    # is a data plane operation and the az cli does not include the data plane API.
+    subscription = parse_resource_id(registry.id)[SUBSCRIPTION]
+    try:
+        token = _get_acr_token(registry.name, subscription)
+        acr_repository_show(
+            cmd=cmd,
+            registry_name=registry.name,
+            repository=f"{CSSC_WORKFLOW_POLICY_REPOSITORY}/{CONTINUOSPATCH_OCI_ARTIFACT_CONFIG}",
+            username=BEARER_TOKEN_USERNAME,
+            password=token)
+    except Exception as exception:
+        if hasattr(exception, 'status_code') and exception.status_code == 404:
+            return False
+        # report on the error only if we get something other than 404
+        logger.debug(f"Failed to find config {CSSC_WORKFLOW_POLICY_REPOSITORY}/{CONTINUOSPATCH_OCI_ARTIFACT_CONFIG} from registry {registry.name} : {exception}")
+        raise
+    return True
+
+
+def _check_task_exists(cmd, registry, task_name=""):
+    acrtask_client = cf_acr_tasks(cmd.cli_ctx)
+    resourceid = parse_resource_id(registry.id)
+    resource_group = resourceid[RESOURCE_GROUP]
+
+    try:
+        task = acrtask_client.get(resource_group, registry.name, task_name)
+    except Exception as exception:
+        logger.debug(f"Failed to find task {task_name} from registry {registry.name} : {exception}")
+        return False
+
+    if task is not None:
+        return True
+    return False
+
+
+def _validate_cadence(cadence):
+    # during update, cadence can be null if we are only updating the config
+    if cadence is None:
+        return
+    # Extract the numeric value and unit from the timespan expression
+    match = re.match(r'(\d+)(d)$', cadence)
+    if not match:
+        raise InvalidArgumentValueError(error_msg=ERROR_MESSAGE_INVALID_TIMESPAN_FORMAT, recommendation=RECOMMENDATION_CADENCE)
+    if match is not None:
+        value = int(match.group(1))
+        unit = match.group(2)
+    if unit == 'd' and (value < 1 or value > 30):  # day of the month
+        raise InvalidArgumentValueError(error_msg=ERROR_MESSAGE_INVALID_TIMESPAN_VALUE, recommendation=RECOMMENDATION_CADENCE)
+
+
+def validate_inputs(cadence, config_file_path=None):
+    _validate_cadence(cadence)
+    if config_file_path is not None:
+        validate_continuouspatch_config_v1(config_file_path)
+
+
+def validate_task_type(task_type):
+    if (task_type not in [item.value for item in CSSCTaskTypes]):
+        raise InvalidArgumentValueError(error_msg=ERROR_MESSAGE_INVALID_TASK)
+
+
+def validate_cssc_optional_inputs(cssc_config_path, cadence):
+    if cssc_config_path is None and cadence is None:
+        raise InvalidArgumentValueError(error_msg="Provide at least one parameter to update: --cadence or --config")

src/acrcssc/azext_acrcssc/azext_metadata.json

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

src/acrcssc/azext_acrcssc/commands.py

@@ -0,0 +1,18 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+# pylint: disable=line-too-long
+from azext_acrcssc._client_factory import cf_acr
+
+
+def load_command_table(self, _):
+    # required to mark the command as 'preview', can be expanded if additional commands are added
+    self.command_group("acr supply-chain", client_factory=cf_acr, is_preview=True)
+
+    with self.command_group("acr supply-chain workflow", client_factory=cf_acr, is_preview=True) as g:
+        g.custom_command("create", "create_acrcssc")
+        g.custom_command("update", "update_acrcssc")
+        g.custom_command("delete", "delete_acrcssc")
+        g.custom_show_command("show", "show_acrcssc")