[minor] New mirror capability based on ISC

.secrets.baseline

@@ -3,7 +3,7 @@
     "files": "build/bin/config/oscap/ssg-rhel9-ds.xml|^.secrets.baseline$",
     "lines": null
   },
-  "generated_at": "2026-01-29T07:04:47Z",
+  "generated_at": "2026-02-12T15:32:55Z",
   "plugins_used": [
     {
       "name": "AWSKeyDetector"
@@ -805,6 +805,46 @@
         "verified_result": null
       }
     ],
+    "python/test/mirror/test_mirror_advanced.py": [
+      {
+        "hashed_secret": "206c80413b9a96c1312cc346b7d2517b84463edd",
+        "is_secret": false,
+        "is_verified": false,
+        "line_number": 52,
+        "type": "Secret Keyword",
+        "verified_result": null
+      }
+    ],
+    "python/test/mirror/test_mirror_auth.py": [
+      {
+        "hashed_secret": "1c58bd92003bbaa0538e249fff6ee19a270dec5f",
+        "is_secret": false,
+        "is_verified": false,
+        "line_number": 118,
+        "type": "Secret Keyword",
+        "verified_result": null
+      }
+    ],
+    "python/test/mirror/test_mirror_basic.py": [
+      {
+        "hashed_secret": "789cbe0407840b1c2041cb33452ff60f19bf58cc",
+        "is_secret": false,
+        "is_verified": false,
+        "line_number": 137,
+        "type": "Secret Keyword",
+        "verified_result": null
+      }
+    ],
+    "python/test/mirror/test_mirror_packages.py": [
+      {
+        "hashed_secret": "206c80413b9a96c1312cc346b7d2517b84463edd",
+        "is_secret": false,
+        "is_verified": false,
+        "line_number": 292,
+        "type": "Secret Keyword",
+        "verified_result": null
+      }
+    ],
     "tekton/src/tasks/fvt/mas-fvt-assist-desktop.yml.j2": [
       {
         "hashed_secret": "ff0c19d6cf999d585c440a2143db929839c48fb6",

image/cli/mascli/mas

@@ -181,7 +181,7 @@ case $1 in
     teardown_mirror_registry "$@"
     ;;
 
-  mirror|mirror-images|mirror-ibm-images)
+  mirror-images|mirror-ibm-images)
     echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" >> $LOGFILE
     echo "!! mirror-ibm-images                                                         !!" >> $LOGFILE
     echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" >> $LOGFILE
@@ -242,6 +242,16 @@ case $1 in
     configure_ocp_for_mirror "$@"
     ;;
 
+  mirror)
+    echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" >> $LOGFILE
+    echo "!! mirror                                                                    !!" >> $LOGFILE
+    echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" >> $LOGFILE
+    # Take the first parameter off (it will be "mirror")
+    shift
+    # Run the new Python-based mirror
+    mas-cli mirror "$@"
+    ;;
+
   install)
     echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" >> $LOGFILE
     echo "!! install                                                                   !!" >> $LOGFILE

python/setup.py

@@ -59,18 +59,19 @@ def get_version(rel_path):
     long_description=long_description,
     install_requires=[
         'mas-devops >= 5.2.0',   # EPL
+        'alive-progress',        # MIT License
         'halo',                  # MIT License
         'prompt_toolkit',        # BSD License
         'openshift',             # Apache Software License
-        'kubernetes == 33.1.0',            # Apache Software License # Version lock to be removed once https://github.com/kubernetes-client/python/issues/2460 is resolved
-        'tabulate'               # MIT License
+        'kubernetes == 33.1.0',  # Apache Software License, version lock to be removed once https://github.com/kubernetes-client/python/issues/2460 is resolved
+        'tabulate',              # MIT License
     ],
     extras_require={
         'dev': [
             'build',       # MIT License
             'flake8',      # MIT License
             'pytest',      # MIT License
-            'pyinstaller'  # GPL, https://pyinstaller.org/en/stable/license.html & https://github.com/pyinstaller/pyinstaller/wiki/FAQ#license
+            'pyinstaller',  # GPL, https://pyinstaller.org/en/stable/license.html & https://github.com/pyinstaller/pyinstaller/wiki/FAQ#license
         ]
     },
     classifiers=[
@@ -82,6 +83,6 @@ def get_version(rel_path):
         'Programming Language :: Python :: 3.12',
         'Topic :: Communications',
         'Topic :: Internet',
-        'Topic :: Software Development :: Libraries :: Python Modules'
+        'Topic :: Software Development :: Libraries :: Python Modules',
     ]
 )

