Truncate Wasb storage account name if it's more than 24 characters

airflow/providers/microsoft/azure/hooks/wasb.py

@@ -163,10 +163,18 @@ def get_conn(self) -> BlobServiceClient:
         account_url = conn.host if conn.host else f"https://{conn.login}.blob.core.windows.net/"
         parsed_url = urlparse(account_url)
 
-        if not parsed_url.netloc and "." not in parsed_url.path:
-            # if there's no netloc and no dots in the path, then user only
-            # provided the Active Directory ID, not the full URL or DNS name
-            account_url = f"https://{conn.login}.blob.core.windows.net/"
+        if not parsed_url.netloc:
+            if "." not in parsed_url.path:
+                # if there's no netloc and no dots in the path, then user only
+                # provided the Active Directory ID, not the full URL or DNS name
+                account_url = f"https://{conn.login}.blob.core.windows.net/"
+            else:
+                # if there's no netloc but there are dots in the path, then user
+                # provided the DNS name without the https:// prefix.
+                # Azure storage account name can only be 3 to 24 characters in length
+                # https://learn.microsoft.com/en-us/azure/storage/common/storage-account-overview#storage-account-name
+                acc_name = account_url.split(".")[0][:24]
+                account_url = f"https://{acc_name}." + ".".join(account_url.split(".")[1:])
 
         tenant = self._get_field(extra, "tenant_id")
         if tenant:
@@ -568,10 +576,18 @@ async def get_async_conn(self) -> AsyncBlobServiceClient:
         account_url = conn.host if conn.host else f"https://{conn.login}.blob.core.windows.net/"
         parsed_url = urlparse(account_url)
 
-        if not parsed_url.netloc and "." not in parsed_url.path:
-            # if there's no netloc and no dots in the path, then user only
-            # provided the Active Directory ID, not the full URL or DNS name
-            account_url = f"https://{conn.login}.blob.core.windows.net/"
+        if not parsed_url.netloc:
+            if "." not in parsed_url.path:
+                # if there's no netloc and no dots in the path, then user only
+                # provided the Active Directory ID, not the full URL or DNS name
+                account_url = f"https://{conn.login}.blob.core.windows.net/"
+            else:
+                # if there's no netloc but there are dots in the path, then user
+                # provided the DNS name without the https:// prefix.
+                # Azure storage account name can only be 3 to 24 characters in length
+                # https://learn.microsoft.com/en-us/azure/storage/common/storage-account-overview#storage-account-name
+                acc_name = account_url.split(".")[0][:24]
+                account_url = f"https://{acc_name}." + ".".join(account_url.split(".")[1:])
 
         tenant = self._get_field(extra, "tenant_id")
         if tenant:

tests/providers/microsoft/azure/hooks/test_wasb.py

@@ -346,8 +346,12 @@ def test_extra_client_secret_auth_config_ad_connection(self):
                 "https://testaccountname.blob.core.windows.net",
             ),
             ("testhost", "https://accountlogin.blob.core.windows.net/"),
-            ("testhost.dns", "testhost.dns"),
-            ("testhost.blob.net", "testhost.blob.net"),
+            ("testhost.dns", "https://testhost.dns"),
+            ("testhost.blob.net", "https://testhost.blob.net"),
+            (
+                "testhostakjhdisdfbearioyo.blob.core.windows.net",
+                "https://testhostakjhdisdfbearioy.blob.core.windows.net",
+            ),  # more than 24 characters
         ],
     )
     def test_proper_account_url_update(