Fix canonical management payload URLs

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 3ec5c077-6e8b-4cb3-916a-12bd3cb13937

Author: andystaples

azure-functions-durable/CHANGELOG.md

@@ -32,6 +32,9 @@ avoiding repeated allocation of unused worker resources.
 
 FIXED
 
+- HTTP management payloads now preserve the host-provided management URL
+templates, include `rewindPostUri`, encode instance IDs, and use forwarded
+request origins consistently in asynchronous and synchronous clients.
 - Fixed asynchronous durable-client construction failing after an application
 event loop had been closed or cleared.
 - Prevented Durable HTTP calls from forwarding managed identity tokens,

azure-functions-durable/azure/durable_functions/client.py

@@ -7,7 +7,7 @@
 import threading
 
 from datetime import datetime, timedelta
-from typing import Any, Optional, Union
+from typing import Any, Mapping, Optional, Union, cast
 from warnings import deprecated
 import azure.functions as func
 from urllib.parse import urlparse, quote
@@ -36,6 +36,68 @@
 _sync_client_cache_lock = threading.Lock()
 
 
+def _first_forwarded_value(value: str) -> str:
+    return value.split(",", 1)[0].strip().strip('"')
+
+
+def _get_request_origin(request: func.HttpRequest) -> str:
+    request_url = urlparse(request.url)
+    proto = request_url.scheme
+    host = request_url.netloc
+    request_headers = cast(Mapping[str, str], request.headers)
+    headers = {
+        name.lower(): value for name, value in request_headers.items()
+    }
+
+    forwarded = headers.get("forwarded")
+    if forwarded:
+        forwarded_values: dict[str, str] = {}
+        for pair in forwarded.split(",", 1)[0].split(";"):
+            name, separator, value = pair.partition("=")
+            if separator:
+                forwarded_values[name.strip().lower()] = value.strip().strip('"')
+
+        proto = forwarded_values.get("proto", proto)
+        forwarded_host = forwarded_values.get("host")
+        if forwarded_host:
+            return f"{proto}://{forwarded_host}"
+
+    forwarded_proto = headers.get("x-forwarded-proto")
+    if forwarded_proto:
+        proto = _first_forwarded_value(forwarded_proto)
+
+    forwarded_host = headers.get("x-forwarded-host")
+    if forwarded_host:
+        host = _first_forwarded_value(forwarded_host)
+
+    return f"{proto}://{host}"
+
+
+def _build_http_management_payload(
+        instance_id: str,
+        management_urls: dict[str, str],
+        base_url: str,
+        required_query_string_parameters: str,
+        request: func.HttpRequest | None) -> HttpManagementPayload:
+    encoded_instance_id = quote(instance_id, safe="")
+    request_origin: str | None = None
+    if request is not None:
+        request_origin = _get_request_origin(request)
+        instance_status_url = (
+            f"{request_origin}/runtime/webhooks/durabletask/instances/"
+            f"{encoded_instance_id}")
+    else:
+        instance_status_url = (
+            f"{base_url.rstrip('/')}/instances/{encoded_instance_id}")
+
+    return HttpManagementPayload(
+        instance_id,
+        instance_status_url,
+        required_query_string_parameters,
+        management_urls=management_urls,
+        request_origin=request_origin)
+
+
 # Client class used for Durable Functions
 class DurableFunctionsClient(AsyncTaskHubGrpcClient):
     """A gRPC client passed to Durable Functions durable client bindings.
@@ -215,21 +277,12 @@ def create_http_management_payload(
         return self._get_client_response_links(resolved_request, instance_id)
 
     def _get_client_response_links(self, request: func.HttpRequest | None, instance_id: str) -> HttpManagementPayload:
-        instance_status_url = self._get_instance_status_url(request, instance_id)
-        return HttpManagementPayload(instance_id, instance_status_url, self.requiredQueryStringParameters)
-
-    def _get_instance_status_url(self, request: func.HttpRequest | None, instance_id: str) -> str:
-        encoded_instance_id = quote(instance_id)
-        if request is not None:
-            request_url = urlparse(request.url)
-            location_url = f"{request_url.scheme}://{request_url.netloc}"
-            location_url = location_url + "/runtime/webhooks/durabletask/instances/" + encoded_instance_id
-        else:
-            # No request available (v1-style call): fall back to the base URL
-            # supplied in the client binding configuration.
-            base_url = self.baseUrl.rstrip("/") if self.baseUrl else ""
-            location_url = base_url + "/instances/" + encoded_instance_id
-        return location_url
+        return _build_http_management_payload(
+            instance_id,
+            self.managementUrls,
+            self.baseUrl,
+            self.requiredQueryStringParameters,
+            request)
 
     # ------------------------------------------------------------------
     # Backwards-compatibility shims for the v1 azure-functions-durable
@@ -593,21 +646,12 @@ def create_http_management_payload(
     def _get_client_response_links(
             self, request: func.HttpRequest | None,
             instance_id: str) -> HttpManagementPayload:
-        return HttpManagementPayload(
+        return _build_http_management_payload(
             instance_id,
-            self._get_instance_status_url(request, instance_id),
-            self.requiredQueryStringParameters)
-
-    def _get_instance_status_url(
-            self, request: func.HttpRequest | None, instance_id: str) -> str:
-        encoded_instance_id = quote(instance_id)
-        if request is not None:
-            request_url = urlparse(request.url)
-            return (
-                f"{request_url.scheme}://{request_url.netloc}"
-                f"/runtime/webhooks/durabletask/instances/{encoded_instance_id}")
-        base_url = self.baseUrl.rstrip("/") if self.baseUrl else ""
-        return f"{base_url}/instances/{encoded_instance_id}"
+            self.managementUrls,
+            self.baseUrl,
+            self.requiredQueryStringParameters,
+            request)
 
 
 def _close_cached_sync_clients() -> None:

azure-functions-durable/azure/durable_functions/http/http_management_payload.py

@@ -2,7 +2,11 @@
 # Licensed under the MIT License.
 
 import json
-from typing import Any
+from typing import Any, Mapping
+from urllib.parse import quote, urlsplit, urlunsplit
+
+
+_INSTANCE_ID_PLACEHOLDER = "INSTANCEID"
 
 
 class HttpManagementPayload(dict[str, str]):
@@ -18,24 +22,52 @@ class HttpManagementPayload(dict[str, str]):
     via ``json.dumps(payload)``.
     """
 