python/src/mas-cli

@@ -20,6 +20,7 @@ from mas.cli.aiservice.upgrade.app import AiServiceUpgradeApp
 from mas.cli.update.app import UpdateApp
 from mas.cli.upgrade.app import UpgradeApp
 from mas.cli.uninstall.app import UninstallApp
+from mas.cli.mirror.app import MirrorApp
 
 from prompt_toolkit import HTML, print_formatted_text
 from urllib3.exceptions import MaxRetryError
@@ -44,7 +45,7 @@ def usage():
         + " - <ForestGreen>mas-cli update</ForestGreen>    Apply updates and security fixes\n"  # noqa: W503
         + " - <ForestGreen>mas-cli upgrade</ForestGreen>   Upgrade to a new MAS release\n"  # noqa: W503
         + " - <ForestGreen>mas-cli uninstall</ForestGreen> Remove MAS from the cluster\n"  # noqa: W503
-
+        + " - <ForestGreen>mas-cli mirror</ForestGreen>    Mirror container images \n"  # noqa: W503
     ))
     print_formatted_text(HTML("For usage information run <ForestGreen>mas-cli [action] --help</ForestGreen>\n"))
 
@@ -71,6 +72,9 @@ if __name__ == '__main__':
         elif function == "upgrade":
             app = UpgradeApp()
             app.upgrade(argv[2:])
+        elif function == "mirror":
+            app = MirrorApp()
+            app.mirror(argv[2:])
         elif function in ["-h", "--help"]:
             usage()
             exit(0)

python/src/mas/cli/cli.py

@@ -19,7 +19,7 @@
 from subprocess import PIPE, Popen, TimeoutExpired
 import threading
 import json
-from typing import List, Dict, Any, Callable, Type
+from typing import List, Dict, Any, Callable, Type, NoReturn
 
 # Use of the openshift client rather than the kubernetes client allows us access to "apply"
 from kubernetes import config
@@ -294,7 +294,7 @@ def getCompatibleVersions(self, coreChannel: str, appId: str) -> List[str]:
             return []
 
     @logMethodCall
-    def fatalError(self, message: str, exception: Exception | None = None) -> None:
+    def fatalError(self, message: str, exception: Exception | None = None) -> NoReturn:
         if exception is not None:
             logger.error(message)
             logger.exception(exception, stack_info=True)

python/src/mas/cli/displayMixins.py

@@ -54,19 +54,25 @@ def printHighlight(self, message: str | list[str]) -> None:
         print_formatted_text(HTML(f"<MediumTurquoise>{message.replace(' & ', ' &amp; ')}</MediumTurquoise>"))
 
     def printWarning(self, message: str) -> None:
+        logger.warning(message)
         print_formatted_text(HTML(f"<Red>Warning: {message.replace(' & ', ' &amp; ')}</Red>"))
 
     def printSummary(self, title: str, value: str) -> None:
         titleLength = len(title)
         message = f"{title} {'.' * (40 - titleLength)} {value}"
+
+        logger.debug(f"Summary: {title} = {value}")
         print_formatted_text(HTML(f"  <{SUMMARYCOLOR}>{message.replace(' & ', ' &amp; ')}</{SUMMARYCOLOR}>"))
 
     def printParamSummary(self, message: str, param: str) -> None:
         if self.getParam(param) is None:  # type: ignore
+            logger.debug(f"Parameter Summary: {param} = undefined")
             self.printSummary(message, f"<{UNDEFINEDPARAMCOLOR}>Undefined</{UNDEFINEDPARAMCOLOR}>")
         elif self.getParam(param) == "":  # type: ignore
+            logger.debug(f"Parameter Summary: {param} = \"\"")
             self.printSummary(message, f"<{UNDEFINEDPARAMCOLOR}>Default</{UNDEFINEDPARAMCOLOR}>")
         else:
+            logger.debug(f"Parameter Summary: {param} = {self.getParam(param)}")  # type: ignore
             self.printSummary(message, self.getParam(param))  # type: ignore
 
 

python/src/mas/cli/install/__init__.py

@@ -1,5 +1,5 @@
 # *****************************************************************************
-# Copyright (c) 2024 IBM Corporation and other Contributors.
+# Copyright (c) 2026 IBM Corporation and other Contributors.
 #
 # All rights reserved. This program and the accompanying materials
 # are made available under the terms of the Eclipse Public License v1.0

python/src/mas/cli/install/app.py

@@ -19,6 +19,8 @@
 import calendar
 from openshift.dynamic.exceptions import NotFoundError
 
+from typing import Dict, Any
+
 from prompt_toolkit import prompt, print_formatted_text, HTML
 from prompt_toolkit.completion import WordCompleter
 
@@ -84,11 +86,14 @@ def wrapper(self, *args, **kwargs):
 class InstallApp(BaseApp, InstallSettingsMixin, InstallSummarizerMixin, ConfigGeneratorMixin, installArgBuilderMixin):
     @logMethodCall
     def validateCatalogSource(self):
-        # Check supported OCP versions
-        ocpVersion = getClusterVersion(self.dynamicClient)
-        supportedReleases = self.chosenCatalog.get("ocp_compatibility", [])
-        if len(supportedReleases) > 0 and not isClusterVersionInRange(ocpVersion, supportedReleases):
-            self.fatalError(f"IBM Maximo Operator Catalog {self.getParam('mas_catalog_version')} is not compatible with OpenShift v{ocpVersion}.  Compatible OpenShift releases are {supportedReleases}")
+        # Check supported OCP versions - but we can only do this in non-development mode because in development mode
+        # we do not load catalog metadata files
+        if not self.devMode:
+            assert self.chosenCatalog is not None, "validateCatalogSource() called before catalog was chosen"
+            ocpVersion = getClusterVersion(self.dynamicClient)
+            supportedReleases = self.chosenCatalog.get("ocp_compatibility", [])
+            if len(supportedReleases) > 0 and not isClusterVersionInRange(ocpVersion, supportedReleases):
+                self.fatalError(f"IBM Maximo Operator Catalog {self.getParam('mas_catalog_version')} is not compatible with OpenShift v{ocpVersion}.  Compatible OpenShift releases are {supportedReleases}")
 
         # Compare with any existing installed catalog
         catalogsAPI = self.dynamicClient.resources.get(api_version="operators.coreos.com/v1alpha1", kind="CatalogSource")
@@ -105,6 +110,7 @@ def validateCatalogSource(self):
                 catalogId = "v8-amd64"
             else:
                 self.fatalError(f"IBM Maximo Operator Catalog is already installed on this cluster. However, it is not possible to identify its version. If you wish to install a new MAS instance using the {self.getParam('mas_catalog_version')} catalog please first run 'mas update' to switch to this catalog, this will ensure the appropriate actions are performed as part of the catalog update")
+                assert False, "fatalError() should have exited"  # Let basepyright know that fatalError() will exit
 
             if catalogId != self.getParam("mas_catalog_version"):
                 self.fatalError(f"IBM Maximo Operator Catalog {catalogId} is already installed on this cluster, if you wish to install a new MAS instance using the {self.getParam('mas_catalog_version')} catalog please first run 'mas update' to switch to this catalog, this will ensure the appropriate actions are performed as part of the catalog update")
@@ -181,6 +187,7 @@ def formatCatalog(self, name: str) -> str:
 
     @logMethodCall
     def processCatalogChoice(self) -> list:
+        assert self.chosenCatalog is not None, "processCatalogChoice() called before catalog was chosen"
         self.catalogDigest = self.chosenCatalog["catalog_digest"]
         self.catalogMongoDbVersion = self.chosenCatalog["mongo_extras_version_default"]
         self.catalogDb2Channel = self.chosenCatalog.get("db2_channel_default", "v110509.0")  # Returns fallback "v110509.0" for old catalogs without this field
