Azure · anjaliratnam-msft · Sep 2, 2026 · Sep 4, 2026 · Jacob Lauzon (jalauzon-msft) · Sep 2, 2026
@@ -4,6 +4,7 @@
 
 ### Features Added
 - Added `list` support to `BlobSasPermissions` for use with directory-scoped SAS tokens.
+- Added opt-in client session-based authentication via the new `use_session` keyword argument. When enabled, eligible blob download requests are authenticated with a short-lived, per-container session credential obtained from the service rather than the bearer token. Requires a `TokenCredential`. Sessions are managed by a session provider, which can be shared across clients via the `session_provider` keyword, and the account name used for signing can be set explicitly with `session_account_name`.
 
 ### Bugs Fixed
 - Fixed an issue where `destination_snapshot` on a blob's copy properties was always `None` when listing blobs with `response_format="arrow"`.

@@ -26,6 +26,7 @@
     UserDelegationKey,
     Services,
 )
+from ._shared.session import ContainerSessionProvider, Session, SessionProvider
 from ._generated.models import RehydratePriority
 from ._models import (
     BlobType,
@@ -267,4 +268,7 @@ def download_blob_from_url(
     "ObjectReplicationPolicy",
     "ObjectReplicationRule",
     "Services",
+    "Session",
+    "SessionProvider",
+    "ContainerSessionProvider",
 ]
@@ -90,6 +90,19 @@ class BlobServiceClient(StorageAccountHostsMixin, StorageEncryptionMixin):
     :keyword str audience: The audience to use when requesting tokens for Azure Active Directory
         authentication. Only has an effect when credential is of type TokenCredential. The value could be
         https://storage.azure.com/ (default) or https://<account>.blob.core.windows.net.
+    :keyword bool use_session: If True, enable session-based authentication for this container.
+        When enabled, eligible GET requests issued by this client will be authenticated using
+        a short-lived session credential obtained from the service instead
+        of the provided TokenCredential. Only supported with a TokenCredential;
+        ValueError is raised otherwise. Defaults to False.
+    :keyword session_provider: Creates, caches, and invalidates the session credentials used for
+        session-based authentication. Supply a shared instance to reuse the session cache across
+        multiple clients; when omitted, one is created and scoped to this client. Only has an
+        effect when `use_session` is True.
+    :paramtype session_provider: ~azure.storage.blob.SessionProvider
+    :keyword str session_account_name: The storage account name used to sign session-authenticated
+        requests. If omitted, it is derived from the account URL. Required when using a custom
+        endpoint, where the account name cannot be determined from the URL.
 
     .. admonition:: Example:
 

@@ -14,6 +14,7 @@ from typing import (
     Optional,
     Union,
 )
+from azure.storage.blob._shared.session import SessionProvider
 from typing_extensions import Self
 
 from azure.core import MatchConditions
@@ -56,6 +57,9 @@ class BlobServiceClient(StorageAccountHostsMixin, StorageEncryptionMixin):
         max_single_get_size: int = 32 * 1024 * 1024,
         max_chunk_get_size: int = 4 * 1024 * 1024,
         audience: Optional[str] = None,
+        use_session: bool = False,
+        session_provider: Optional[SessionProvider] = None,
+        session_account_name: Optional[str] = None,
         **kwargs: Any
     ) -> None: ...
     def __enter__(self) -> Self: ...

@@ -103,6 +103,19 @@ class ContainerClient(StorageAccountHostsMixin, StorageEncryptionMixin):  # pyli
     :keyword str audience: The audience to use when requesting tokens for Azure Active Directory
         authentication. Only has an effect when credential is of type TokenCredential. The value could be
         https://storage.azure.com/ (default) or https://<account>.blob.core.windows.net.
+    :keyword bool use_session: If True, enable session-based authentication for this container.
+        When enabled, eligible GET requests issued by this client will be authenticated using
+        a short-lived session credential obtained from the service instead
+        of the provided TokenCredential. Only supported with a TokenCredential;
+        ValueError is raised otherwise. Defaults to False.
+    :keyword session_provider: Creates, caches, and invalidates the session credentials used for
+        session-based authentication. Supply a shared instance to reuse the session cache across
+        multiple clients; when omitted, one is created and scoped to this client. Only has an
+        effect when `use_session` is True.
+    :paramtype session_provider: ~azure.storage.blob.SessionProvider
+    :keyword str session_account_name: The storage account name used to sign session-authenticated
+        requests. If omitted, it is derived from the account URL. Required when using a custom
+        endpoint, where the account name cannot be determined from the URL.
 
     .. admonition:: Example:
 