-    def __init__(self, instance_id: str, instance_status_url: str, required_query_string_parameters: str):
+    def __init__(
+            self,
+            instance_id: str,
+            instance_status_url: str,
+            required_query_string_parameters: str,
+            *,
+            management_urls: Mapping[str, str] | None = None,
+            request_origin: str | None = None):
         """Initializes the HttpManagementPayload with the necessary URLs.
 
         Args:
             instance_id (str): The ID of the Durable Function instance.
             instance_status_url (str): The base URL for the instance status.
             required_query_string_parameters (str): The required URL parameters provided by the Durable extension.
+            management_urls (Mapping[str, str] | None): Canonical URL templates
+                provided by the Durable extension.
+            request_origin (str | None): Externally visible request origin used
+                to replace the templates' internal origin.
         """
-        super().__init__({
-            'id': instance_id,
+        fallback_urls = {
             'purgeHistoryDeleteUri': instance_status_url + "?" + required_query_string_parameters,
             'restartPostUri': instance_status_url + "/restart?" + required_query_string_parameters,
             'sendEventPostUri': instance_status_url + "/raiseEvent/{eventName}?" + required_query_string_parameters,
             'statusQueryGetUri': instance_status_url + "?" + required_query_string_parameters,
             'terminatePostUri': instance_status_url + "/terminate?reason={text}&" + required_query_string_parameters,
+            'rewindPostUri': instance_status_url + "/rewind?reason={text}&" + required_query_string_parameters,
             'resumePostUri': instance_status_url + "/resume?reason={text}&" + required_query_string_parameters,
-            'suspendPostUri': instance_status_url + "/suspend?reason={text}&" + required_query_string_parameters
-        })
+            'suspendPostUri': instance_status_url + "/suspend?reason={text}&" + required_query_string_parameters,
+        }
+        templates = management_urls or {}
+        placeholder = templates.get("id") or _INSTANCE_ID_PLACEHOLDER
+        encoded_instance_id = quote(instance_id, safe="")
+
+        urls = {'id': instance_id}
+        for name, fallback_url in fallback_urls.items():
+            template = templates.get(name)
+            if not template:
+                urls[name] = fallback_url
+                continue
+
+            url = template.replace(placeholder, encoded_instance_id)
+            if placeholder != _INSTANCE_ID_PLACEHOLDER:
+                url = url.replace(_INSTANCE_ID_PLACEHOLDER, encoded_instance_id)
+            urls[name] = _replace_origin(url, request_origin)
+
+        super().__init__(urls)
 
     def __str__(self) -> str:
         return json.dumps(self)
