Change sign_result_loc to cmd flag both for sign request and maven upload

Author: yma955

charon/cmd/cmd_upload.py

@@ -136,6 +136,16 @@
     default=False
 )
 @option("--dryrun", "-n", is_flag=True, default=False)
+@option(
+    "--sign_result_loc",
+    "-l",
+    default="/tmp/sign",
+    help="""
+    The local save path for oras to pull the radas signature result.
+    Sign request will use this path to download the signature result,
+    Upload will use the file on this path to generate the corresponding .asc files
+    """,
+)
 @command()
 def upload(
     repo: str,
@@ -150,7 +160,8 @@ def upload(
     sign_key: str = "redhatdevel",
     debug=False,
     quiet=False,
-    dryrun=False
+    dryrun=False,
+    sign_result_loc="/tmp/sign"
 ):
     """Upload all files from a released product REPO to Ronda
     Service. The REPO points to a product released tarball which
@@ -221,7 +232,8 @@ def upload(
                 key=sign_key,
                 dry_run=dryrun,
                 manifest_bucket_name=manifest_bucket_name,
-                config=config
+                config=config,
+                sign_result_loc=sign_result_loc
             )
             if not succeeded:
                 sys.exit(1)

charon/config.py

@@ -48,7 +48,7 @@ def validate(self) -> bool:
             logger.error("Missing host name setting for UMB!")
             return False
         if not self.__result_queue:
-            logger.error("Missing the queue setting to receive siging result in UMB!")
+            logger.error("Missing the queue setting to receive signing result in UMB!")
             return False
         if not self.__request_queue:
             logger.error("Missing the queue setting to send signing request in UMB!")
@@ -124,8 +124,10 @@ def __init__(self, data: Dict):
         self.__ignore_signature_suffix: Dict = data.get("ignore_signature_suffix", None)
         self.__signature_command: str = data.get("detach_signature_command", None)
         self.__aws_cf_enable: bool = data.get("aws_cf_enable", False)
+        self.__radas_config_enable: bool = data.get("radas_config_enable", False)
         radas_config: Dict = data.get("radas", None)
         if radas_config:
+            self.__radas_config_enable: bool = True
             self.__radas_config__: RadasConfig = RadasConfig(radas_config)
 
     def get_ignore_patterns(self) -> List[str]:
@@ -155,6 +157,9 @@ def get_detach_signature_command(self) -> str:
     def is_aws_cf_enable(self) -> bool:
         return self.__aws_cf_enable
 
+    def is_radas_config_enable(self) -> bool:
+        return self.__radas_config_enable
+
     def get_radas_config(self) -> RadasConfig:
         return self.__radas_config__
 

charon/constants.py

@@ -175,6 +175,5 @@
 DEFAULT_ERRORS_LOG = "errors.log"
 
 DEFAULT_REGISTRY = "localhost"
-DEFAULT_SIGN_RESULT_LOC = "/tmp/sign"
 DEFAULT_RADAS_SIGN_TIMEOUT_RETRY_COUNT = 10
 DEFAULT_RADAS_SIGN_TIMEOUT_RETRY_INTERVAL = 60

charon/pkgs/maven.py

@@ -275,7 +275,8 @@ def handle_maven_uploading(
     key=None,
     dry_run=False,
     manifest_bucket_name=None,
-    config=None
+    config=None,
+    sign_result_loc="/tmp/sign"
 ) -> Tuple[str, bool]:
     """ Handle the maven product release tarball uploading process.
         * repo is the location of the tarball in filesystem
@@ -415,9 +416,11 @@ def handle_maven_uploading(
         if not conf:
             sys.exit(1)
 
-        if conf.is_radas_sign_enabled():
+        if conf.is_radas_config_enable():
             logger.info("Start generating radas signature files for s3 bucket %s\n", bucket_name)
-            (_failed_metas, _generated_signs) = radas_signature.generate_radas_sign(top_level)
+            (_failed_metas, _generated_signs) = radas_signature.generate_radas_sign(
+                top_level=top_level, sign_result_loc=sign_result_loc
+            )
             if not _generated_signs:
                 logger.error(
                     "No sign result files were downloaded, "
@@ -426,17 +429,17 @@ def handle_maven_uploading(
 
             failed_metas.extend(_failed_metas)
             generated_signs.extend(_generated_signs)
-            logger.info("Singature generation against radas done.\n")
+            logger.info("Radas signature files generation done.\n")
 
-            logger.info("Start upload radas singature files to s3 bucket %s\n", bucket_name)
+            logger.info("Start upload radas signature files to s3 bucket %s\n", bucket_name)
             _failed_metas = s3_client.upload_signatures(
                 meta_file_paths=generated_signs,
                 target=(bucket_name, prefix),
                 product=None,
                 root=top_level
             )
             failed_metas.extend(_failed_metas)
-            logger.info("Signature files uploading against radas done.\n")
+            logger.info("Radas signature files uploading done.\n")
 
         elif gen_sign:
             suffix_list = __get_suffix(PACKAGE_TYPE_MAVEN, conf)

charon/pkgs/radas_signature_handler.py

@@ -14,8 +14,6 @@
 limitations under the License.
 """
 
-import proton
-import proton.handlers
 import threading
 import logging
 import json
@@ -25,109 +23,84 @@
 import time
 from typing import List, Any, Tuple, Callable, Dict
 from charon.config import get_config
-from charon.constants import DEFAULT_SIGN_RESULT_LOC
 from charon.constants import DEFAULT_RADAS_SIGN_TIMEOUT_RETRY_COUNT
 from charon.constants import DEFAULT_RADAS_SIGN_TIMEOUT_RETRY_INTERVAL
 from charon.pkgs.oras_client import OrasClient
+from proton import Event
+from proton.handlers import MessagingHandler
 
 logger = logging.getLogger(__name__)
 
 
-class SignHandler:
-    """
-    Handle the sign result status management
-    """
-
-    _is_processing: bool = True
-    _downloaded_files: List[str] = []
-
-    @classmethod
-    def is_processing(cls) -> bool:
-        return cls._is_processing
-
-    @classmethod
-    def get_downloaded_files(cls) -> List[str]:
-        return cls._downloaded_files.copy()
-
-    @classmethod
-    def set_processing(cls, value: bool) -> None:
-        cls._is_processing = value
-
-    @classmethod
-    def set_downloaded_files(cls, files: List[str]) -> None:
-        cls._downloaded_files = files
-
-
-class UmbListener(proton.handlers.MessagingHandler):
+class UmbListener(MessagingHandler):
     """
     UmbListener class (AMQP version), register this when setup UmbClient
+    Attributes:
+        sign_result_loc (str): Local save path (e.g. “/tmp/sign”) for oras pull result,
+        this value transfers from the cmd flag, should register UmbListener when the client starts
     """
 
-    def __init__(self) -> None:
+    def __init__(self, sign_result_loc: str) -> None:
         super().__init__()
+        self.sign_result_loc = sign_result_loc
 
-    def on_start(self, event: proton.Event) -> None:
+    def on_start(self, event: Event) -> None:
         """
         On start callback
         """
         conf = get_config()
-        if not conf:
+        rconf = conf.get_radas_config() if conf else None
+        if not rconf:
             sys.exit(1)
-        event.container.create_receiver(conf.get_amqp_queue())
+        conn = event.container.connect(rconf.umb_target())
+        event.container.create_receiver(conn, rconf.result_queue())
+        logger.info("Listening on %s, queue: %s", rconf.umb_target(), rconf.result_queue())
 
-    def on_message(self, event: proton.Event) -> None:
+    def on_message(self, event: Event) -> None:
         """
         On message callback
         """
         # handle response from radas in a thread
         thread = threading.Thread(target=self._process_message, args=[event.message.body])
         thread.start()
 
-    def on_error(self, event: proton.Event) -> None:
+    def on_connection_error(self, event: Event) -> None:
         """
-        On error callback
+        On connection error callback
         """
         logger.error("Received an error event:\n%s", event)
 
-    def on_disconnected(self, event: proton.Event) -> None:
+    def on_disconnected(self, event: Event) -> None:
         """
         On disconnected callback
         """
         logger.error("Disconnected from AMQP broker.")
 
-    def _process_message(msg: Any) -> None:
+    def _process_message(self, msg: Any) -> None:
         """
         Process a message received from UMB
         Args:
             msg: The message body received
         """
-        try:
-            msg_dict = json.loads(msg)
-            result_reference_url = msg_dict.get("result_reference")
+        msg_dict = json.loads(msg)
+        result_reference_url = msg_dict.get("result_reference")
 
-            if not result_reference_url:
-                logger.warning("Not found result_reference in message，ignore.")
-                return
-
-            conf = get_config()
-            if not conf:
-                sign_result_loc = DEFAULT_SIGN_RESULT_LOC
-            sign_result_loc = os.getenv("SIGN_RESULT_LOC") or conf.get_sign_result_loc()
-            logger.info("Using SIGN RESULT LOC: %s", sign_result_loc)
+        if not result_reference_url:
+            logger.warning("Not found result_reference in message，ignore.")
+            return
 
-            sign_result_parent_dir = os.path.dirname(sign_result_loc)
-            os.makedirs(sign_result_parent_dir, exist_ok=True)
+        logger.info("Using SIGN RESULT LOC: %s", self.sign_result_loc)
+        sign_result_parent_dir = os.path.dirname(self.sign_result_loc)
+        os.makedirs(sign_result_parent_dir, exist_ok=True)
 
-            oras_client = OrasClient()
-            files = oras_client.pull(
-                result_reference_url=result_reference_url, sign_result_loc=sign_result_loc
-            )
-            SignHandler.set_downloaded_files(files)
-        finally:
-            SignHandler.set_processing(False)
+        oras_client = OrasClient()
+        files = oras_client.pull(
+            result_reference_url=result_reference_url, sign_result_loc=self.sign_result_loc
+        )
+        logger.info("Number of files pulled: %d, path: %s", len(files), files[0])
 
 
-def generate_radas_sign(top_level: str) -> Tuple[List[str], List[str]]:
+def generate_radas_sign(top_level: str, sign_result_loc: str) -> Tuple[List[str], List[str]]:
     """
     Generate .asc files based on RADAS sign result json file
     """
@@ -142,17 +115,30 @@ def generate_radas_sign(top_level: str) -> Tuple[List[str], List[str]]:
         else DEFAULT_RADAS_SIGN_TIMEOUT_RETRY_INTERVAL
     )
     wait_count = 0
-    while SignHandler.is_processing():
+
+    # Wait until files appear in the sign_result_loc directory
+    while True:
+        files = [
+            os.path.join(sign_result_loc, f)
+            for f in os.listdir(sign_result_loc)
+            if os.path.isfile(os.path.join(sign_result_loc, f))
+        ]
+        if files:  # If files exist, break the loop
+            break
+
         wait_count += 1
         if wait_count > timeout_retry_count:
             logger.warning("Timeout when waiting for sign response.")
             break
         time.sleep(timeout_retry_interval)
 
-    files = SignHandler.get_downloaded_files()
     if not files:
         return [], []
 
+    if len(files) > 1:
+        logger.error("Multiple files found in %s. Expected only one file.", sign_result_loc)
+        return [], []
+
     # should only have the single sign result json file from the radas registry
     json_file_path = files[0]
     try: