SNOW-2183023: fixed user id

Author: sfc-gh-fpawlowski

src/snowflake/connector/wif_util.py

@@ -243,79 +243,6 @@ def get_aws_region(session_manager: SessionManager | None = None) -> str | None:
     return None
 
 
-def get_aws_arn(session_manager: SessionManager | None = None) -> str | None:
-    """Get the current AWS workload's ARN by calling GetCallerIdentity.
-
-    Note: This function makes a network call to AWS STS and is only used for
-    assertion content generation (logging and backward compatibility purposes).
-    The ARN is not required for authentication - it's just used as a unique
-    identifier for the workload in logs and assertion content.
-
-    Returns the ARN of the current AWS identity, or None if it cannot be determined.
-    """
-    credentials = get_aws_credentials(session_manager)
-    if not credentials:
-        logger.debug("No AWS credentials available for ARN lookup.")
-        return None
-
-    region = get_aws_region(session_manager)
-    if not region:
-        logger.debug("No AWS region available for ARN lookup.")
-        return None
-
-    try:
-        # Create the GetCallerIdentity request
-        sts_hostname = get_aws_sts_hostname(region)
-        url = f"https://{sts_hostname}/?Action=GetCallerIdentity&Version=2011-06-15"
-
-        base_headers = {
-            "Content-Type": "application/x-amz-json-1.1",
-        }
-
-        signed_headers = aws_signature_v4_sign(
-            credentials=credentials,
-            method="POST",
-            url=url,
-            region=region,
-            service="sts",
-            headers=base_headers,
-        )
-
-        # Make the actual request to get caller identity
-        response = http_request(
-            method="POST",
-            url=url,
-            headers=signed_headers,
-            timeout_sec=10,
-            session_manager=session_manager,
-        )
-
-        if response and response.ok:
-            # Parse the XML response to extract the ARN
-            import xml.etree.ElementTree as ET
-
-            # Ensure content is bytes and decode it
-            content = response.content
-            if isinstance(content, bytes):
-                content_str = content.decode("utf-8")
-            else:
-                content_str = str(content) if content else ""
-
-            if content_str:
-                root = ET.fromstring(content_str)
-
-                # Find the Arn element in the response
-                for elem in root.iter():
-                    if elem.tag.endswith("Arn") and elem.text:
-                        return elem.text.strip()
-
-        logger.debug("Failed to get ARN from GetCallerIdentity response.")
-        return None
-    except Exception as e:
-        logger.debug(f"Error getting AWS ARN: {e}")
-        return None
-
-
 def get_aws_sts_hostname(region: str) -> str | None:
     """Constructs the AWS STS hostname for a given region.
 
@@ -481,11 +408,7 @@ def create_aws_attestation(
     credential = b64encode(json.dumps(attestation_request).encode("utf-8")).decode(
         "utf-8"
     )
-
-    # Get the ARN for user identifier components (used only for assertion content - logging and backward compatibility)
-    # The ARN is not required for authentication, but provides a unique identifier for the workload
-    arn = get_aws_arn(session_manager)
-    user_identifier_components = {"arn": arn} if arn else {}
+    user_identifier_components = {"region": region}
 
     return WorkloadIdentityAttestation(
         AttestationProvider.AWS,

test/unit/test_auth_workload_identity.py

@@ -173,10 +173,11 @@ def test_explicit_aws_generates_unique_assertion_content(
     auth_class = AuthByWorkloadIdentity(provider=AttestationProvider.AWS)
     auth_class.prepare(conn=None)
 
-    assert (
-        '{"_provider":"AWS","arn":"arn:aws:sts::123456789:assumed-role/A-Different-Role/i-34afe100cad287fab"}'
-        == auth_class.assertion_content
-    )
+    expected = {
+        "_provider": "AWS",
+        "region": fake_aws_environment.region,
+    }
+    assert json.loads(auth_class.assertion_content) == expected
 
 
 @pytest.mark.parametrize(