diff --git a/src/pydo/agents/custom_triggers.py b/src/pydo/agents/custom_triggers.py index 16168d22..d6a81001 100644 --- a/src/pydo/agents/custom_triggers.py +++ b/src/pydo/agents/custom_triggers.py @@ -157,16 +157,25 @@ def delete(self, trigger_id: str) -> None: """ self._send("DELETE", f"{_TRIGGERS_PATH}/{_quote(trigger_id)}") - def rotate_secret(self, trigger_id: str) -> Any: + def rotate_secret(self, trigger_id: str, *, revoke_previous: bool = False) -> Any: """Issue a new webhook secret (``POST .../{id}/rotate-secret``). - Webhook triggers only (``409`` for cron). The new secret is shown - once; the previous value stays valid briefly for in-flight deliveries. + Webhook triggers only (``409`` for cron). The new secret is shown once. + + By default the outgoing secret keeps verifying deliveries for a short + server-configured window, because the provider signs with the old value + until someone pastes the new one in, and ``previous_secret_expires_at`` + in the response says when it dies. Pass ``revoke_previous=True`` to + retire it on this call instead — intended for a compromised secret, + since deliveries still signed with the old value fail immediately, and + the response then carries ``previous_secret_revoked`` instead of an + expiry. Exactly one of the two is present. """ return self._parse_json( self._send( "POST", f"{_TRIGGERS_PATH}/{_quote(trigger_id)}/rotate-secret", + params={"revoke_previous": "true" if revoke_previous else None}, ), ) diff --git a/src/pydo/aio/agents/custom_triggers.py b/src/pydo/aio/agents/custom_triggers.py index dc1bb5ea..a24254e7 100644 --- a/src/pydo/aio/agents/custom_triggers.py +++ b/src/pydo/aio/agents/custom_triggers.py @@ -120,12 +120,27 @@ async def delete(self, trigger_id: str) -> None: """Soft-delete a trigger (``DELETE /v2/agents/triggers/{id}``).""" await self._send("DELETE", f"{_TRIGGERS_PATH}/{_quote(trigger_id)}") - async def rotate_secret(self, trigger_id: str) -> Any: - """Issue a new webhook secret (shown once).""" + async def rotate_secret( + self, trigger_id: str, *, revoke_previous: bool = False + ) -> Any: + """Issue a new webhook secret (``POST .../{id}/rotate-secret``). + + Webhook triggers only (``409`` for cron). The new secret is shown once. + + By default the outgoing secret keeps verifying deliveries for a short + server-configured window, because the provider signs with the old value + until someone pastes the new one in, and ``previous_secret_expires_at`` + in the response says when it dies. Pass ``revoke_previous=True`` to + retire it on this call instead — intended for a compromised secret, + since deliveries still signed with the old value fail immediately, and + the response then carries ``previous_secret_revoked`` instead of an + expiry. Exactly one of the two is present. + """ return await self._parse_json( await self._send( "POST", f"{_TRIGGERS_PATH}/{_quote(trigger_id)}/rotate-secret", + params={"revoke_previous": "true" if revoke_previous else None}, ), ) diff --git a/tests/agents/test_async_triggers.py b/tests/agents/test_async_triggers.py index a5e6e927..d9ba252f 100644 --- a/tests/agents/test_async_triggers.py +++ b/tests/agents/test_async_triggers.py @@ -59,6 +59,19 @@ def _make_async_resources(responses: List[_FakeAsyncResponse]) -> AsyncAgentsRes ) +def _find_call(resources: AsyncAgentsResources, path_suffix: str): + """Return the single recorded call whose URL path ends with path_suffix.""" + matches = [ + c + for c in resources._proxy._original._pipeline.calls + if c.request.url.split("?")[0].endswith(path_suffix) + ] + assert ( + len(matches) == 1 + ), f"expected exactly one {path_suffix} call, got {len(matches)}" + return matches[0] + + @pytest.mark.asyncio async def test_async_list_and_create_triggers(): resources = _make_async_resources( @@ -130,6 +143,11 @@ async def test_async_update_delete_rotate_and_executions(): rotated = await resources.triggers.rotate_secret("t1") assert rotated.webhook_secret == "new" + # Located by URL rather than by index: this test walks a fixed sequence of + # calls, and a positional assertion silently starts checking someone else's + # request the moment a call is inserted above. + rotate_call = _find_call(resources, "/rotate-secret") + assert "revoke_previous" not in rotate_call.request.url executions = await resources.triggers.list_executions("t1") assert executions.executions[0].execution_id == "e1" @@ -152,3 +170,21 @@ async def test_async_update_delete_rotate_and_executions(): assert resources._proxy._original._pipeline.calls[7].request.url.endswith( "/v2/agents/webhook-providers" ) + + +@pytest.mark.asyncio +async def test_async_rotate_secret_revoke_previous(): + resources = _make_async_resources( + [ + _FakeAsyncResponse( + 200, {"webhook_secret": "new", "previous_secret_revoked": True} + ) + ] + ) + + rotated = await resources.triggers.rotate_secret("t1", revoke_previous=True) + + call = resources._proxy._original._pipeline.calls[0] + assert call.request.method == "POST" + assert "revoke_previous=true" in call.request.url + assert rotated.previous_secret_revoked is True diff --git a/tests/agents/test_triggers.py b/tests/agents/test_triggers.py index 182658f9..73bd8748 100644 --- a/tests/agents/test_triggers.py +++ b/tests/agents/test_triggers.py @@ -197,7 +197,10 @@ def test_delete_trigger_returns_none_on_204(): def test_rotate_secret(): - body = {"webhook_secret": "whsec_rotated"} + body = { + "webhook_secret": "whsec_rotated", + "previous_secret_expires_at": "2026-07-01T12:05:00Z", + } resources = _make_resources([_FakeResponse(200, body)]) resp = resources.triggers.rotate_secret("t1") @@ -205,7 +208,22 @@ def test_rotate_secret(): call = _last_call(resources) assert call.request.method == "POST" assert call.request.url.endswith("/v2/agents/triggers/t1/rotate-secret") + assert "revoke_previous" not in call.request.url assert resp.webhook_secret == "whsec_rotated" + assert resp.previous_secret_expires_at == "2026-07-01T12:05:00Z" + + +def test_rotate_secret_revoke_previous(): + body = {"webhook_secret": "whsec_rotated", "previous_secret_revoked": True} + resources = _make_resources([_FakeResponse(200, body)]) + + resp = resources.triggers.rotate_secret("t1", revoke_previous=True) + + call = _last_call(resources) + assert call.request.method == "POST" + assert _path(call.request.url).endswith("/v2/agents/triggers/t1/rotate-secret") + assert "revoke_previous=true" in call.request.url + assert resp.previous_secret_revoked is True # ---------------------------------------------------------------------------