@@ -28,6 +28,7 @@ from azure.core.credentials import AzureNamedKeyCredential, AzureSasCredential,
 from azure.core.paging import ItemPaged
 from azure.core.pipeline.transport import HttpResponse
 from azure.core.tracing.decorator import distributed_trace
+from azure.storage.blob._shared.session import SessionProvider
 from ._blob_client import BlobClient
 from ._blob_service_client import BlobServiceClient
 from ._download import StorageStreamDownloader
@@ -71,6 +72,9 @@ class ContainerClient(StorageAccountHostsMixin, StorageEncryptionMixin):
         max_single_get_size: int = 32 * 1024 * 1024,
         min_large_block_upload_threshold: int = 4 * 1024 * 1024 + 1,
         use_byte_buffer: Optional[bool] = None,
+        use_session: bool = False,
+        session_provider: Optional[SessionProvider] = None,
+        session_account_name: Optional[str] = None,
         **kwargs: Any,
     ) -> None: ...
     def __enter__(self) -> Self: ...

@@ -57,12 +57,14 @@
     StorageRequestHook,
     StorageResponseHook,
     StorageSensitiveHeaderCleanupPolicy,
+    StorageSessionPolicy,
 )
 from .request_handlers import serialize_batch_body, _get_batch_request_delimiter
 from .response_handlers import PartialBatchErrorException, process_storage_error
 from .shared_access_signature import QueryStringConstants
 from .._version import VERSION
 from .._shared_access_signature import _is_credential_sastoken
+from .session import ContainerSessionProvider
 
 if TYPE_CHECKING:
     from azure.core.credentials_async import AsyncTokenCredential
@@ -295,12 +297,13 @@ def _create_pipeline(
         **kwargs: Any,
     ) -> Tuple[StorageConfiguration, Pipeline]:
         self._credential_policy: Any = None
+        audience = kwargs.pop("audience", None)
         if hasattr(credential, "get_token"):
-            if kwargs.get("audience"):
-                audience = str(kwargs.pop("audience")).rstrip("/") + DEFAULT_OAUTH_SCOPE
+            if audience:
+                scope = str(audience).rstrip("/") + DEFAULT_OAUTH_SCOPE
             else:
-                audience = STORAGE_OAUTH_SCOPE
-            self._credential_policy = StorageBearerTokenCredentialPolicy(cast(TokenCredential, credential), audience)
+                scope = STORAGE_OAUTH_SCOPE
+            self._credential_policy = StorageBearerTokenCredentialPolicy(cast(TokenCredential, credential), scope)
         elif isinstance(credential, SharedKeyCredentialPolicy):
             self._credential_policy = credential
         elif isinstance(credential, AzureSasCredential):
@@ -330,12 +333,38 @@ def _create_pipeline(
             config.headers_policy,
             StorageRequestHook(**kwargs),
             self._credential_policy,
-            config.logging_policy,
-            StorageResponseHook(**kwargs),
-            DistributedTracingPolicy(**kwargs),
-            HttpLoggingPolicy(**kwargs),
             StorageSensitiveHeaderCleanupPolicy(**kwargs),
         ]
+        use_session = bool(kwargs.pop("use_session", False))
+        session_provider = kwargs.pop("session_provider", None)
+        session_account_name = kwargs.pop("session_account_name", None)
+        if use_session:
+            if session_provider is None:
+                sub_kwargs = dict(kwargs)
+                sub_kwargs.pop("_configuration", None)
+                sub_kwargs.pop("pipeline", None)
+                sub_kwargs["transport"] = transport
+                session_provider = ContainerSessionProvider(
+                    f"{self.scheme}://{self.primary_hostname}",
+                    cast(TokenCredential, credential),
+                    audience=audience,
+                    **sub_kwargs,
+                )
+
+            policies.append(
+                StorageSessionPolicy(
+                    account_name=session_account_name or self.account_name,
+                    session_provider=session_provider,
+                )
+            )
+        policies.extend(
+            [
+                config.logging_policy,
+                StorageResponseHook(**kwargs),
+                DistributedTracingPolicy(**kwargs),
+                HttpLoggingPolicy(**kwargs),
+            ]
+        )
         if kwargs.get("_additional_pipeline_policies"):
             policies = policies + kwargs.get("_additional_pipeline_policies")  # type: ignore
         config.transport = transport  # type: ignore

@@ -34,6 +34,7 @@ class StorageErrorCode(str, Enum, metaclass=CaseInsensitiveEnumMeta):
     CONDITION_HEADERS_NOT_SUPPORTED = "ConditionHeadersNotSupported"
     CONDITION_NOT_MET = "ConditionNotMet"
     EMPTY_METADATA_KEY = "EmptyMetadataKey"
+    FEATURE_NOT_ENABLED = "FeatureNotEnabled"
     INSUFFICIENT_ACCOUNT_PERMISSIONS = "InsufficientAccountPermissions"
     INTERNAL_ERROR = "InternalError"
     INVALID_AUTHENTICATION_INFO = "InvalidAuthenticationInfo"
@@ -64,6 +65,7 @@ class StorageErrorCode(str, Enum, metaclass=CaseInsensitiveEnumMeta):
     RESOURCE_ALREADY_EXISTS = "ResourceAlreadyExists"
     RESOURCE_NOT_FOUND = "ResourceNotFound"
     SERVER_BUSY = "ServerBusy"
+    SESSIONS_UNAVAILABLE = "SessionOperationsTemporarilyUnavailable"
     UNSUPPORTED_HEADER = "UnsupportedHeader"
     UNSUPPORTED_XML_NODE = "UnsupportedXmlNode"
     UNSUPPORTED_QUERY_PARAMETER = "UnsupportedQueryParameter"

@@ -9,11 +9,13 @@
 import random
 import re
 import uuid
+from datetime import timezone
 from io import BytesIO, SEEK_SET, UnsupportedOperation
 from time import time
-from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union
+from typing import Any, Dict, List, Optional, Tuple, TYPE_CHECKING, Union
 from urllib.parse import (
     parse_qsl,
+    unquote,
     urlencode,
     urlparse,
     urlunparse,
@@ -30,9 +32,11 @@
     SansIOHTTPPolicy,
 )
 
-from .authentication import AzureSigningError, StorageHttpChallenge
+from . import sign_string
+from .authentication import AzureSigningError, _storage_header_sort, StorageHttpChallenge
 from .constants import DEFAULT_OAUTH_SCOPE, DATA_BLOCK_SIZE
 from .models import LocationMode, StorageErrorCode