@@ -499,6 +506,7 @@ def configCP4D(self):
         if self.getParam("mas_catalog_version") in self.catalogOptions:
             # Note: this will override any version provided by the user (which is intentional!)
             logger.debug(f"Using automatic CP4D product version: {self.getParam('cpd_product_version')}")
+            assert self.chosenCatalog is not None, "chosenCatalog should be set in this scenario but was not"
             self.setParam("cpd_product_version", self.chosenCatalog["cpd_product_version_default"])
         elif self.getParam("cpd_product_version") == "":
             if self.noConfirm:
@@ -594,7 +602,7 @@ def configCATrust(self) -> None:
             ])
             self.yesOrNo("Trust default CAs", "mas_trust_default_cas")
         else:
-            self.setParam("mas_trust_default_cas", True)
+            self.setParam("mas_trust_default_cas", "true")
 
     @logMethodCall
     def configOperationMode(self):
@@ -888,8 +896,8 @@ def configDNSAndCerts(self):
                     self.setParam("mas_domain", "")
                     self.setParam("mas_cluster_issuer", "")
                 self.manualCerts = self.yesOrNo("Configure manual certificates")
-                self.setParam("mas_manual_cert_mgmt", self.manualCerts)
-                if self.getParam("mas_manual_cert_mgmt"):
+                self.setParam("mas_manual_cert_mgmt", str(self.manualCerts).lower())
+                if self.getParam("mas_manual_cert_mgmt").lower() == "true":
                     self.manualCertsDir = self.promptForDir("Enter the path containing the manual certificates", mustExist=True)
                 else:
                     self.manualCertsDir = None
@@ -1124,7 +1132,7 @@ def assistSettings(self) -> None:
 
             # We still use the old name for ODF (OCS)
             if self.getParam("cos_type") == "odf":
-                self.setParam("cos_type") == "ocs"
+                self.setParam("cos_type", "ocs")
 
             if self.getParam("cos_type") == "ibm":
                 self.promptForString("IBM Cloud API Key", "cos_apikey", isPassword=True)
@@ -1462,7 +1470,7 @@ def nonInteractiveMode(self) -> None:
         self.installAIService = False
         self.slsLicenseFileLocal = None
 
