feat: Enable automatic cloud registration v2 again

After being reverted for release in 8b25a85 and 548a279, we can enable it again.
candlepin · Sep 10, 2024 · 12b3bda · 12b3bda
1 parent 1ed56ac
commit 12b3bda
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 12 deletions.
diff --git a/src/rhsm/connection.py b/src/rhsm/connection.py
@@ -1515,11 +1515,10 @@ def getCloudJWT(self, cloud_id: str, metadata: str, signature: str) -> Dict[str,
         }
         headers = {
             "Content-Type": "application/json",
-            "Accept": "text/plain",
         }
 
         return self.conn.request_post(
-            method="/cloud/authorize",
+            method="/cloud/authorize?version=2",
             params=data,
             headers=headers,
             description=_("Fetching cloud token"),

diff --git a/src/subscription_manager/scripts/rhsmcertd_worker.py b/src/subscription_manager/scripts/rhsmcertd_worker.py
@@ -191,7 +191,7 @@ def _auto_register(cp_provider: "CPProvider") -> ExitStatus:
 
     # Obtain automatic registration token
     try:
-        token: Dict[str, str] = cache.CloudTokenCache._get_from_server(
+        token: Dict[str, str] = cache.CloudTokenCache.get(
             uep=uep,
             cloud_id=cloud_info["cloud_id"],
             metadata=cloud_info["metadata"],
@@ -201,14 +201,29 @@ def _auto_register(cp_provider: "CPProvider") -> ExitStatus:
         log.exception("Cloud token could not be obtained. Unable to perform automatic registration.")
         return ExitStatus.NO_REGISTRATION_TOKEN
 
-    try:
-        _auto_register_standard(uep=uep, token=token)
-    except Exception:
-        log.exception("Standard automatic registration failed.")
-        return ExitStatus.REGISTRATION_FAILED
-    else:
-        log.info("Standard automatic registration was successful.")
-        return ExitStatus.OK
+    if token["tokenType"] == "CP-Cloud-Registration":
+        try:
+            _auto_register_standard(uep=uep, token=token)
+        except Exception:
+            log.exception("Standard automatic registration failed.")
+            return ExitStatus.REGISTRATION_FAILED
+        else:
+            log.info("Standard automatic registration was successful.")
+            return ExitStatus.OK
+
+    if token["tokenType"] == "CP-Anonymous-Cloud-Registration":
+        try:
+            _auto_register_anonymous(uep=uep, token=token)
+            cache.CloudTokenCache.delete_cache()
+        except Exception:
+            log.exception("Anonymous automatic registration failed.")
+            return ExitStatus.REGISTRATION_FAILED
+        else:
+            log.info("Anonymous automatic registration was successful.")
+            return ExitStatus.OK
+
+    log.error(f"Unsupported token type for automatic registration: {token['tokenType']}.")
+    return ExitStatus.BAD_TOKEN_TYPE
 
 
 def _auto_register_standard(uep: "UEPConnection", token: Dict[str, str]) -> None:
@@ -221,7 +236,7 @@ def _auto_register_standard(uep: "UEPConnection", token: Dict[str, str]) -> None
     log.debug("Registering the system through standard automatic registration.")
 
     service = RegisterService(cp=uep)
-    service.register(org=None, jwt_token=token)
+    service.register(org=None, jwt_token=token["token"])
 
 
 def _auto_register_anonymous(uep: "UEPConnection", token: Dict[str, str]) -> None: