Vinothini Dharmaraj (v-dharmarajv) · Oct 26, 2026 · Oct 23, 2026 · Oct 26, 2026
@@ -1,5 +1,15 @@
 # Release History
 
+## 1.6.1 (2026-10-25)
+
+### Bugs Fixed
+
+- Security: Hardened recording URL validation against ambiguous authorities that could send authenticated requests to unintended endpoints. Synchronous and asynchronous recording downloads and deletes now validate each redirect destination before authentication.
+
+### Other Changes
+
+- Deprecated version 1.6.0 as of 2026-10-22 due to the recording URL validation security issue. Upgrade to 1.6.1 or a later fixed release to receive the security fix. This deprecation applies to version 1.6.0 only, not to the library as a whole.
+
 ## 1.6.0 (2026-06-24)
 
 ### Features Added

@@ -8,6 +8,26 @@ This package contains a Python SDK for Azure Communication Call Automation. Call
 _Azure SDK Python packages support for Python 2.7 has ended 01 January 2022. For more information and questions, please
 refer to https://github.com/Azure/azure-sdk-for-python/issues/20691_
 
+## Security notice: version 1.6.0 deprecated
+
+Version **1.6.0** of `azure-communication-callautomation` is deprecated as of
+**2026-10-22** because of a recording URL validation security issue. This notice
+applies to version 1.6.0, not to the library as a whole.
+
+In applications that pass attacker-influenced recording URLs to the synchronous
+`download_recording` or `delete_recording` methods using the default Requests
+transport, an ambiguous URL can bypass hostname validation and send an
+authenticated request to an unintended endpoint, potentially exposing a Bearer
+token.
+
+Upgrade to **1.6.1** or a later fixed release to receive the security fix.
+Use only service-generated recording URLs received through trusted,
+integrity-protected events; do not pass untrusted recording URLs to these methods.
+
+Refer to the [release history](CHANGELOG.md) and
+[Azure SDK support policy](https://azure.github.io/azure-sdk/policies_support.html)
+for more information.
+
 ## Getting started
 ### Prerequisites
 - Python 3.8 or later is required to use this package.
@@ -29,6 +49,17 @@ pip install azure-communication-callautomation
 | Callback Events      | Callback events are events sent back during duration of the call. It gives information and state of the call, such as `CallConnected`. `CallbackUrl` must be provided during `createCall` and `answerCall`, and callback events will be sent to this url. |
 | Incoming Call Event  | When incoming call happens (that can be answered with `answerCall`), incoming call eventgrid event will be sent. This is different from Callback events above, and should be setup on Azure portal. See [Incoming Call][incomingcall] for detail.                                                                                        |
 
+### Recording URL security
+
+`download_recording` and `delete_recording` accept only HTTPS URLs with ASCII DNS
+hostnames under `*.asm.skype.com`, `*.asyncgw.teams.microsoft.com`, or
+`*.ams.gov.teams.microsoft.us`. URLs containing user information, raw backslashes,
+whitespace, control characters, or malformed authorities are rejected with
+`ValueError` before authentication. Redirect destinations are subject to the same
+validation. Valid signed query parameters are preserved.
+
+Use service-generated recording URLs from trusted, integrity-protected events.
+
 ## Examples
 ### Initialize CallAutomationClient
 ```Python

@@ -15,6 +15,7 @@
 from ._call_connection_client import CallConnectionClient
 from ._generated._client import AzureCommunicationCallAutomationService
 from ._shared.auth_policy_utils import get_authentication_policy
+from .recording_url_validator import RecordingURLValidationPolicy
 from ._shared.utils import parse_connection_str
 from ._shared.models import CommunicationIdentifier
 from ._generated.models import (
@@ -107,6 +108,11 @@ def __init__(
         if not parsed_url.netloc:
             raise ValueError(f"Invalid URL: {format(endpoint)}")
 
+        per_retry_policies = kwargs.pop("per_retry_policies", None) or []
+        if not isinstance(per_retry_policies, Iterable):
+            per_retry_policies = [per_retry_policies]
+        kwargs["per_retry_policies"] = [*per_retry_policies, RecordingURLValidationPolicy()]
+
         self._client = AzureCommunicationCallAutomationService(
             endpoint,
             credential,  # type: ignore[arg-type]

@@ -66,7 +66,8 @@ def download_streaming(
             raise ValueError("Recording client endpoint must not be None.")
 
         # Validate recording URL before sending authenticated request
-        validate_recording_url(source_location, "source_location")
+        source_location = validate_recording_url(source_location, "source_location")
+        kwargs["_recording_url_parameter_name"] = "source_location"
 
         _headers = kwargs.pop("headers", {}) or {}
         _params = kwargs.pop("params", {}) or {}
@@ -116,7 +117,8 @@ def delete_recording(self, recording_location: str, **kwargs: Any) -> None:
             raise ValueError("Recording client endpoint must not be None.")
 
         # Validate recording URL before sending authenticated request
-        validate_recording_url(recording_location, "recording_location")
+        recording_location = validate_recording_url(recording_location, "recording_location")
+        kwargs["_recording_url_parameter_name"] = "recording_location"
 
         _headers = kwargs.pop("headers", {}) or {}
         _params = kwargs.pop("params", {}) or {}

@@ -4,6 +4,6 @@
 # license information.
 # --------------------------------------------------------------------------
 
-VERSION = "1.6.0"
+VERSION = "1.6.1"
 
 SDK_MONIKER = f"communication-callautomation/{format(VERSION)}"
@@ -3,7 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
-from typing import Sequence, Union, Optional, TYPE_CHECKING, AsyncIterable, overload
+from typing import Sequence, Union, Optional, TYPE_CHECKING, AsyncIterable, Iterable, overload
 from urllib.parse import urlparse
 import warnings
 
@@ -15,6 +15,7 @@
 from ._call_connection_client_async import CallConnectionClient
 from .._generated.aio import AzureCommunicationCallAutomationService
 from .._shared.auth_policy_utils import get_authentication_policy
+from ..recording_url_validator import RecordingURLValidationPolicy
 from .._shared.utils import parse_connection_str
 from .._shared.models import CommunicationIdentifier
 from .._generated.models import (
@@ -108,6 +109,11 @@ def __init__(
         if not parsed_url.netloc:
             raise ValueError(f"Invalid URL: {format(endpoint)}")
 
+        per_retry_policies = kwargs.pop("per_retry_policies", None) or []
+        if not isinstance(per_retry_policies, Iterable):
+            per_retry_policies = [per_retry_policies]
+        kwargs["per_retry_policies"] = [*per_retry_policies, RecordingURLValidationPolicy()]
+
         self._client = AzureCommunicationCallAutomationService(
             endpoint,
             credential,  # type: ignore[arg-type]

@@ -67,7 +67,8 @@ async def download_streaming(
             raise ValueError("Recording client endpoint must not be None.")
 
         # Validate recording URL before sending authenticated request
-        validate_recording_url(source_location, "source_location")
+        source_location = validate_recording_url(source_location, "source_location")
+        kwargs["_recording_url_parameter_name"] = "source_location"
 
         _headers = kwargs.pop("headers", {}) or {}
         _params = kwargs.pop("params", {}) or {}
@@ -118,7 +119,8 @@ async def delete_recording(self, recording_location: str, **kwargs: Any) -> None
             raise ValueError("Recording client endpoint must not be None.")
 
         # Validate recording URL before sending authenticated request
-        validate_recording_url(recording_location, "recording_location")
+        recording_location = validate_recording_url(recording_location, "recording_location")
+        kwargs["_recording_url_parameter_name"] = "recording_location"
 
         _headers = kwargs.pop("headers", {}) or {}
         _params = kwargs.pop("params", {}) or {}

@@ -4,9 +4,12 @@
 # license information.
 # --------------------------------------------------------------------------
 
+import re
 from typing import Tuple
 from urllib.parse import urlparse
 
+from azure.core.pipeline import PipelineRequest, PipelineResponse
+from azure.core.pipeline.policies import SansIOHTTPPolicy
 
 # Allowed recording endpoint host suffixes.
 # These are the only domains permitted for recording URLs to prevent credential exfiltration.
@@ -17,7 +20,12 @@
 )
 
 
-def validate_recording_url(recording_url: str, parameter_name: str) -> None:
+def _validate_url_characters(url: str, parameter_name: str) -> None:
+    if any(char == "\\" or char.isspace() or ord(char) < 32 or 127 <= ord(char) <= 159 for char in url):
+        raise ValueError(f"{parameter_name} must not contain backslashes, whitespace, or control characters.")
+
+
+def validate_recording_url(recording_url: str, parameter_name: str) -> str:
     """
     Validate that a recording URL points to Azure Communication Services
     endpoint before credentials are attached.
@@ -27,12 +35,18 @@ def validate_recording_url(recording_url: str, parameter_name: str) -> None:
     :type recording_url: str
     :param parameter_name: The parameter name for exception messages.
     :type parameter_name: str
+    :return: The validated URL with a canonical ASCII authority.
+    :rtype: str
     :raises TypeError: If the recording URL is None or empty.
     :raises ValueError: If the recording URL is not a valid absolute HTTPS URI or not from an allowed domain.
     """
     if not recording_url:
         raise TypeError(f"{parameter_name} cannot be null or undefined.")
+    if not isinstance(recording_url, str):
+        raise TypeError(f"{parameter_name} must be a string.")
 
+    # Reject syntax that URL parsers may discard or interpret as a different authority.
+    _validate_url_characters(recording_url, parameter_name)
     parsed_url = urlparse(recording_url)
 
     # Validate it's a valid absolute URI
@@ -43,11 +57,22 @@ def validate_recording_url(recording_url: str, parameter_name: str) -> None:
     if parsed_url.scheme.lower() != "https":
         raise ValueError(f"{parameter_name} must use HTTPS scheme for security.")
 
+    if "@" in parsed_url.netloc or not parsed_url.netloc.isascii():
+        raise ValueError(f"{parameter_name} must have an ASCII hostname and must not contain user information.")
+
     host = parsed_url.hostname
     if not host:
         raise ValueError(f"{parameter_name} must have a valid hostname.")
 
     host_lower = host.lower()
+    if len(host_lower) > 253 or not all(
+        re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", label) for label in host_lower.split(".")
+    ):
+        raise ValueError(f"{parameter_name} must have a valid DNS hostname.")
+
+    port = parsed_url.port
+    if ":" in parsed_url.netloc and not parsed_url.netloc.rsplit(":", 1)[1].isdigit():
+        raise ValueError(f"{parameter_name} must have a valid port.")
 
     # Check against allowed suffixes
     is_valid_endpoint = any(host_lower.endswith(suffix) for suffix in ALLOWED_HOST_SUFFIXES)
@@ -57,3 +82,25 @@ def validate_recording_url(recording_url: str, parameter_name: str) -> None:
             f"{parameter_name} host '{host}' is not a valid Azure Communication Services recording endpoint. "
             "Only URLs pointing to *.asm.skype.com, *.asyncgw.teams.microsoft.com are allowed."
         )
+
+    authority = host_lower if port is None else f"{host_lower}:{port}"
+    return parsed_url._replace(scheme="https", netloc=authority).geturl()
+
+
+class RecordingURLValidationPolicy(SansIOHTTPPolicy):
+    """Validate recording destinations on every attempt, before authentication."""
+
+    def on_request(self, request: PipelineRequest) -> None:
+        parameter_name = request.context.options.pop("_recording_url_parameter_name", None)
+        if parameter_name is not None:
+            request.context["_recording_url_parameter_name"] = parameter_name
+        parameter_name = request.context.get("_recording_url_parameter_name")
+        if parameter_name:
+            request.http_request.url = validate_recording_url(request.http_request.url, parameter_name)
+
+    def on_response(self, request: PipelineRequest, response: PipelineResponse) -> None:
+        parameter_name = request.context.get("_recording_url_parameter_name")
+        location = response.http_response.headers.get("location")
+        if parameter_name and location and response.http_response.status_code in (300, 301, 302, 303, 307, 308):
+            # RedirectPolicy parses Location before the next on_request, potentially discarding controls.
+            _validate_url_characters(location, parameter_name)