+from .session import Session, SessionProvider
 from .streams import (
     StructuredMessageDecoder,
     StructuredMessageEncodeStream,
@@ -55,12 +59,26 @@
 
 
 _LOGGER = logging.getLogger(__name__)
+_SESSION_SIGNED_HEADERS = (
+    "content-encoding",
+    "content-language",
+    "content-length",
+    "content-md5",
+    "content-type",
+    "date",
+    "if-modified-since",
+    "if-match",
+    "if-none-match",
+    "if-unmodified-since",
+    "byte_range",
+)
 CONTENT_LENGTH_HEADER = "Content-Length"
 MD5_HEADER = "Content-MD5"
 CRC64_HEADER = "x-ms-content-crc64"
 SM_HEADER = "x-ms-structured-body"
 SM_HEADER_V1_CRC64 = "XSM/1.0; properties=crc64"
 SM_LENGTH_HEADER = "x-ms-structured-content-length"
+SESSION_BEARER_AUTH_KEY = "_session_bearer_auth"
 
 
 def encode_base64(data: Union[bytes, str]) -> str:
@@ -70,6 +88,50 @@ def encode_base64(data: Union[bytes, str]) -> str:
     return encoded.decode("utf-8")
 
 
+def _apply_session_auth(
+    request: "PipelineRequest", session_token: str, session_key: str, account_name: str
+) -> None:
+    """Sign an eligible request with the SharedKey protocol under the Session scheme.
+
+    Shared by the sync and async session policies; ``account_name`` is passed in
+    rather than read from ``self`` so neither policy needs to instantiate the other.
+
+    :param ~azure.core.pipeline.PipelineRequest request: The request to sign in place.
+    :param str session_token: The session token to embed in the Authorization header.
+    :param str session_key: The HMAC signing key for the session.
+    :param str account_name: Storage account name; the signer identity.
+    :raises ~azure.storage.blob._shared.authentication.AzureSigningError: if signing fails.
+    """
+    http_request = request.http_request
+    http_request.headers["x-ms-date"] = format_date_time(time())
+
+    # 1) Standard headers. Storage omits content-length when it is "0".
+    headers = {name.lower(): value for name, value in http_request.headers.items() if value}
+    if headers.get("content-length") == "0":
+        del headers["content-length"]
+    signed_headers = "\n".join(headers.get(h, "") for h in _SESSION_SIGNED_HEADERS) + "\n"
+
+    # 2) Canonicalized x-ms-* headers, sorted by the service-emulating comparator.
+    x_ms_headers = _storage_header_sort(
+        [(n.lower(), v) for n, v in http_request.headers.items() if n.lower().startswith("x-ms-")]
+    )
+    canonicalized_headers = "".join(f"{n}:{v}\n" for n, v in x_ms_headers if v is not None)
+
+    # 3) Canonicalized resource + query (query values must be url-decoded).
+    canonicalized_resource = "/" + account_name + urlparse(http_request.url).path
+    canonicalized_resource += "".join(
+        f"\n{n.lower()}:{unquote(v)}" for n, v in sorted(http_request.query.items()) if v is not None
+    )
+
+    string_to_sign = http_request.method + "\n" + signed_headers + canonicalized_headers + canonicalized_resource
+
+    try:
+        signature = sign_string(session_key, string_to_sign)
+    except Exception as ex:  # pylint: disable=broad-except
+        raise AzureSigningError(str(ex)) from ex
+    http_request.headers["Authorization"] = f"Session {session_token}:{signature}"
+
+
 # Are we out of retries?
 def is_exhausted(settings):
     retry_counts = (
@@ -927,3 +989,95 @@ def on_request(self, request: "PipelineRequest") -> None:
             # Clean up request headers
             for header in self._blocked_redirect_headers:
                 request.http_request.headers.pop(header, None)
+
+
+class StorageSessionPolicy(HTTPPolicy):
+    """
+    A pipeline policy that selects between session token and bearer token authentication.
+
+    Eligible requests are authenticated with a session token obtained from the
+    session provider. Everything else is left to the bearer token policy that
+    sits earlier in the pipeline.
+    """
+
+    def __init__(
+        self,
+        *,
+        account_name: Optional[str],
+        session_provider: SessionProvider,
+    ) -> None:
+        """Constructs a StorageSessionPolicy.
+
+        :keyword str account_name: Storage account name; used as the signer
+            identity when signing session-authenticated requests.
+        :keyword session_provider: Creates, caches, and invalidates per-container sessions.
+        :paramtype session_provider: ~azure.storage.blob._shared.session.SessionProvider
+        :raises ValueError: if `account_name` is `None`.
+        """
+        if account_name is None:
+            raise ValueError(
+                "Unable to determine the account name from the service URL. "
+                "Supply session_account_name when using a custom endpoint."
+            )
+        super().__init__()
+        self._account_name = account_name
+        self._session_provider = session_provider
+
+    def send(self, request: "PipelineRequest") -> "PipelineResponse":
+        """Orchestrate session auth.
+
+        :param ~azure.core.pipeline.PipelineRequest request: The outgoing request.
+        :return: The pipeline response.
+        :rtype: ~azure.core.pipeline.PipelineResponse
+        """
+        session = self.on_request(request)
+        response = self.next.send(request)
+        return self.on_response(request, response, session)
+
+    def on_request(self, request: "PipelineRequest") -> Optional[Session]:
+        """Stamp session auth if eligible, otherwise leave the bearer header intact.
+
+        :param ~azure.core.pipeline.PipelineRequest request: The request to (maybe) sign.
+        :return: The session that was applied, else None.
+        :rtype: ~azure.storage.blob._shared.session.Session or None
+        """
+        session = self._session_provider.get_session(request)
+        if session is None or not session.session_token or not session.session_key:
+            return None
+
+        # Kept so a 401 can serve this request with bearer instead of re-signing.
+        request.context.options[SESSION_BEARER_AUTH_KEY] = request.http_request.headers.get("Authorization")
+        _apply_session_auth(request, session.session_token, session.session_key, self._account_name)
+        return session
+
+    def on_response(
+        self,
+        request: "PipelineRequest",
+        response: "PipelineResponse",
+        session: Optional[Session],
+    ) -> "PipelineResponse":
+        """On 401, invalidate the cached session and serve the request with bearer.
+
+        :param ~azure.core.pipeline.PipelineRequest request: The original request.
+        :param ~azure.core.pipeline.PipelineResponse response: The response to inspect.
+        :param session: The session that signed the request, or `None` if bearer was used.
+        :type session: ~azure.storage.blob._shared.session.Session or None
+        :return: The final response.
+        :rtype: ~azure.core.pipeline.PipelineResponse
+        """
+        if session is None:
+            return response  # bearer was used; nothing session-related to react to
+
+        status = response.http_response.status_code
+
+        # 401 → drop the cached session and serve this request with bearer.
+        if status == 401:
+            _LOGGER.info("Session authentication: HTTP 401; invalidating session and retrying with bearer.")
+            self._session_provider.invalidate_session(request, session)
+            bearer = request.context.options.get(SESSION_BEARER_AUTH_KEY)
+            if bearer:
+                request.http_request.headers["Authorization"] = bearer
+                return self.next.send(request)
+            return response
+
+        return response