sign in radas function implementation init

Author: shokakucarrier

charon/cmd/cmd_sign.py

@@ -15,7 +15,11 @@
 """
 from typing import List
 
+<<<<<<< HEAD
 from charon.config import get_config
+=======
+from charon.config import get_config, RadasConfig
+>>>>>>> c4ad584 (sign in radas function implementation init)
 from charon.pkgs.radas_signature_handler import sign_in_radas
 from charon.cmd.internal import (
     _decide_mode, _safe_delete

charon/pkgs/radas_signature_handler.py

@@ -19,11 +19,13 @@
 import os
 import asyncio
 import sys
+import uuid
 from typing import List, Any, Tuple, Callable, Dict
 from charon.config import get_config, RadasConfig
 from charon.pkgs.oras_client import OrasClient
-from proton import Event
+from proton import SSLDomain, SSLException, Message, Event
 from proton.handlers import MessagingHandler
+from proton.utils import BlockingConnection
 
 logger = logging.getLogger(__name__)
 
@@ -111,6 +113,59 @@ def _process_message(self, msg: Any) -> None:
         logger.info("Number of files pulled: %d, path: %s", len(files), files[0])
 
 
+def sign_in_radas(repo_url: str,
+                  requester: str,
+                  sign_key: str,
+                  result_path: str,
+                  ignore_patterns: List[str],
+                  radas_config: RadasConfig):
+    """
+    This function will be responsible to do the overall controlling of the whole process,
+    like trigger the send and register the receiver, and control the wait and timeout there.
+    """
+    logger.debug("params. repo_url: %s, requester: %s, sign_key: %s, result_path: %s,"
+                 "radas_config: %s", repo_url, requester, sign_key, result_path, radas_config)
+    request_id = str(uuid.uuid4())
+    exclude = list(ignore_patterns) if ignore_patterns else []
+    
+    payload = {
+        "request_id": request_id,
+        "requested_by": requester,
+        "type": "mrrc",
+        "file_reference": repo_url,
+        "sig_keyname": sign_key,
+        "exclude": exclude
+    }
+
+    try:
+        ssl_domain = SSLDomain(SSLDomain.MODE_CLIENT)
+        ssl_domain.set_credentials(
+            radas_config.client_ca(),
+            radas_config.client_key(),
+            radas_config.client_key_password()
+        )
+        ssl_domain.set_trusted_ca_db(radas_config.root_ca())
+        ssl_domain.set_peer_authentication(SSLDomain.VERIFY_PEER)
+
+        conn = BlockingConnection(radas_config.umb_target(), ssl_domain=ssl_domain)
+        try:
+            sender = conn.create_sender(radas_config.request_queue())
+            message = Message(body=json.dumps(payload))
+            sender.send(message)
+            logger.info("Successfully sent signing request ID: %s", request_id)
+        finally:
+            conn.close()
+            
+    except SSLException as e:
+        logger.error("SSL connection failed: %s", str(e))
+        sys.exit(1)
+    except Exception as e:
+        logger.error("Failed to send signing request: %s", str(e))
+        sys.exit(1)
+
+    # wait for AMQP message to be consumed then get response message from UMB
+
+
 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