@@ -48,3 +80,21 @@ def urls(self) -> dict[str, Any]:
     def to_json(self) -> dict[str, Any]:
         """Return the management URLs as a plain ``dict``."""
         return dict(self)
+
+
+def _replace_origin(url: str, request_origin: str | None) -> str:
+    if request_origin is None:
+        return url
+
+    parsed_url = urlsplit(url)
+    parsed_origin = urlsplit(request_origin)
+    if not parsed_url.scheme or not parsed_url.netloc:
+        return url
+
+    return urlunsplit((
+        parsed_origin.scheme,
+        parsed_origin.netloc,
+        parsed_url.path,
+        parsed_url.query,
+        parsed_url.fragment,
+    ))

tests/azure-functions-durable/test_client_compat.py

@@ -29,11 +29,50 @@
     "managementUrls": {"id": "INSTANCEID"},
 })
 
+_MANAGEMENT_QUERY = "taskHub=HostHub&connection=HostStorage&code=host-key"
+_MANAGEMENT_URLS = {
+    "id": "INSTANCEID",
+    "statusQueryGetUri": (
+        "http://internal-host/custom/manage/INSTANCEID?"
+        f"{_MANAGEMENT_QUERY}"),
+    "sendEventPostUri": (
+        "http://internal-host/custom/manage/INSTANCEID/raiseEvent/{eventName}?"
+        f"{_MANAGEMENT_QUERY}"),
+    "terminatePostUri": (
+        "http://internal-host/custom/manage/INSTANCEID/terminate?reason={text}&"
+        f"{_MANAGEMENT_QUERY}"),
+    "rewindPostUri": (
+        "http://internal-host/custom/manage/INSTANCEID/rewind?reason={text}&"
+        f"{_MANAGEMENT_QUERY}"),
+    "purgeHistoryDeleteUri": (
+        "http://internal-host/custom/manage/INSTANCEID?"
+        f"{_MANAGEMENT_QUERY}"),
+    "restartPostUri": (
+        "http://internal-host/custom/manage/INSTANCEID/restart?"
+        f"{_MANAGEMENT_QUERY}"),
+    "suspendPostUri": (
+        "http://internal-host/custom/manage/INSTANCEID/suspend?reason={text}&"
+        f"{_MANAGEMENT_QUERY}"),
+    "resumePostUri": (
+        "http://internal-host/custom/manage/INSTANCEID/resume?reason={text}&"
+        f"{_MANAGEMENT_QUERY}"),
+}
+
 
 def _make_client() -> df.DurableFunctionsClient:
     return df.DurableFunctionsClient(_CLIENT_CONFIG)
 
 
+def _make_template_config() -> str:
+    return json.dumps({
+        "taskHubName": "TestHub",
+        "requiredQueryStringParameters": "code=fallback-key",
+        "baseUrl": "http://fallback/runtime/webhooks/durabletask",
+        "rpcBaseUrl": "http://localhost:8080/",
+        "managementUrls": _MANAGEMENT_URLS,
+    })
+
+
 def test_client_handles_null_max_grpc_message_size():
     # The Durable Functions host may send ``maxGrpcMessageSizeInBytes``
     # explicitly as ``null`` (not just omit it). ``dict.get(key, 0)`` returns
@@ -160,6 +199,85 @@ async def test_create_http_management_payload_requires_instance_id():
         await client.close()
 
 