-        self.approvals = {
+        self.approvals: Dict[str, Dict[str, Any]] = {
             "approval_core": {"id": "suite-verify"},  # After Core Platform verification has completed
             "approval_assist": {"id": "app-cfg-assist"},  # After Assist workspace has been configured
             "approval_iot": {"id": "app-cfg-iot"},  # After IoT workspace has been configured
@@ -1577,7 +1585,7 @@ def nonInteractiveMode(self) -> None:
                     self.setParam("db2_action_aiservice", "install")
                     self.installAIService = True
                     # Set manage - bind - AI Service params same as provided AI Service's params
-                    self.setParam("manage_bind_aiservice_instance_id", vars(self.args).get("aiservice_instance_id"))
+                    self.setParam("manage_bind_aiservice_instance_id", vars(self.args).get("aiservice_instance_id", ""))
                     self.setParam("manage_bind_aiservice_tenant_id", "user")
             elif key == "manage_bind_aiservice_instance_id":
                 # only set if AI Service not being installed
@@ -1658,14 +1666,14 @@ def nonInteractiveMode(self) -> None:
 
             elif key == "manual_certificates":
                 if value is not None:
-                    self.setParam("mas_manual_cert_mgmt", True)
+                    self.setParam("mas_manual_cert_mgmt", "true")
                     self.manualCertsDir = value
                 else:
-                    self.setParam("mas_manual_cert_mgmt", False)
+                    self.setParam("mas_manual_cert_mgmt", "false")
                     self.manualCertsDir = None
 
             elif key == "enable_ipv6":
-                self.setParam("enable_ipv6", True)
+                self.setParam("enable_ipv6", "true")
 
             elif key == "install_minio_aiservice":
                 if vars(self.args).get("aiservice_instance_id"):
@@ -1729,6 +1737,7 @@ def nonInteractiveMode(self) -> None:
             # Verifiy if any of the props that needs to be in a file are given
             if self.getParam("mas_ws_facilities_storage_log_size") != "" or self.getParam("mas_ws_facilities_storage_userfiles_size") != "" or self.getParam("mas_ws_facilities_db_maxconnpoolsize") or self.getParam("mas_ws_facilities_dwfagents"):
                 self.selectLocalConfigDir()
+                assert self.localConfigDir is not None, "localConfigDir is None"
                 facilitiesConfigsPath = path.join(self.localConfigDir, "facilities-configs.yaml")
                 self.generateFacilitiesCfg(destination=facilitiesConfigsPath)
                 self.setParam("mas_ws_facilities_config_map_name", "facilities-config")
@@ -1826,6 +1835,7 @@ def install(self, argv):
         self.isAirgap()
 
         # Configure the installOptions for the appropriate architecture
+        assert self.architecture is not None, "Target architecture is not set"
         self.catalogOptions = supportedCatalogs[self.architecture]
 
         # Basic settings before the user provides any input
@@ -1962,9 +1972,9 @@ def install(self, argv):
 
         # Based on the parameters set the annotations correctly
         self.configAnnotations()
-
         self.displayInstallSummary()
 
+        continueWithInstall = True
         if not self.noConfirm:
             print()
             self.printDescription([
@@ -1973,7 +1983,7 @@ def install(self, argv):
             continueWithInstall = self.yesOrNo("Proceed with these settings")
 
         # Prepare the namespace and launch the installation pipeline
-        if self.noConfirm or continueWithInstall:
+        if continueWithInstall:
             self.createTektonFileWithDigest()
 
             self.printH1("Launch Install")

python/src/mas/cli/install/argBuilder.py

@@ -92,10 +92,10 @@ def buildCommand(self) -> str:
         if self.operationalMode == 2:
             command += f"  --non-prod{newline}"
 
-        if self.getParam('mas_trust_default_cas') == "false":
+        if self.getParam('mas_trust_default_cas').lower() == "false":
             command += f"  --disable-ca-trust{newline}"
 
-        if self.getParam('mas_manual_cert_mgmt') is True:
+        if self.getParam('mas_manual_cert_mgmt').lower() == "true":
             command += f"  --manual-certificates \"{self.manualCertsDir}\"{newline}"
 
         if self.getParam('mas_routing_mode') != "":
@@ -104,7 +104,7 @@ def buildCommand(self) -> str:
         if self.getParam('mas_ingress_controller_name') != "":
             command += f"  --ingress-controller \"{self.getParam('mas_ingress_controller_name')}\"{newline}"
 
-        if self.getParam('mas_configure_ingress') is True:
+        if self.getParam('mas_configure_ingress').lower() == "true":
             command += f"  --configure-ingress{newline}"
 
         if self.getParam('mas_domain') != "":
@@ -125,19 +125,19 @@ def buildCommand(self) -> str:
         if self.getParam('mas_cluster_issuer') != "":
             command += f"  --mas-cluster-issuer \"{self.getParam('mas_cluster_issuer')}\"{newline}"
 
-        if self.getParam('mas_enable_walkme') == "false":
+        if self.getParam('mas_enable_walkme').lower() == "false":
             command += f"  --disable-walkme{newline}"
 
-        if self.getParam('mas_feature_usage') == "false":
+        if self.getParam('mas_feature_usage').lower() == "false":
             command += f"  --disable-feature-usage{newline}"
 
-        if self.getParam('mas_usability_metrics') == "false":
+        if self.getParam('mas_usability_metrics').lower() == "false":
             command += f"  --disable-usability-metrics{newline}"
 
-        if self.getParam('mas_deployment_progression') == "false":
+        if self.getParam('mas_deployment_progression').lower() == "false":
             command += f"  --disable-deployment-progression{newline}"
 
-        if self.getParam('enable_ipv6') is True:
+        if self.getParam('enable_ipv6').lower() == "true":
             command += f"  --enable-ipv6{newline}"
 
         # Storage