+@pytest.mark.parametrize(
+    ("headers", "expected_origin"),
+    [
+        ({}, "http://request-internal:7071"),
+        ({"Forwarded": 'for=10.0.0.1;proto=https;host="public.example:8443"'},
+         "https://public.example:8443"),
+        ({"X-Forwarded-Proto": "https", "X-Forwarded-Host": "proxy.example"},
+         "https://proxy.example"),
+    ],
+)
+async def test_management_payload_uses_host_templates_and_external_origin(
+        headers, expected_origin):
+    config = _make_template_config()
+    async_client = df.DurableFunctionsClient(config)
+    sync_client = df.SyncDurableFunctionsClient(config)
+    request = func.HttpRequest(
+        method="POST",
+        url="http://request-internal:7071/api/start",
+        headers=headers,
+        body=b"")
+    instance_id = "folder/instance ?"
+    encoded_instance_id = "folder%2Finstance%20%3F"
+
+    try:
+        async_payload = async_client.create_http_management_payload(
+            request, instance_id)
+        sync_payload = sync_client.create_http_management_payload(
+            request, instance_id)
+
+        assert async_payload == sync_payload
+        assert async_payload["id"] == instance_id
+        assert async_payload["statusQueryGetUri"] == (
+            f"{expected_origin}/custom/manage/{encoded_instance_id}?"
+            f"{_MANAGEMENT_QUERY}")
+        assert async_payload["sendEventPostUri"] == (
+            f"{expected_origin}/custom/manage/{encoded_instance_id}/"
+            f"raiseEvent/{{eventName}}?{_MANAGEMENT_QUERY}")
+        assert async_payload["terminatePostUri"] == (
+            f"{expected_origin}/custom/manage/{encoded_instance_id}/"
+            f"terminate?reason={{text}}&{_MANAGEMENT_QUERY}")
+        assert async_payload["rewindPostUri"] == (
+            f"{expected_origin}/custom/manage/{encoded_instance_id}/"
+            f"rewind?reason={{text}}&{_MANAGEMENT_QUERY}")
+        assert async_payload["purgeHistoryDeleteUri"] == (
+            async_payload["statusQueryGetUri"])
+        assert async_payload["restartPostUri"] == (
+            f"{expected_origin}/custom/manage/{encoded_instance_id}/"
+            f"restart?{_MANAGEMENT_QUERY}")
+        assert async_payload["suspendPostUri"] == (
+            f"{expected_origin}/custom/manage/{encoded_instance_id}/"
+            f"suspend?reason={{text}}&{_MANAGEMENT_QUERY}")
+        assert async_payload["resumePostUri"] == (
+            f"{expected_origin}/custom/manage/{encoded_instance_id}/"
+            f"resume?reason={{text}}&{_MANAGEMENT_QUERY}")
+        assert async_payload.urls == async_payload.to_json()
+
+        async_response = async_client.create_check_status_response(
+            request, instance_id)
+        sync_response = sync_client.create_check_status_response(
+            request, instance_id)
+        assert json.loads(async_response.get_body()) == async_payload
+        assert json.loads(sync_response.get_body()) == sync_payload
+        assert json.loads(async_response.get_body())["rewindPostUri"] == (
+            async_payload["rewindPostUri"])
+    finally:
+        await async_client.close()
+        sync_client.close()
+
+
+async def test_management_payload_without_request_preserves_template_origin():
+    client = df.DurableFunctionsClient(_make_template_config())
+    try:
+        payload = client.create_http_management_payload("instance")
+        assert payload["statusQueryGetUri"] == (
+            f"http://internal-host/custom/manage/instance?{_MANAGEMENT_QUERY}")
+    finally:
+        await client.close()
+
+
 # ---------------------------------------------------------------------------
 # Deprecated client method aliases
 # ---------------------------------------------------------------------------
@@ -659,6 +777,8 @@ async def test_http_management_payload_is_mapping_like():
         payload = client.create_http_management_payload("inst1")
         assert payload["id"] == "inst1"
         assert "statusQueryGetUri" in payload
+        assert "rewindPostUri" in payload
+        assert payload.urls["rewindPostUri"] == payload.to_json()["rewindPostUri"]
         assert "id" in list(payload.keys())
         assert dict(payload.items())["id"] == "inst1"
     finally: